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:

async/lib/src/single_subscription_transformer.dart

Lines 25 to 26 in f708371

	// TODO(nweiz): When we release a new major version, get rid of the second
	// type parameter and avoid this conversion.

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 😄