Поведение Subject StreamTransformer

Question

Я ищу способ превратить обычный поток в BehaviorSubject.

Это код, который я сделал, но я не знаю, как указать, что мой класс возвращает BehaviorSubject. И мой преобразователь выдает ошибку времени выполнения (слушание вызвано на ноль).

import 'dart:async';

import 'package:rxdart/subjects.dart';

class BehaviorSubjectTransformer<S, T> extends StreamTransformerBase<S, T> {

  StreamController _controller;

  StreamSubscription _subscription;

  bool cancelOnError;

  // Original Stream
  Stream<S> _stream;

  BehaviorSubjectTransformer({bool sync: false, this.cancelOnError}) {
    _controller = new BehaviorSubject(onListen: _onListen, onCancel: _onCancel, sync: sync);
  }

  BehaviorSubjectTransformer.broadcast({bool sync: false, bool this.cancelOnError}) {
    _controller = new BehaviorSubject(onListen: _onListen, onCancel: _onCancel, sync: sync);
  }

  void _onListen() {
    _subscription = _stream.listen(onData,
        onError: _controller.addError,
        onDone: _controller.close,
        cancelOnError: cancelOnError);
  }

  void _onCancel() {
    _subscription.cancel();
    _subscription = null;
  }

  void onData(S data) {
    _controller.add(data);
  }

  Stream<T> bind(Stream<S> stream) {
    this._stream = stream;
    return _controller.stream;
  }

}