Expose Sink.cancelSubscription()
Source now supports un- and re-subscribing

Although some elements may be lost when unsubscribing, if the Source
already emitted them to a buffer or queue which the previous Sink hadn't
processed yet, so be careful.