Channel buffers

ci/licenses_golden/licenses_flutter

@@ -230,6 +230,7 @@ FILE: ../../../flutter/lib/io/dart_io.cc
 FILE: ../../../flutter/lib/io/dart_io.h
 FILE: ../../../flutter/lib/snapshot/libraries.json
 FILE: ../../../flutter/lib/snapshot/snapshot.h
+FILE: ../../../flutter/lib/ui/channel_buffers.dart
 FILE: ../../../flutter/lib/ui/compositing.dart
 FILE: ../../../flutter/lib/ui/compositing/scene.cc
 FILE: ../../../flutter/lib/ui/compositing/scene.h
@@ -428,6 +429,7 @@ FILE: ../../../flutter/lib/web_ui/lib/src/engine/validators.dart
 FILE: ../../../flutter/lib/web_ui/lib/src/engine/vector_math.dart
 FILE: ../../../flutter/lib/web_ui/lib/src/engine/window.dart
 FILE: ../../../flutter/lib/web_ui/lib/src/ui/canvas.dart
+FILE: ../../../flutter/lib/web_ui/lib/src/ui/channel_buffers.dart
 FILE: ../../../flutter/lib/web_ui/lib/src/ui/compositing.dart
 FILE: ../../../flutter/lib/web_ui/lib/src/ui/geometry.dart
 FILE: ../../../flutter/lib/web_ui/lib/src/ui/hash_codes.dart

lib/ui/channel_buffers.dart

@@ -0,0 +1,173 @@
+// Copyright 2013 The Flutter Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+part of dart.ui;
+
+/// A saved platform message for a channel with its callback.
+class _StoredMessage {
+  /// Default constructor, takes in a [ByteData] that represents the
+  /// payload of the message and a [PlatformMessageResponseCallback]
+  /// that represents the callback that will be called when the message
+  /// is handled.
+  _StoredMessage(this._data, this._callback);
+
+  final ByteData _data;
+  final PlatformMessageResponseCallback _callback;
+
+  /// Getter for _data field, represents the message's payload.
+  ByteData get data => _data;
+  /// Getter for the _callback field, to be called when the message is received.
+  PlatformMessageResponseCallback get callback => _callback;
+}
+
+/// A fixed-size circular queue.
+class _RingBuffer<T> {
+  final collection.ListQueue<T> _queue;
+  int _capacity;
+  Function(T) _dropItemCallback;
+
+  _RingBuffer(this._capacity)
+    : _queue = collection.ListQueue<T>(_capacity);
+
+  int get length => _queue.length;
+
+  int get capacity => _capacity;
+
+  bool get isEmpty => _queue.isEmpty;
+
+  set dropItemCallback(Function(T) callback) {
+    _dropItemCallback = callback;
+  }
+
+  /// Returns true on overflow.
+  bool push(T val) {
+    bool overflow = false;
+    while (_queue.length >= _capacity) {
+      final T item = _queue.removeFirst();
+      if (_dropItemCallback != null) {
+        _dropItemCallback(item);
+      }
+      overflow = true;
+    }
+    _queue.addLast(val);
+    return overflow;
+  }
+
+  /// Returns null when empty.
+  T pop() {
+    return _queue.isEmpty ? null : _queue.removeFirst();
+  }
+
+  /// Returns the number of discarded items resulting from resize.
+  int resize(int newSize) {
+    int result = 0;
+
+    while (length > newSize) {
+      result += 1;
+      final T item = _queue.removeFirst();
+      if (_dropItemCallback != null) {
+        _dropItemCallback(item);
+      }
+    }
+
+    _capacity = newSize;
+
+    return result;
+  }
+}
+
+typedef DrainChannelCallback = Future<void> Function(ByteData, PlatformMessageResponseCallback);
+
+/// Storage of channel messages until the channels are completely routed,
+/// i.e. when a message handler is attached to the channel on the framework side.
+///
+/// Each channel has a finite buffer capacity and in a FIFO manner messages will
+/// be deleted if the capacity is exceeded.  The intention is that these buffers
+/// will be drained once a callback is setup on the BinaryMessenger in the
+/// Flutter framework.
+class ChannelBuffers {
+  /// By default we store one message per channel.  There are tradeoffs associated
+  /// with any size.  The correct size should be chosen for the semantics of your
+  /// channel.
+  ///
+  /// Size 0 implies you want to ignore any message that gets sent before the engine
+  /// is ready (keeping in mind there is no way to know when the engine is ready).
+  ///
+  /// Size 1 implies that you only care about the most recent value.
+  ///
+  /// Size >1 means you want to process every single message and want to chose a
+  /// buffer size that will avoid any overflows.
+  static const int kDefaultBufferSize = 1;
+
+  final Map<String, _RingBuffer<_StoredMessage>> _messages =
+    <String, _RingBuffer<_StoredMessage>>{};
+
+  _RingBuffer<_StoredMessage> _makeRingBuffer(int size) {
+    final _RingBuffer<_StoredMessage> result = _RingBuffer<_StoredMessage>(size);
+    result.dropItemCallback = _onDropItem;
+    return result;
+  }
+
+  void _onDropItem(_StoredMessage message) {
+    message.callback(null);
+  }
+
+  /// Returns true on overflow.
+  bool push(String channel, ByteData data, PlatformMessageResponseCallback callback) {
+    _RingBuffer<_StoredMessage> queue = _messages[channel];
+    if (queue == null) {
+      queue = _makeRingBuffer(kDefaultBufferSize);
+      _messages[channel] = queue;
+    }
+    final bool result = queue.push(_StoredMessage(data, callback));
+    if (result) {
+      // TODO(aaclarke): Update this message to include instructions on how to resize
+      // the buffer once that is available to users.
+      _Logger._printString('Overflow on channel: $channel.  '
+                           'Messages on this channel are being sent faster '
+                           'than they are being processed which is resulting '
+                           'in the dropping of messages.  The engine may not be '
+                           'running or you need to adjust the buffer size.');
+    }
+    return result;
+  }
+
+  /// Returns null on underflow.
+  _StoredMessage _pop(String channel) {
+    final _RingBuffer<_StoredMessage> queue = _messages[channel];
+    final _StoredMessage result = queue?.pop();
+    return result;
+  }
+
+  bool _isEmpty(String channel) {
+    final _RingBuffer<_StoredMessage> queue = _messages[channel];
+    return (queue == null) ? true : queue.isEmpty;
+  }
+
+  void resize(String channel, int newSize) {
+    _RingBuffer<_StoredMessage> queue = _messages[channel];
+    if (queue == null) {
+      queue = _makeRingBuffer(newSize);
+      _messages[channel] = queue;
+    } else {
+      final int numberOfDroppedMessages = queue.resize(newSize);
+      if (numberOfDroppedMessages > 0) {
+        _Logger._printString('Dropping messages on channel "$channel" as a result of shrinking the buffer size.');
+      }
+    }
+  }
+
+  /// Remove and process all stored messages for a given channel.
+  ///
+  /// This should be called once a channel is prepared to handle messages
+  /// (ie when a message handler is setup in the framework).
+  Future<void> drain(String channel, DrainChannelCallback callback) async {
+    while (!_isEmpty(channel)) {
+      final _StoredMessage message = _pop(channel);
+      await callback(message.data, message.callback);
+    }
+  }
+}
+
+final ChannelBuffers channelBuffers = ChannelBuffers();

lib/ui/dart_ui.gni

@@ -3,6 +3,7 @@
 # found in the LICENSE file.
 
 dart_ui_files = [
+  "$flutter_root/lib/ui/channel_buffers.dart",
   "$flutter_root/lib/ui/compositing.dart",
   "$flutter_root/lib/ui/geometry.dart",
   "$flutter_root/lib/ui/hash_codes.dart",

lib/ui/hooks.dart

@@ -161,7 +161,9 @@ void _dispatchPlatformMessage(String name, ByteData data, int responseId) {
       },
     );
   } else {
-    window._respondToPlatformMessage(responseId, null);
+    channelBuffers.push(name, data, (ByteData responseData) {
+      window._respondToPlatformMessage(responseId, responseData);
+    });
   }
 }
 

lib/ui/ui.dart

@@ -22,6 +22,7 @@ import 'dart:math' as math;
 import 'dart:nativewrappers';
 import 'dart:typed_data';
 
+part 'channel_buffers.dart';
 part 'compositing.dart';
 part 'geometry.dart';
 part 'hash_codes.dart';

lib/web_ui/lib/src/ui/channel_buffers.dart

@@ -0,0 +1,33 @@
+// Copyright 2013 The Flutter Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+part of ui;
+
+typedef DrainChannelCallback = Future<void> Function(ByteData, PlatformMessageResponseCallback);
+
+/// Storage of channel messages until the channels are completely routed,
+/// i.e. when a message handler is attached to the channel on the framework side.
+///
+/// Each channel has a finite buffer capacity and in a FIFO manner messages will
+/// be deleted if the capacity is exceeded.  The intention is that these buffers
+/// will be drained once a callback is setup on the BinaryMessenger in the
+/// Flutter framework.
+class ChannelBuffers {
+  /// Returns true on overflow.
+  bool push(String channel, ByteData data, PlatformMessageResponseCallback callback) {
+    callback(null);
+    return true;
+  }
+
+  void resize(String channel, int newSize) {}
+
+  /// Remove and process all stored messages for a given channel.
+  ///
+  /// This should be called once a channel is prepared to handle messages
+  /// (ie when a message handler is setup in the framework).
+  Future<void> drain(String channel, DrainChannelCallback callback) async {
+  }
+}
+
+final ChannelBuffers channelBuffers = ChannelBuffers();

lib/web_ui/lib/ui.dart

@@ -23,6 +23,7 @@ export 'src/engine.dart'
         webOnlyInitializeEngine;
 
 part 'src/ui/canvas.dart';
+part 'src/ui/channel_buffers.dart';
 part 'src/ui/compositing.dart';
 part 'src/ui/geometry.dart';
 part 'src/ui/hash_codes.dart';

testing/dart/channel_buffers_test.dart

@@ -0,0 +1,120 @@
+import 'dart:ui' as ui;
+import 'dart:typed_data';
+import 'dart:convert';
+
+import 'package:test/test.dart';
+
+void main() {
+
+  ByteData _makeByteData(String str) {
+    var list = utf8.encode(str);
+    var buffer = list is Uint8List ? list.buffer : new Uint8List.fromList(list).buffer;
+    return ByteData.view(buffer);
+  }
+
+  String _getString(ByteData data) {
+    final buffer = data.buffer;
+    var list = buffer.asUint8List(data.offsetInBytes, data.lengthInBytes);
+    return utf8.decode(list);
+  }
+
+  test('push drain', () async {
+    String channel = "foo";
+    ByteData data = _makeByteData('bar');
+    ui.ChannelBuffers buffers = ui.ChannelBuffers();
+    ui.PlatformMessageResponseCallback callback = (ByteData responseData) {};
+    buffers.push(channel, data, callback);
+    await buffers.drain(channel, (ByteData drainedData, ui.PlatformMessageResponseCallback drainedCallback) {
+      expect(drainedData, equals(data));
+      expect(drainedCallback, equals(callback));
+    });
+  });
+
+  test('empty', () async {
+    String channel = "foo";
+    ByteData data = _makeByteData('bar');
+    ui.ChannelBuffers buffers = ui.ChannelBuffers();
+    ui.PlatformMessageResponseCallback callback = (ByteData responseData) {};
+    bool didCall = false;
+    await buffers.drain(channel, (ByteData drainedData, ui.PlatformMessageResponseCallback drainedCallback) {
+      didCall = true;
+    });
+    expect(didCall, equals(false));
+  });
+
+  test('overflow', () async {
+    String channel = "foo";
+    ByteData one = _makeByteData('one');
+    ByteData two = _makeByteData('two');
+    ByteData three = _makeByteData('three');
+    ByteData four = _makeByteData('four');
+    ui.ChannelBuffers buffers = ui.ChannelBuffers();
+    ui.PlatformMessageResponseCallback callback = (ByteData responseData) {};
+    buffers.resize(channel, 3);
+    expect(buffers.push(channel, one, callback), equals(false));
+    expect(buffers.push(channel, two, callback), equals(false));
+    expect(buffers.push(channel, three, callback), equals(false));
+    expect(buffers.push(channel, four, callback), equals(true));
+    int counter = 0;
+    await buffers.drain(channel, (ByteData drainedData, ui.PlatformMessageResponseCallback drainedCallback) {
+      if (counter++ == 0) {
+        expect(drainedData, equals(two));
+        expect(drainedCallback, equals(callback));
+      }
+    });
+    expect(counter, equals(3));
+  });
+
+  test('resize drop', () async {
+    String channel = "foo";
+    ByteData one = _makeByteData('one');
+    ByteData two = _makeByteData('two');
+    ui.ChannelBuffers buffers = ui.ChannelBuffers();
+    buffers.resize(channel, 100);
+    ui.PlatformMessageResponseCallback callback = (ByteData responseData) {};
+    expect(buffers.push(channel, one, callback), equals(false));
+    expect(buffers.push(channel, two, callback), equals(false));
+    buffers.resize(channel, 1);
+    int counter = 0;
+    await buffers.drain(channel, (ByteData drainedData, ui.PlatformMessageResponseCallback drainedCallback) {
+      if (counter++ == 0) {
+        expect(drainedData, equals(two));
+        expect(drainedCallback, equals(callback));
+      }
+    });
+    expect(counter, equals(1));
+  });
+
+  test('resize dropping calls callback', () async {
+    String channel = "foo";
+    ByteData one = _makeByteData('one');
+    ByteData two = _makeByteData('two');
+    ui.ChannelBuffers buffers = ui.ChannelBuffers();
+    bool didCallCallback = false;
+    ui.PlatformMessageResponseCallback oneCallback = (ByteData responseData) {
+      didCallCallback = true;
+    };
+    ui.PlatformMessageResponseCallback twoCallback = (ByteData responseData) {};
+    buffers.resize(channel, 100);
+    expect(buffers.push(channel, one, oneCallback), equals(false));
+    expect(buffers.push(channel, two, twoCallback), equals(false));
+    buffers.resize(channel, 1);
+    expect(didCallCallback, equals(true));
+  });
+
+  test('overflow calls callback', () async {
+    String channel = "foo";
+    ByteData one = _makeByteData('one');
+    ByteData two = _makeByteData('two');
+    ui.ChannelBuffers buffers = ui.ChannelBuffers();
+    bool didCallCallback = false;
+    ui.PlatformMessageResponseCallback oneCallback = (ByteData responseData) {
+      didCallCallback = true;
+    };
+    ui.PlatformMessageResponseCallback twoCallback = (ByteData responseData) {};
+    buffers.resize(channel, 1);
+    expect(buffers.push(channel, one, oneCallback), equals(false));
+    expect(buffers.push(channel, two, twoCallback), equals(true));
+    expect(didCallCallback, equals(true));
+  });
+}

testing/dart/window_hooks_integration_test.dart

@@ -18,6 +18,7 @@ import 'dart:typed_data';
 import 'package:test/test.dart';
 
 // HACK: these parts are to get access to private functions tested here.
+part '../../lib/ui/channel_buffers.dart';
 part '../../lib/ui/compositing.dart';
 part '../../lib/ui/geometry.dart';
 part '../../lib/ui/hash_codes.dart';