如何在Dart中创建StreamTransformer

尝试构建一个自定义的StreamTransformer类，但是那里的许多示例似乎已经过时了，并且文档中找到的示例不是（某些类型语言无论如何可能会考虑的）作为一个类（在这里找到：https://api.dartlang.org/apidocs/channels/stable/dartdoc-viewer/dart:async.StreamTransformer）。这似乎不是一种非常像Dart的方法，而更像是一种类似Javascript的方式（我使用Dart来避免）。

许多在线资源说这就是您创建StreamTransformer的方式，但是在扩展它时会出现错误。

class exampleStreamTransformer extends StreamTransformer
{
  //... (This won't work)
}

"实现"似乎是要走的路，以及实现所需的绑定函数：

class exampleStreamTransformer implements StreamTransformer
{
  Stream bind(Stream stream)
  {
    //... (Go on to return new stream, etc)
  }
}

我似乎找不到任何这种方式的例子，但自己把一些东西放在一起（在我的 IDE 中被接受，但在运行时不被接受，当它尝试使用 pause getter 时，我收到一个空对象错误）：

class exampleStreamTransformer implements StreamTransformer
{
  StreamController<String> _controller;
  StreamSubscription<String> _subscription;
  Stream bind(Stream stream)
  {
    _controller = new StreamController<String>(
        onListen: ()
        {
          _subscription = stream.listen((data)
          {
            // Transform the data.
            _controller.add(data);
          },
          onError: _controller.addError,
          onDone: _controller.close,
          cancelOnError: true); // Unsure how I'd pass this in?????
        },
        onPause: _subscription.pause,
        onResume: _subscription.resume,
        onCancel: _subscription.cancel,
        sync: true
    );
    return _controller.stream;
  }
}

希望以这种方式实现它，就像以"类型化"的方式制作类一样，非常感谢任何帮助，谢谢。

为什么不使用StreamTransformer.fromHandler()：

import 'dart:async';
void handleData(data, EventSink sink) {
  sink.add(data*2);
}
void main() {
  StreamTransformer doubleTransformer = new StreamTransformer.fromHandlers(handleData: handleData);
  StreamController controller = new StreamController();
  controller.stream.transform(doubleTransformer).listen((data) {
    print('data: $data');
  });
  controller.add(1);
  controller.add(2);
  controller.add(3);
}

输出：

data: 2
data: 4
data: 6

好的。这是另一个工作示例：

import 'dart:async';
class DuplicateTransformer<S, T> implements StreamTransformer<S, T> {
  StreamController _controller;
  StreamSubscription _subscription;
  bool cancelOnError;
  // Original Stream
  Stream<S> _stream;
  DuplicateTransformer({bool sync: false, this.cancelOnError}) {
    _controller = new StreamController<T>(onListen: _onListen, onCancel: _onCancel, onPause: () {
      _subscription.pause();
    }, onResume: () {
      _subscription.resume();
    }, sync: sync);
  }
  DuplicateTransformer.broadcast({bool sync: false, bool this.cancelOnError}) {
    _controller = new StreamController<T>.broadcast(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;
  }
  /**
   * Transformation
   */
  void onData(S data) {
    _controller.add(data);
    _controller.add(data); /* DUPLICATE EXAMPLE!! REMOVE FOR YOUR OWN IMPLEMENTATION!! */
  }
  /**
   * Bind
   */
  Stream<T> bind(Stream<S> stream) {
    this._stream = stream;
    return _controller.stream;
  }
}
void main() {
  // Create StreamController
  StreamController controller = new StreamController.broadcast();
  // Transform
  Stream s = controller.stream.transform(new DuplicateTransformer.broadcast());
  s.listen((data) {
    print('data: $data');
  }).cancel();
  s.listen((data) {
    print('data2: $data');
  }).cancel();
  s.listen((data) {
    print('data3: $data');
  });
  // Simulate data
  controller.add(1);
  controller.add(2);
  controller.add(3);
}

让我补充一些注意事项：

在查看其他 dart 内部变压器的源代码时，使用 implements 似乎是正确的方法。
我为常规流和广播流实现了两个版本。
如果是常规流，您可以直接在新的流控制器上调用取消/暂停/恢复，因为我们只能侦听一次。
如果您使用广播流，我发现只有在没有人已经收听该流时才调用 listen（）。 onCancel 的行为相同。如果最后一个订阅者取消其订阅，则调用 onCancel。这就是为什么在这里使用相同的功能是安全的。

与map不同，变压器功能更强大，允许您维护内部状态，并随时发出值。它可以实现map做不到的事情，例如延迟、复制值、选择性省略某些值等。

实质上，实现需要一个bind方法，该方法基于传入的旧流提供新流，以及一个cast方法，该方法有助于在运行时进行类型检查。

下面是实现"TallyTransformer"的过度简化示例，该转换器将整数值流转换为总和流。例如，如果到目前为止的输入流有1, 1, 1, -2, 0, ...，输出流将被1, 2, 3, 1, 1, ...，即对到目前为止的所有输入求和。

示例用法：stream.transform(TallyTransformer())

class TallyTransformer implements StreamTransformer {
  StreamController _controller = StreamController();
  int _sum = 0; // sum of all values so far
  @override
  Stream bind(Stream stream) {
    // start listening on input stream
    stream.listen((value) {
      _sum += value; // add the new value to sum
      _controller.add(_sum); // emit current sum to our listener
    });
    // return an output stream for our listener
    return _controller.stream;
  }
  @override
  StreamTransformer<RS, RT> cast<RS, RT>() {
    return StreamTransformer.castFrom(this);
  }
}

此示例过于简化（但仍然有效），并且不包括流暂停、恢复或取消等情况。如果遇到"Stream has already been listened"错误，请确保流正在广播。

https://github.com/dart-lang/sdk/issues/27740#issuecomment-258073139

您可以使用 StreamTransformer.fromHandlers 轻松创建仅将输入事件转换为输出事件的转换器。

例：

new StreamTransformer.fromHandlers(handleData: (String event, EventSink output) {
  if (event.startsWith('data:')) {
    output.add(JSON.decode(event.substring('data:'.length)));
  } else if (event.isNotEmpty) {
    output.addError('Unexpected data from CloudBit stream: "$event"');
  }
});

如果你想简单地使用这样的函数转换值

int handleData(int data) {
  return data * 2;
}

使用流的映射方法

stream
  .map(handleData)
  .listen((data) {
    print('data: $data');
  });

完整示例：

import 'dart:async';
int handleData(int data) {
  return data * 2;
}
void main() {
  final controller = StreamController<int>();
  controller.stream
    .map(handleData)
    .listen((data) {
      print('data: $data');
    });
  controller.add(1);
  controller.add(2);
  controller.add(3);
}

查看有关 dart.dev 的更多示例

相关内容

最新更新

热门标签：