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

StreamsChannel for Flutter plugin development

StreamsChannel 是受 EventChannel 启发而开发的，它允许在 Flutter 和平台端之间创建事件流。

Rationale

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

为了在同一时间支持多个打开的流，并为流初始化提供不同的参数，需要创建多个 EventChannel。如果流的数量是动态的，这会变得复杂。

Installation

请按照以下说明进行安装：
https://pub.dev/packages/streams_channel#-installing-tab-

How it works

StreamsChannel 的工作方式与 EventChannel 类似，但它的特点是能够同时支持多个打开的流。

平台端

在平台端，通道接受一个流处理器工厂：

let channel = FlutterStreamsChannel(name: "my_channel", binaryMessenger: plugin.registrar.messenger())
channel.setStreamHandlerFactory { arguments in
  return MyHandler(arguments) // MyHandler 是一个 FlutterStreamHandler 实例
}

Flutter 端

在 Flutter 端，每次调用 receiveBroadcastStream(args) 都会创建一个新的流，不会替换任何仍在运行的旧流。

1. 第一次订阅流时，会在平台端使用工厂创建一个新的流处理器，然后触发处理器的 onListen 方法；
2. 最后一次取消订阅时，触发处理器的 onCancel 方法，然后销毁处理器实例；
3. 如果之后有新的第一次订阅，重复步骤 1。

底层实现中，StreamsChannel 使用简单的自增唯一 ID 来跟踪每个流。

Example

Flutter 端

final StreamsChannel streamsChannel = StreamsChannel('streams_channel_example');

// 持续从平台端接收带有某些参数的事件流
streamsChannel.receiveBroadcastStream('Stream A')
  .listen((data) {
    debugPrint('Received from Stream A: $data');
  });

// 另一个持续从平台端接收带有其他参数的事件流
streamsChannel.receiveBroadcastStream('Stream B')
  .listen((data) {
    debugPrint('Received from Stream B: $data');
  });

平台端

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 端代码 (main.dart)

// Copyright (c) 2021 Kfir Matityahu
// Licensed under Apache License v2.0

import 'dart:async';

import 'package:flutter/material.dart';
import 'package:streams_channel2/streams_channel2.dart';

void main() => runApp(MyApp());

final StreamsChannel streamsChannel = StreamsChannel('streams_channel_example');

class MyApp extends StatefulWidget {
  [@override](/user/override)
  _MyAppState createState() => _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 MaterialApp(
      home: Scaffold(
        appBar: AppBar(
          title: Text('Demo'),
        ),
        body: Center(
          child: Column(
            mainAxisAlignment: MainAxisAlignment.center,
            children: <Widget>[
              FlatButton(
                onPressed: () => _start(true),
                child: Text(_subscriptionA != null ? 'Stop A' : 'Start A'),
              ),
              FlatButton(
                onPressed: () => _start(false),
                child: Text(_subscriptionB != null ? 'Stop B' : 'Start B'),
              ),
            ],
          ),
        ),
      ),
    );
  }
}