buffer method

@override Future<List<T>> buffer(int count, { Duration timeLimit = const Duration(seconds: 10), })

Buffer changes of a Atom. The count is a number of a buffered items. The timeLimit is 10 seconds by default.

Implementation

@override
Future<List<T>> buffer(int count, {Duration timeLimit = const Duration(seconds: 10)}) async {
  final completer = Completer<List<T>>();
  final values = <T>[];

  final disposable = rxObserver<T>(
    () => value,
    effect: (_) {
      values.add(_value);
      if (count == values.length) {
        completer.complete(values);
      }
    },
  );

  final result = await completer.future.timeout(
    timeLimit,
    onTimeout: () => values,
  );

  disposable();
  return result;
}