Dart What does "StreamTransformer<RS, RT> cast<RS, RT>()" do?

Question

I've implemented the stream transformer. Please note that it is only an exercise (in order to learn Dart). This transformer converts integers into strings. I give the code below, and you can also find it on GitHub.

// Conceptually, a transformer is simply a function from Stream to Stream that
// is encapsulated into a class.
//
// A transformer is made of:
// - A stream controller. The controller provides the "output" stream that will
//   receive the transformed values.
// - A "bind()" method. This method is called by the "input" stream "transform"
//   method (inputStream.transform(<the stream transformer>).

import 'dart:async';

/// This class defines the implementation of a class that emulates a function
/// that converts a data with a given type (S) into a data with another type (T).
abstract class TypeCaster<S, T> {
  T call(S value);
}

/// This class emulates a converter from integers to strings.
class Caster extends TypeCaster<int, String> {
  String call(int value) {
    return "<${value.toString()}>";
  }
}

// StreamTransformer<S, T> is an abstract class. The functions listed below must
// be implemented:
// - Stream<T> bind(Stream<S> stream)
// - StreamTransformer<RS, RT> cast<RS, RT>()

class CasterTransformer<S, T> implements StreamTransformer<S, T> {

  StreamController<T> _controller;
  bool _cancelOnError;
  TypeCaster<S, T> _caster;

  // Original (or input) stream.
  Stream<S> _stream;

  // The stream subscription returned by the call to the function "listen", of
  // the original (input) stream (_stream.listen(...)).
  StreamSubscription<S> _subscription;

  /// Constructor that creates a unicast stream.
  /// [caster] An instance of "type caster".
  CasterTransformer(TypeCaster<S, T> caster, {
    bool sync: false,
    bool cancelOnError: true
  }) {
    _controller = new StreamController<T>(
        onListen: _onListen,
        onCancel: _onCancel,
        onPause: () => _subscription.pause(),
        onResume: () => _subscription.resume(),
        sync: sync
    );
    _cancelOnError = cancelOnError;
    _caster = caster;
  }

  /// Constructor that creates a broadcast stream.
  /// [caster] An instance of "type caster".
  CasterTransformer.broadcast(TypeCaster<S, T> caster, {
    bool sync: false,
    bool cancelOnError: true
  }) {
      _cancelOnError = cancelOnError;
      _controller = new StreamController<T>.broadcast(
          onListen: _onListen,
          onCancel: _onCancel,
          sync: sync
      );
      _caster = caster;
  }

  /// Handler executed whenever a listener subscribes to the controller's stream.
  /// Note: when the transformer is applied to the original stream, through call
  ///       to the method "transform", the method "bind()" is called behind the
  ///       scenes. The method "bind()" returns the controller stream.
  ///       When a listener is applied to the controller stream, then this function
  ///       (that is "_onListen()") will be executed. This function will set the
  ///       handler ("_onData") that will be executed each time a value appears
  ///       in the original stream. This handler takes the incoming value, casts
  ///       it, and inject it to the (controller) output stream.
  /// Note: this method is called only once. On the other hand, the method "_onData"
  ///       is called as many times as there are values to transform.
  void _onListen() {
    _subscription = _stream.listen(
        _onData,
        onError: _controller.addError,
        onDone: _controller.close,
        cancelOnError: _cancelOnError
    );
  }

  /// Handler executed whenever the subscription to the controller's stream is cancelled.
  void _onCancel() {
    _subscription.cancel();
    _subscription = null;
  }

  /// Handler executed whenever data comes from the original (input) stream.
  /// Please note that the transformation takes place here.
  /// Note: this method is called as many times as there are values to transform.
  void _onData(S data) {
    _controller.add(_caster(data));
  }

  /// This method is called once, when the stream transformer is assigned to the
  /// original (input) stream. It returns the stream provided by the controller.
  /// Note: here, you can see that the process transforms a value of type
  ///       S into a value of type T. Thus, it is necessary to provide a function
  ///       that performs the conversion from type S to type T.
  /// Note: the returned stream may accept only one, or more than one, listener.
  ///       This depends on the method called to instantiate the transformer.
  ///       * CasterTransformer() => only one listener.
  ///       * CasterTransformer.broadcast() => one or more listener.
  Stream<T> bind(Stream<S> stream) {
    _stream = stream;
    return _controller.stream;
  }

  // TODO: what should this method do ? Find the answer.
  StreamTransformer<RS, RT> cast<RS, RT>() {
    return StreamTransformer<RS, RT>((Stream<RS> stream, bool b) {
      // What should we do here ?
    });
  }
}


main() {

  // ---------------------------------------------------------------------------
  // TEST: unicast controller.
  // ---------------------------------------------------------------------------

  // Create a controller that will be used to inject integers into the "input"
  // stream.
  StreamController<int> controller_unicast = new StreamController<int>();
  // Get the stream "to control".
  Stream<int> integer_stream_unicast = controller_unicast.stream;
  // Apply a transformer on the "input" stream.
  // The method "transform" calls the method "bind", which returns the stream that
  // receives the transformed values.
  Stream<String> string_stream_unicast = integer_stream_unicast.transform(CasterTransformer<int, String>(new Caster()));

  string_stream_unicast.listen((data) {
    print('String => $data');
  });

  // Inject integers into the "input" stream.
  controller_unicast.add(1);
  controller_unicast.add(2);
  controller_unicast.add(3);

  // ---------------------------------------------------------------------------
  // TEST: broadcast controller.
  // ---------------------------------------------------------------------------

  StreamController<int> controller_broadcast = new StreamController<int>.broadcast();
  Stream<int> integer_stream_broadcast = controller_broadcast.stream;
  Stream<String> string_stream_broadcast = integer_stream_broadcast.transform(CasterTransformer<int, String>.broadcast(new Caster()));

  string_stream_broadcast.listen((data) {
    print('Listener 1: String => $data');
  });

  string_stream_broadcast.listen((data) {
    print('Listener 2: String => $data');
  });

  controller_broadcast.add(1);
  controller_broadcast.add(2);
  controller_broadcast.add(3);
}

The class CasterTransformer<S, T> extends the abstract class StreamTransformer<S, T>.

Thus, it implements the method StreamTransformer<RS, RT> cast<RS, RT>().

On the documentation, it is said that :

The resulting transformer will check at run-time that all data events of the stream it transforms are actually instances of S, and it will check that all data events produced by this transformer are actually instances of RT.

See: https://api.dartlang.org/stable/2.1.0/dart-async/StreamTransformer/cast.html

First, I think that there is a typo in this documentation : it should say "...it transforms are actually instances of RS" (instead of S).

However, this seems obscure to me.

• Why do we need a stream transformer to check values types ? The purpose of a transformer is to transform, isn't it ? If the purpose of a component is to check, so why don't we call it a checker ?
• And, also, why would we need to check that the transformer (we implement) produces the required data ? If it doesn't, then we face a bug that should be fixed.

Can someone explain the purpose of the method Cast() ?

Answer 1

The cast method is there to help typing the operation.

If you have a StreamTransformer<num, int>, it transforms numbers to integers (say, by calling .toInt() on them and then adding 42, because that is obviously useful!). If you want to use that transformer in some place that expects a StreamTransformer<int, num>, then you can't. Since num is not a sub-type of int, the transformer is not assignable to that type.

But you know, because you understand how a stream transformer actually works, that the first type argument is only used for inputs. Something that accepts any num should safely be useable where it's only given ints. So, to convince the type system that you know what you are doing, you write:

StreamTransformer<int, num> transform = myTranformer.cast<int, num>();

Now, the tranformer takes any integer (RS), checks that it's a num (S), passes it to myTransformer which calls toInt() and adds 42, then the resulting int (T) is passed back and transformer checks that it is a num (RT) and emits that.

Everything works and the type system is happy.

You can use cast to do things that will never work at run-time, because all it does is to add extra run-time checks that convinces the static type system that things will either succeed or throw at those checks.

The easiest way to get an implementation of StreamTransformer.cast is to use th e StreamTransformer.castFrom static method:

StreamTransformer<RS, RT> cast<RS, RT>() => StreamTransformer.castFrom(this);

That will use the system's default cast wrapper on your own transformer.