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

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

StreamsChannel for Flutter插件开发

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

理由

EventChannel仅允许每个通道创建一个开放的流。第一次订阅时，该流将被打开，并且可以从Flutter向平台端传递参数。后续订阅将重用先前打开的流或使用新的参数覆盖它。

为了在同一时间支持多个开放的流，且这些流具有不同的初始化参数，必须为每个流创建多个EventChannel。如果流的数量是动态的，这会变得复杂。

安装

请参考以下说明进行安装：

https://pub.dev/packages/streams_channel#-installing-tab-

如何工作

StreamsChannel的工作方式与EventChannel类似，区别在于它支持在同一时间打开多个流。

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

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

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

1. 第一次订阅流时，将使用平台端的工厂创建一个新的流处理器，然后触发处理器上的onListen；
2. 最后一次订阅流时，将触发处理器上的onCancel，然后销毁处理器实例；
3. 如果之后再次发生第一次订阅，则重复步骤1。

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

示例

Flutter端

final StreamsChannel channel = new StreamsChannel('some_channel');

// 持续从平台端接收事件，匹配某些参数
channel.receiveBroadcastStream('some args')
  .listen((data) {
    // 处理数据
  });

// 另一个持续从平台端接收事件，匹配其他参数
channel.receiveBroadcastStream('some other args')
  .listen((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

示例代码

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

import 'dart:async';

import 'package:flutter/material.dart';
import 'package:streams_channel3/streams_channel3.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 ? '停止 A' : '开始 A'),
              ),
              FlatButton(
                onPressed: () => _start(false),
                child: Text(_subscriptionB != null ? '停止 B' : '开始 B'),
              ),
            ],
          ),
        ),
      ),
    );
  }
}