Flutter数据流通信插件streams_channel的使用

StreamsChannel 是受 EventChannel 启发而创建的，它允许在 Flutter 和平台端之间创建多条数据流。

背景

EventChannel 允许为每个通道创建一个单一的开放流。首次订阅时，流会被打开，并且可以从 Flutter 端向平台端传递参数。后续的订阅要么复用现有的开放流，要么用新的流覆盖旧的流。

为了同时支持多个流的开放，并且具有不同的初始化参数，需要为每个流创建多个 EventChannel。如果流的数量是动态的，这种方式会变得复杂。StreamsChannel 提供了对多个流的支持，解决了上述问题。

安装

在 pubspec.yaml 文件中添加 streams_channel：

dependencies:
  streams_channel: ^0.2.3

然后运行 flutter packages get 来安装依赖。

使用方法

StreamsChannel 的工作方式与 EventChannel 类似，但支持同时打开多个流。

平台端实现

在 Android 和 iOS 平台端，都需要设置一个流处理器工厂。

Android 示例代码：

public class DemoPlugin {
  public static void registerWith(PluginRegistry.Registrar registrar) {
    final StreamsChannel channel = new StreamsChannel(registrar.messenger(), "streams_channel_test");
    channel.setStreamHandlerFactory(new StreamsChannel.StreamHandlerFactory() {
      @Override
      public EventChannel.StreamHandler create(Object arguments) {
        return new StreamHandler();
      }
    });
  }

  // 发送 "Hello" 10次，每秒发送一次，最后结束流
  public static class StreamHandler implements EventChannel.StreamHandler {

    private final Handler handler = new Handler();
    private final Runnable runnable = new Runnable() {
      @Override
      public void run() {
        if (count > 10) {
            eventSink.endOfStream();
        } else {
            eventSink.success("Hello " + count + "/10");
        }
        count++;
        handler.postDelayed(this, 1000);
      }
    };

    private EventChannel.EventSink eventSink;
    private int count = 1;

    @Override
    public void onListen(Object o, final EventChannel.EventSink eventSink) {
      this.eventSink = eventSink;
      runnable.run();
    }

    @Override
    public void onCancel(Object o) {
      handler.removeCallbacks(runnable);
    }
  }
}

iOS 示例代码：

@interface StreamHandler : NSObject<FlutterStreamHandler>
@property(strong, nonatomic) NSTimer *timer;
@property(assign, nonatomic) NSInteger count;
@end

@implementation DemoPlugin

+ (void)registerWithRegistrar:(NSObject<FlutterPluginRegistrar>*)registrar {

  FlutterStreamsChannel *channel = [FlutterStreamsChannel streamsChannelWithName:@"streams_channel_test" binaryMessenger:registrar.messenger];
  [channel setStreamHandlerFactory:^NSObject<FlutterStreamHandler> *(id arguments) {
    return [StreamHandler new];
  }];
}

@end

// 发送 "Hello" 10次，每秒发送一次，最后结束流
@implementation StreamHandler

- (FlutterError *)onListenWithArguments:(id)arguments eventSink:(FlutterEventSink)events {
  self.count = 1;
  self.timer = [NSTimer scheduledTimerWithTimeInterval:1 repeats:YES block:^(NSTimer * _Nonnull timer) {
    if(self.count > 10) {
      events(FlutterEndOfEventStream);
    } else {
      events([NSString stringWithFormat:@"Hello %ld/10", (long)self.count]);
      self.count++;
    }
  }];

  return nil;
}

- (FlutterError *)onCancelWithArguments:(id)arguments {
  [self.timer invalidate];
  self.timer = nil;
  return nil;
}
@end

Flutter端实现

在 Flutter 端，可以通过调用 receiveBroadcastStream 方法来接收来自平台端的数据流。

final StreamsChannel streamsChannel = StreamsChannel('streams_channel_example');

class MyApp extends StatefulWidget {
  [@override](/user/override)
  _MyAppState createState() => new _MyAppState();
}

class _MyAppState extends State<MyApp> {
  StreamSubscription<dynamic> _subscriptionA;
  StreamSubscription<dynamic> _subscriptionB;

  void _start(bool a) {
    StreamSubscription<dynamic> subscription = a ? _subscriptionA : _subscriptionB;

    if (subscription != null) {
      subscription.cancel();
      subscription = null;
    } else {
      final streamId = 'Stream ${a ? 'A' : 'B'}';
      subscription = streamsChannel
          .receiveBroadcastStream(streamId)
          .listen((data) => debugPrint('Received from $streamId: $data'));

      subscription.onDone(() {
        setState(() {
          if (a) {
            _subscriptionA = null;
          } else {
            _subscriptionB = null;
          }
        });
      });
    }

    setState(() {
      if (a) {
        _subscriptionA = subscription;
      } else {
        _subscriptionB = subscription;
      }
    });
  }

  [@override](/user/override)
  Widget build(BuildContext context) {
    return new MaterialApp(
      home: new Scaffold(
        appBar: new AppBar(
          title: new Text('Demo'),
        ),
        body: new Center(
          child: Column(
            mainAxisAlignment: MainAxisAlignment.center,
            children: <Widget>[
              new FlatButton(
                onPressed: () => _start(true),
                child: Text(_subscriptionA != null ? 'Stop A' : 'Start A'),
              ),
              new FlatButton(
                onPressed: () => _start(false),
                child: Text(_subscriptionB != null ? 'Stop B' : 'Start B'),
              ),
            ],
          ),
        ),
      ),
    );
  }
}