Thanks for your help and that's the right API for sure.
But I think it still doesn't send .finish immediately after the predicate fails. So in the above example if "garbage" doesn't arrive for 1 hour after "end-transmission" then the .finish will also not be sent until 1 hour after "end-transmission" ... I "think" that's what I'm seeing in debugger.
Do I need a custom publisher for this behavior?
Jesse