refactor(async): refactor and tests

Author: hoc081098

lib/src/async/async_queue.dart

@@ -22,30 +22,40 @@ typedef AsyncQueueBlock<T> = Future<T> Function();
 /// Serial queue are often used to synchronize access to a specific value or resource to prevent data races to occur.
 @internal
 class AsyncQueue<T> {
- static const _timeout = Duration(seconds: 30);
-
  final _blockS = StreamController<_AsyncQueueEntry<T>>();
- final _countS = StreamController<int>(sync: true);
  late final DisposeBag _bag;
 
- /// Construct [AsyncQueue].
- AsyncQueue(Object key, void Function() onTimeout) {
+ final _countS = StateSubject<int>(0, sync: true);
+
+ /// Construct an [AsyncQueue].
+ AsyncQueue({
+ required Object key,
+ required Duration timeout,
+ required void Function() onTimeout,
+ }) {
  _bag = DisposeBag(
  const <Object>[], '( AsyncQueue ~ $key ~ ${shortHash(this)} )');
 
  _blockS.disposedBy(_bag);
  _countS.disposedBy(_bag);
 
- final count$ = _countS.stream
- .scan<int?>((acc, value, _) => acc! + value, 0)
- .shareValueNotReplay(null);
- count$
+ // when the queue is empty, we wait for a timeout to occur
+ // and then we call the onTimeout callback.
+ _countS
  .where((count) => count == 0)
- .switchMap((_) => Rx.timer<void>(null, _timeout)
- .where((_) => count$.value == 0)
- .takeUntil(count$.where((count) => count != null && count > 0)))
- .listen((_) => onTimeout())
- .disposedBy(_bag);
+ .switchMap((_) => Rx.timer<void>(null, timeout)
+ .where((_) => _countS.value == 0)
+ .takeUntil(_countS.where((count) => count > 0)))
+ .listen((_) {
+ assert(() {
+ if (_countS.value != 0) {
+ throw StateError('AsyncQueue is not empty!');
+ }
+ return true;
+ }());
+
+ onTimeout();
+ }).disposedBy(_bag);
 
  Future<T> executeBlock(_AsyncQueueEntry<T> entry) {
  final completer = entry.completer;
@@ -64,7 +74,7 @@ class AsyncQueue<T> {
  }).onError<Object>((e, s) {
  completer.completeError(e, s);
  throw e;
- }).whenComplete(() => _countS.add(-1));
+ }).whenComplete(() => _countS.value = _countS.value - 1);
  }
 
  _blockS.stream
@@ -73,7 +83,7 @@ class AsyncQueue<T> {
  .disposedBy(_bag);
  }
 
- /// Close queue.
+ /// Close queue, discard all pending entries.
  Future<void> dispose() => _bag.dispose();
 
  /// Add block to queue.
@@ -87,7 +97,7 @@ class AsyncQueue<T> {
 
  final completer = Completer<T>.sync();
  _blockS.add(_AsyncQueueEntry(completer, block));
- _countS.add(1);
+ _countS.value = _countS.value + 1;
  return completer.future;
  }
 }

lib/src/impl/real_storage.dart

@@ -20,6 +20,7 @@ class RealRxStorage<Key extends Object, Options,
  S extends Storage<Key, Options>> implements RxStorage<Key, Options> {
  static const _initialKeyValue = KeyAndValue<Object, Object>(
  'rx_storage', 'Petrus Nguyen Thai Hoc <hoc081098@gmail.com>', String);
+ static const _asyncQueueTimeout = Duration(seconds: 30);
 
  /// Trigger subject
  final _keyValuesSubject =
@@ -111,8 +112,9 @@ class RealRxStorage<Key extends Object, Options,
  .putIfAbsent(
  key,
  () => AsyncQueue<Object?>(
- key,
- () => _writeQueueResources.remove(key)?.dispose(),
+ key: key,
+ timeout: _asyncQueueTimeout,
+ onTimeout: () => _writeQueueResources.remove(key)?.dispose(),
  ),
  )
  .enqueue(block)

lib/src/interface/rx_storage.dart

@@ -30,7 +30,12 @@ abstract class RxStorage<Key extends Object, Options>
  Stream<Map<Key, Object?>> observeAll([Options options]);
 
  /// Clean up resources - Closes the streams.
+ ///
  /// This method should be called when a [RxStorage] is no longer needed.
- /// Once `dispose` is called, all streams will `not` emit changed value when value changed.
+ /// But in a real application, this method is rarely called.
+ ///
+ /// Once `dispose` is called:
+ /// - All streams will **not** emit changed value when value changed.
+ /// - All pending writing tasks will be discarded.
  Future<void> dispose();
 }

test/async/async_queue_test.dart

@@ -0,0 +1,94 @@
+import 'dart:async';
+import 'dart:math';
+
+import 'package:rx_storage/src/async/async_queue.dart';
+import 'package:rxdart_ext/rxdart_ext.dart';
+import 'package:test/test.dart';
+
+void main() {
+ group('AsyncQueue', () {
+ const timeout = Duration(milliseconds: 500);
+ final extraTimeout = (timeout * 1.1).inMilliseconds;
+ final halfTimeout = (timeout * 0.5).inMilliseconds;
+
+ test('execute in serial order', () async {
+ late AsyncQueue<int> queue;
+
+ final running = <int>[];
+ final ran = <int>[];
+ var simultaneous = 0;
+ var timeoutOccurred = false;
+
+ queue = AsyncQueue<int>(
+ key: 'key',
+ onTimeout: () {
+ queue.dispose();
+ timeoutOccurred = true;
+ },
+ timeout: timeout,
+ );
+
+ final tasks = <Future<void>>[];
+ for (var i = 0; i < 1000; i++) {
+ tasks.add(
+ queue.enqueue(() async {
+ running.add(i);
+ simultaneous = max(simultaneous, running.length);
+
+ await delay(1);
+
+ ran.add(i);
+ running.remove(i);
+ return i;
+ }),
+ );
+ }
+
+ await Future.wait(tasks);
+ expect(simultaneous, lessThanOrEqualTo(1));
+ expect(ran, [...ran]..sort());
+
+ await delay(extraTimeout);
+ expect(timeoutOccurred, isTrue);
+ });
+
+ test('cannot enqueue task after disposed', () async {
+ final queue = AsyncQueue<int>(
+ key: 'key',
+ onTimeout: () {},
+ timeout: timeout,
+ );
+
+ await pumpEventQueue();
+ unawaited(queue.dispose());
+
+ expect(
+ () => queue.enqueue(() => Future.value(0)),
+ throwsA(isStateError),
+ );
+ });
+
+ test('onTimeout is called when timeout occurred', () async {
+ var count = 0;
+
+ final queue = AsyncQueue<void>(
+ key: 'key',
+ onTimeout: () => ++count,
+ timeout: timeout,
+ );
+
+ await pumpEventQueue();
+
+ for (var i = 0; i < 10; i++) {
+ unawaited(queue.enqueue(() => Future.value()));
+ await delay(halfTimeout);
+ expect(count, 0);
+ }
+
+ await delay(extraTimeout);
+ expect(count, 1);
+ });
+ });
+}
+
+void unawaited(Future<void>? future) {}

test/rx_storage_test.dart

@@ -1,3 +1,4 @@
+import 'async/async_queue_test.dart' as async_queue_test;
 import 'logger/default_logger_test.dart' as default_logger_test;
 import 'logger/logger_adapter_test.dart' as logger_adapter_test;
 import 'model/key_and_value_test.dart' as key_and_value_test;
@@ -8,6 +9,9 @@ import 'storage/streams_test.dart' as streams_test;
 void main() async {
  await perf.main();
 
+ // async
+ async_queue_test.main();
+
  // logger tests
  default_logger_test.main();
  logger_adapter_test.main();