Switch to SingleSubscriptionTransformer<S>

[natebosch]

Currently its SingleSubscriptionTransformer<S,T>, but the purpose is not to allow changing the generic on the stream type. See TODO comment on in the file.

[lrhn]

Not clear what should be changed. Can you elaborate?

[natebosch]

Sorry, I mistyped the name.

This is a tracking issue for:

https://github.com/dart-lang/async/blob/f708371eff958ed0e12cfba3958001375522f581/lib/src/single_subscription_transformer.dart#L25-L26

This was never intended to be used in any way other than the case where both generics are the same.

Concretely, this class should be defined as:

/// A transformer that converts a broadcast stream into a single-subscription
/// stream.
///
/// This buffers the broadcast stream's events, which means that it starts
/// listening to a stream as soon as it's bound.
class SingleSubscriptionTransformer<S> extends StreamTransformerBase<S, S> {
  const SingleSubscriptionTransformer();

  @override
  Stream<S> bind(Stream<S> stream) {
    StreamSubscription<S> subscription;
    var controller =
        StreamController<S>(sync: true, onCancel: () => subscription.cancel());
    subscription = stream.listen(controller.add,
        onError: controller.addError, onDone: controller.close);
    return controller.stream;
  }
}

[lrhn]

Ack.

FWIW, I'd implement it as:

  @override
  Stream<S> bind(Stream<S> stream) {
    var controller = StreamController<S>(sync: true);
    controller.addStream(stream).whenComplete(controller.close);
    return controller.stream;
  }

(I also think it's a bad idea to listen immediately and buffer an unboundend number of events, but if that's the desired semantics, I guess it has to happen).

Let's make the Null Safe release a new major version, that should let us fix things like this.

[natebosch]

Let's make the Null Safe release a new major version, that should let us fix things like this.

We really hope to be able to do that 😄