collect method

@override Future collect()

Implementation

@override
Future<dynamic> collect () async {
  final completer = Completer();
  subscription = controller.stream.listen((message) async {
    if (_filter(message)) {
      _messages.putIfAbsent(message.id, () => message);
    }

    if (_messages.length == _max) {
      await _unsubscribe();
      completer.complete(_messages);
    }
  });

  return completer.future;
}