Stream<Y implements X> cannot be consumed by StreamConsumer<X>

[thosakwe]

• Dart VM version: 2.3.2-dev.0.1 (Tue Jun 4 10:56:51 2019 +0200) on "macos_x64"

It seems that in newer versions of Dart, if I have a Stream<Uint8List>, I cannot use it with a StreamConsumer<List<int>>. This worked in the past, and in my head makes sense - because Uint8List is a descendant of List<int>, anything that accepts a List<int> should accept a Uint8List.

I've included two examples here:
https://github.com/thosakwe/uint8_repro/tree/master/bin

  error • The argument type 'AConsumer' can't be assigned to the parameter type 'StreamConsumer<B>' at bin/general_case.dart:16:21 • argument_type_not_assignable
  error • The argument type 'IntListConsumer' can't be assigned to the parameter type 'StreamConsumer<Uint8List>' at bin/uint8_list.dart:22:29 • argument_type_not_assignable

Both cases result in an argument_type_not_assignable error.

---

• Why did this work before? A downcast from Uint8List to List<int> was already illegal in the analyzer before.
• When were these semantics changed? I didn't see it in the changelog, and this worked as recently as Dart 2.2.0.
• Stream.pipe is not very useful without support for polymorphism. I can also imagine that this change could break code for a lot of Dart users. For example, I ran into this error while reading files from package:file (using File.openRead), and piping the resulting stream to a StreamConsumer<List<int>>. In [email protected], Uint8List is now returned instead of List<int>, and so anyone depending on that package (20 other packages on Pub) might already be running into this.

Thanks in advance.

[eernstg]

Consider the general case:

import 'dart:async';

class A {}

class B implements A {}

class AConsumer implements StreamConsumer<A> {
  @override
  Future close() => null;

  @override Future<void> addStream(Stream<A> stream) async => null;
}

main() async {
  var stream = Stream.fromIterable([B()]);
  await stream.pipe(AConsumer());
}

The reason why this fails is that stream has type Stream<B>, so stream.pipe accepts an argument of type StreamConsumer<B>, and there is no subtype relationship between this type and the given type AConsumer. (We do have AConsumer <: StreamConsumer<A> and StreamConsumer<B> <: StreamConsumer<A>, but that's an up-then-down-cast and that doesn't amount to an upcast nor a downcast.)

You can avoid this conflict if you can make sure that stream has type Stream<A>:

main() async {
  var stream = Stream<A>.fromIterable([B()]);
  await stream.pipe(AConsumer());
  
  // or:
  var stream2 = Stream.fromIterable([B()]);
  await stream2.cast<A>().pipe(AConsumer());
}

However, there is an underlying reason for this conflict which is a bit deeper: The class StreamConsumer<S> uses its type argument just once, and that's in a method parameter type Stream<S>. This means that it could be declared in languages like C#, Kotlin, and Scala with a contravariant type argument (with different syntax, e.g., C# uses in as a type parameter modifier, and Scala uses -), but in Dart there is currently no support for declaring contravariant type arguments.

If it had been possible to declare the type argument of StreamConsumer as contravariant then we would have had AConsumer <: StreamConsumer<A> <: StreamConsumer<B>.

But given that we can't do that we can use the special case of invariance: Make sure that the same type argument is used everywhere (statically and dynamically).

In general, it fits much better for Dart to use a function type (where parameter types are actually contravariant) than a class type, in this kind of situation where there is a type parameter that should have been contravariant, but that's of course not an option when using a class like StreamConsumer which is required by the context.

[eernstg]

Again, the situation doesn't allow for making changes to StreamConsumer, but just to illustrate the idea here's the core structure of the example, adjusted to use a function type such that we get the desired variance:

class A {}
class B implements A {}

class Stream<T> {
  Stream.fromIterable(Iterable<T> elements);
  Future pipe(StreamConsumer<Future Function(Stream<T>)> streamConsumer) =>
      null;
}

typedef F<X> = Future Function(Stream<X>);

abstract class StreamConsumer<F0 extends F<Null>> {
  F0 get addStream;
  Future close();
}

class AConsumer implements StreamConsumer<F<A>> {
  F<A> get addStream => null;
  Future close() => null;
}

main() async {
  var stream = Stream.fromIterable([B()]);
  await stream.pipe(AConsumer());
}

Now we have no problem passing an AConsumer to Stream<B>.pipe, because AConsumer <: StreamConsumer<F<A>> <: StreamConsumer<F<B>>.

In any case, the tools are working as specified, and the fact that we cannot directly express contravariance for a type parameter is known. So I'll close this issue with 'working as intended'.

[thosakwe]

Thanks for the response! Seems like I'll have some reading to do...

Appreciate it.

[eernstg]

Thanks!