Allow sending platform messages from plugins to the framework and implement EventChannel (#40638)

* Allow sending messages from the platform to the framework

* Add ability to send messages from the plugin to the platform.

Also adds [PluginEventChannel] which is the plugin counterpart to
EventChannel.

Fixes https://github.com/flutter/flutter/issues/39981

* fix analyzer errors

* Enhance doc comments

* Remove dead code in test

packages/flutter_web_plugins/lib/flutter_web_plugins.dart

@@ -2,4 +2,5 @@
 // Use of this source code is governed by a BSD-style license that can be
 // found in the LICENSE file.
 
+export 'src/plugin_event_channel.dart';
 export 'src/plugin_registry.dart';

packages/flutter_web_plugins/lib/src/plugin_event_channel.dart

@@ -0,0 +1,110 @@
+// Copyright 2019 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+import 'dart:async';
+
+import 'package:flutter/services.dart';
+
+import 'plugin_registry.dart';
+
+/// A named channel for sending events to the framework-side using streams.
+///
+/// This is the platform-side equivalent of [EventChannel]. Whereas
+/// [EventChannel] receives a stream of events from platform plugins, this
+/// channel sends a stream of events to the handler listening on the
+/// framework-side.
+///
+/// The channel [name] must not be null. If no [codec] is provided, then
+/// [StandardMethodCodec] is used. If no [binaryMessenger] is provided, then
+/// [pluginBinaryMessenger], which sends messages to the framework-side,
+/// is used.
+class PluginEventChannel<T> {
+  /// Creates a new plugin event channel.
+  const PluginEventChannel(
+    this.name, [
+    this.codec = const StandardMethodCodec(),
+    BinaryMessenger binaryMessenger,
+  ])  : assert(name != null),
+        assert(codec != null),
+        _binaryMessenger = binaryMessenger;
+
+  /// The logical channel on which communication happens.
+  ///
+  /// This must not be null.
+  final String name;
+
+  /// The message codec used by this channel.
+  ///
+  /// This must not be null. This defaults to [StandardMethodCodec].
+  final MethodCodec codec;
+
+  /// The messenger used by this channel to send platform messages.
+  ///
+  /// This must not be null. If not provided, defaults to
+  /// [pluginBinaryMessenger], which sends messages from the platform-side
+  /// to the framework-side.
+  BinaryMessenger get binaryMessenger =>
+      _binaryMessenger ?? pluginBinaryMessenger;
+  final BinaryMessenger _binaryMessenger;
+
+  /// Set the stream controller for this event channel.
+  set controller(StreamController<T> controller) {
+    final _EventChannelHandler<T> handler = _EventChannelHandler<T>(
+      name,
+      codec,
+      controller,
+      binaryMessenger,
+    );
+    binaryMessenger.setMessageHandler(
+        name, controller == null ? null : handler.handle);
+  }
+}
+
+class _EventChannelHandler<T> {
+  _EventChannelHandler(this.name, this.codec, this.controller, this.messenger);
+
+  final String name;
+  final MethodCodec codec;
+  final StreamController<T> controller;
+  final BinaryMessenger messenger;
+
+  StreamSubscription<T> subscription;
+
+  Future<ByteData> handle(ByteData message) {
+    final MethodCall call = codec.decodeMethodCall(message);
+    switch (call.method) {
+      case 'listen':
+        return _listen();
+      case 'cancel':
+        return _cancel();
+    }
+    return null;
+  }
+
+  // TODO(hterkelsen): Support arguments.
+  Future<ByteData> _listen() async {
+    if (subscription != null) {
+      await subscription.cancel();
+    }
+    subscription = controller.stream.listen((dynamic event) {
+      messenger.send(name, codec.encodeSuccessEnvelope(event));
+    }, onError: (dynamic error) {
+      messenger.send(name,
+          codec.encodeErrorEnvelope(code: 'error', message: error.toString()));
+    });
+
+    return codec.encodeSuccessEnvelope(null);
+  }
+
+  // TODO(hterkelsen): Support arguments.
+  Future<ByteData> _cancel() async {
+    if (subscription == null) {
+      return codec.encodeErrorEnvelope(
+          code: 'error', message: 'No active stream to cancel.');
+    }
+    await subscription.cancel();
+    subscription = null;
+    return codec.encodeSuccessEnvelope(null);
+  }
+}

packages/flutter_web_plugins/lib/src/plugin_registry.dart

@@ -96,8 +96,20 @@ class _PlatformBinaryMessenger extends BinaryMessenger {
   /// Sends a platform message from the platform side back to the framework.
   @override
   Future<ByteData> send(String channel, ByteData message) {
-    throw FlutterError(
-        'Cannot send messages from the platform side to the framework.');
+    final Completer<ByteData> completer = Completer<ByteData>();
+    ui.window.onPlatformMessage(channel, message, (ByteData reply) {
+      try {
+        completer.complete(reply);
+      } catch (exception, stack) {
+        FlutterError.reportError(FlutterErrorDetails(
+          exception: exception,
+          stack: stack,
+          library: 'flutter web shell',
+          context: ErrorDescription('during a plugin-to-framework message'),
+        ));
+      }
+    });
+    return completer.future;
   }
 
   @override

packages/flutter_web_plugins/test/plugin_event_channel_test.dart

@@ -0,0 +1,90 @@
+// Copyright 2019 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+@TestOn('chrome') // Uses web-only Flutter SDK
+
+import 'dart:async';
+
+import 'package:flutter/services.dart';
+import 'package:flutter_test/flutter_test.dart';
+import 'package:flutter_web_plugins/flutter_web_plugins.dart';
+
+void main() {
+  group('Plugin Event Channel', () {
+    setUp(() {
+      TestWidgetsFlutterBinding.ensureInitialized();
+      webPluginRegistry.registerMessageHandler();
+    });
+
+    test('can send events to an $EventChannel', () async {
+      const EventChannel listeningChannel = EventChannel('test');
+      const PluginEventChannel<String> sendingChannel =
+          PluginEventChannel<String>('test');
+
+      final StreamController<String> controller = StreamController<String>();
+      sendingChannel.controller = controller;
+
+      expect(listeningChannel.receiveBroadcastStream(),
+          emitsInOrder(<String>['hello', 'world']));
+
+      controller.add('hello');
+      controller.add('world');
+      await controller.close();
+    });
+
+    test('can send errors to an $EventChannel', () async {
+      const EventChannel listeningChannel = EventChannel('test2');
+      const PluginEventChannel<String> sendingChannel =
+          PluginEventChannel<String>('test2');
+
+      final StreamController<String> controller = StreamController<String>();
+      sendingChannel.controller = controller;
+
+      expect(
+          listeningChannel.receiveBroadcastStream(),
+          emitsError(predicate<dynamic>((dynamic e) =>
+              e is PlatformException && e.message == 'Test error')));
+
+      controller.addError('Test error');
+      await controller.close();
+    });
+
+    test('receives a listen event', () async {
+      const EventChannel listeningChannel = EventChannel('test3');
+      const PluginEventChannel<String> sendingChannel =
+          PluginEventChannel<String>('test3');
+
+      final StreamController<String> controller = StreamController<String>(
+          onListen: expectAsync0<void>(() {}, count: 1));
+      sendingChannel.controller = controller;
+
+      expect(listeningChannel.receiveBroadcastStream(),
+          emitsInOrder(<String>['hello']));
+
+      controller.add('hello');
+      await controller.close();
+    });
+
+    test('receives a cancel event', () async {
+      const EventChannel listeningChannel = EventChannel('test4');
+      const PluginEventChannel<String> sendingChannel =
+          PluginEventChannel<String>('test4');
+
+      final StreamController<String> controller =
+          StreamController<String>(onCancel: expectAsync0<void>(() {}));
+      sendingChannel.controller = controller;
+
+      final Stream<dynamic> eventStream =
+          listeningChannel.receiveBroadcastStream();
+      StreamSubscription<dynamic> subscription;
+      subscription =
+          eventStream.listen(expectAsync1<void, dynamic>((dynamic x) {
+        expect(x, equals('hello'));
+        subscription.cancel();
+      }));
+
+      controller.add('hello');
+    });
+  });
+}

packages/flutter_web_plugins/test/plugin_registry_test.dart

@@ -31,27 +31,43 @@ void main() {
     setUp(() {
       TestWidgetsFlutterBinding.ensureInitialized();
       webPluginRegistry.registerMessageHandler();
-    });
-
-    test('Can register a plugin', () {
-      TestPlugin.calledMethods.clear();
-
       final Registrar registrar = webPluginRegistry.registrarFor(TestPlugin);
       TestPlugin.registerWith(registrar);
+    });
+
+    test('can register a plugin', () {
+      TestPlugin.calledMethods.clear();
 
       const MethodChannel frameworkChannel =
           MethodChannel('test_plugin', StandardMethodCodec());
       frameworkChannel.invokeMethod<void>('test1');
 
-      expect(TestPlugin.calledMethods, <String>['test1']);
+      expect(TestPlugin.calledMethods, equals(<String>['test1']));
     });
 
-    test('Throws when trying to send a platform message to the framework', () {
-      expect(() => pluginBinaryMessenger.send('test', ByteData(0)),
-          throwsFlutterError);
+    test('can send a message from the plugin to the framework', () async {
+      const StandardMessageCodec codec = StandardMessageCodec();
+
+      final List<String> loggedMessages = <String>[];
+      ServicesBinding.instance.defaultBinaryMessenger
+          .setMessageHandler('test_send', (ByteData data) {
+        loggedMessages.add(codec.decodeMessage(data));
+        return null;
+      });
+
+      await pluginBinaryMessenger.send(
+          'test_send', codec.encodeMessage('hello'));
+      expect(loggedMessages, equals(<String>['hello']));
+
+      await pluginBinaryMessenger.send(
+          'test_send', codec.encodeMessage('world'));
+      expect(loggedMessages, equals(<String>['hello', 'world']));
+
+      ServicesBinding.instance.defaultBinaryMessenger
+          .setMessageHandler('test_send', null);
     });
 
-    test('Throws when trying to set a mock handler', () {
+    test('throws when trying to set a mock handler', () {
       expect(
           () => pluginBinaryMessenger.setMockMessageHandler(
               'test', (ByteData data) async => ByteData(0)),