Description
Serialised operators rely on the producer to to close the channel and propagate a stop signal.
If the producer does not do a close, the serialised operator will not notify that it has stopped.
Augment the serializedSend() method to flag that the operator has stopped and do an early sendStop.
Augment close method to only send a stop if processItem() did not fail.