Fixed a scenario of downstream interruptions being dropped.

[andersio]

Scope

The issue apparently exists since 2.0 with producer lifting optimisation (#140), and is also applicable to TransformerCore.

Symptom

With a terminated upstream, manual interruption of time shifted producers, including delay, observe(on:), throttle, debounce and lazyMap, would be ignored.

This is due to the operator lifting scheme introduced in #140, which blatantly assumes the upstream always lives longer than the downstream & therefore is always available to handle the downstream/user interruption.

Simple example:

// Completed upstream + `delay`.
SignalProducer.empty
    .delay(10.0, on: QueueScheduler.main)
    .startWithCompleted { print("Value should have been discarded!") }
    .dispose()

// Console(t+10): Value should have been discarded!

Proposed Fix

Transformation Core

The first part of the fix establishes a CompositeDisposable specifically for the event transformation (the combined version if nested), which is disposed of only if the transformation outputs a terminal event.

(Pretty much like operator lifting in 1.0 and earlier, but one CompositeDisposable being shared by all operators).

This allows asynchronous event transformation to response to user/downstream interruption properly.

Operator Lifting for SignalProducer

The second part of the fix reverts the implementation of SignalProducer.lift to pre-#140 times.

Operator lifting has a significantly reduced role since 3.0. With a large set of operators already being lowered as event transforms, I believe the impact would be minimal.

Note on async operators & interrupted delivery

The fix still maintains the semantic of interrupted respecting async operators, introduced in 2.0 (#140).

Summary: When the product of async operators like observe(on:) gets interrupted, unless otherwise specified, the interrupted event should be delivered on the respective scheduler.

lazyMap does not follow this yet, and we should probably reimplement it as event transform at a later point.

timeout is an odd one with only the failure delivered asynchronously, so it isn't bound by the said semantic.

Acknowledgement

Special thanks to Takeru Chuganji (@chuganzy) raising this on Slack.

Checklist

• Updated CHANGELOG.md.

[mdiep]

This should also be added to the changelog.

[mdiep]

This init doesn't have a disposables parameter. The line also seems incomplete.

[andersio]

@mdiep I've inlined the now diverged code paths in Signal.Observer into Signal.flatMapEvent and TransformerCore alongside the (refined) documentation. Hopefully it would make it clearer. I've also updated the changelog.