DelegatingStream.asBroadcastStream() calls wrapped stream directly

[TastyPi]

• Dart SDK Version (dart --version) 2.2.1-dev.2.0+google3-v2.2.1.dev.2.0

Consider the following test:

class TestStream<T> extends DelegatingStream<T> {
  bool listenCalled = false;

  TestStream(Stream<T> stream) : super(stream);

  @override
  StreamSubscription<T> listen(void onData(T event),
      {Function onError, void onDone(), bool cancelOnError}) {
    listenCalled = true;
    return super.listen(onData,
        onError: onError, onDone: onDone, cancelOnError: cancelOnError);
  }
}

main() {
  test('asBroadcastStream should subscribe to TestStream', () {
    final stream = TestStream(Stream.fromIterable([1, 2, 3]);
    stream.asBroadcastStream().listen((_) {});
    expect(stream.listenCalled, isTrue);
  });
}

This fails with:

  Expected: true
    Actual: false

The problem is StreamView which DelegatingStream extends from overrides asBroadcastStream to directly use the wrapped stream, not itself. DelegatingStream should not extend StreamView, it should just override listen and isBroadcast.

[lrhn]

The problem here is that asBroadcastStream() is indeed using the asBroadcastStream of the wrapped stream, which uses the listen of the wrapped stream.

I agree that the DelegatingStream should not depend on StreamView. The two classes have different purposes. The StreamView is there to hide methods that are not on Stream (for classes that implement Stream and something more, which is bad idea). The DelegatingStream intends to work like the wrapped stream, but allows you to override specific methods. For a Stream, the only method that really matters is listen (and the isBroadcastStream getter to some extent). Everything else is defined in terms of that, in a very consistent way.

So, if some class overrides asBroadcastStream, what is the correct behavior?

• Should DelegatingStream use the overridden asBroadcastStream?
• Should DelegatingStream ignore the overridden asBroadcastStream and use the default behavior?

Currently we are doing the former. That bites us here because the overridden asBroadcastStream has no way of seeing the DelegatingStream wrapper. On the other hand, it's a weird "delegating" stream wrapper that doesn't actually use the stream it wraps. On the third hand (I wish!), that's what we do for all the other methods, they just use the default implementation based on listen, even though the wrapped stream might override last more efficiently.

No clear and good solution here. There is a consistency argument for not forwarding asBroadcastStream to the underlying stream, and a practical argument that nobody overrides asBroadcastStream in practice (and those who do, they do it so badly that they are better ignored).

Changing the behavior is theoretically a breaking change - if someone wraps a stream which overrides asBroadcastStream and depends on the overridden method to be called, changing it will break that code.

All in all, I'm not sure the risk of breakage, no matter how hypothetical, is worth it.

I think the best solution would be for DelegatingStream to stop extending StreamView, and just do the listen/isBroadcastStream forwarding itself.

[TastyPi]

I'm not talking about overriding asBroadcastStream, I just want to override listen, however I would expect asBroadcastStream to subscribe to the DelegatingStream subclass, not directly to the wrapped stream. I.e., I would expect this implementation:

class DelegatingStream<T> extends Stream<T> {
  @override
  bool isBroadcastStream => _delegate.isBroadcastStream;

  @override
  StreamSubscription<T> listen(...) =>
      _delegate.listen(...);

  final Stream<T> _delegate;

  DelegatingStream(this._delegate);
}

I realise it is technically a breaking change, but I think it is rare for developers to assume DelegatingStream extends StreamView, so changing that shouldn't be an issue. It's a weird thing for the class to be doing anyway. I think it is more likely that this problem hasn't been noticed before because asBroadcastStream is not used a lot, which really means every DelegatingStream is a ticking time-bomb just waiting for someone to call asBroadcastStream.

[lrhn]

I can understand that expectation, but it is not what other DelegatingX classes do. Instead they delegate every function to the wrapped object.

If DelegatingStream did the same thing, then overriding listen would not change anything except direct calls to listen, all other methods would forward to the wrapped stream's methods which uses the wrapped stream's listen.

So, should DelegatingStream be consistent with other DelegatingX classes, or should it do what you expect? That depends entirely on the use-case.

With the "only forward listen" approach, you can't meaningfully override any other method. If you do, it might not be called anyway, because the receiver might also wrap your stream and only forward the listen.

In practice we might two different classes, one that only forwards Listen and one that forwards everything.

Or, since only forwarding listen means that there is only one function to care about, we could introduce a constructor like Stream(StreamSubscription<T> onListen(), {bool isBroadcast = false}) => _OnListenStream<T>(onListen, isBroadcast);, with a supporting class of:

class _OnListenStream<T> extends Stream<T> {
  StreamSubscription<T> Function(bool cancelOnError) _onListen;
  final bool isBroadcast;
  _OnListenStream(this._onListen, this.isBroadcast);
  StreamSubscription<T> listen(
      void onData(T data), { Function onError, void onDone(), bool cancelOnError = false }) {
    var sub = _onListen(cancelOnError);
    if (onData != null) sub.onData(onData);
    if (onError != null) sub.onError(onError);      
    if (onDone != null) sub.onDone(onDone);
    return sub;
  }
}

(Not saying that we are in any way consistent now, forwarding asBroadcastStream and nothing else, but I do think that the change should be made to DelegatingStream, not StreamView).

[TastyPi]

I see, that makes sense. My assumption would be that most people who want to wrap a stream would want all the methods to go via the overridden listen, but your point of other classes named Delegating forwarding all calls makes sense too.

It's pretty straightforward to just extend Stream and override the two methods myself, so this isn't a big issue for me, but I think we're in agreement that DelegatingStream is not implemented correctly, whatever "correct" means in this case.

[kevmoo]

@lrhn – is there an SDK issue here?

[lrhn]

I don't think we will change the SDK classes. The DelegatingStream from package:async may want to change to either forward all members, or not forward asBroadcastStream (or having both versions available).

[kevmoo]

Transferred the issue