Refactor KeyAndValue, Logger. Add AsyncQueue (#5)

* wip

* check

* @nonvirtual

* stack_trace

* update fake_storage.dart

* default_logger_test.dart and fix default_logger.dartr

* test WriteFailureEvent

* clear event tests

* remove events

* ==

* test throws error

* throws all

* update tests

* fixed tests and add async_queue.dart

* fix async_queue.dart

* fix async_queue.dart

* fix async_queue.dart

* fix async_queue.dart

* cancel then close

* rm ex

Author: hoc081098

lib/src/async_memoizer.dart renamed to lib/src/async/async_memoizer.dart

@@ -0,0 +1,67 @@
+import 'dart:async';
+
+import 'package:meta/meta.dart';
+
+class _AsyncQueueEntry<T> {
+  final Completer<T> completer;
+
+  final AsyncQueueBlock<T> block;
+
+  _AsyncQueueEntry(this.completer, this.block);
+}
+
+/// Function that returns a [Future].
+@internal
+typedef AsyncQueueBlock<T> = Future<T> Function();
+
+/// A serial queue that executes single block at a time and it does this in FIFO order, first in, first out.
+/// Serial queue are often used to synchronize access to a specific value or resource to prevent data races to occur.
+@internal
+class AsyncQueue<T> {
+  final _blockS = StreamController<_AsyncQueueEntry<T>>();
+  StreamSubscription<T>? _subscription;
+
+  /// Construct [AsyncQueue].
+  AsyncQueue() {
+    _subscription = _blockS.stream.asyncMap((entry) {
+      final completer = entry.completer;
+
+      Future<T> future;
+      try {
+        future = entry.block();
+      } catch (e, s) {
+        completer.completeError(e, s);
+        rethrow;
+      }
+
+      return future.then((v) {
+        completer.complete(v);
+        return v;
+      }).onError<Object>((Object e, StackTrace s) {
+        completer.completeError(e, s);
+        throw e;
+      });
+    }).listen(null, onError: (Object _) {});
+  }
+
+  /// Close queue.
+  Future<void> dispose() {
+    if (_subscription == null || _blockS.isClosed) {
+      throw StateError('AsyncQueue has been disposed!');
+    }
+    final future = _subscription!.cancel().then<void>((_) => _blockS.close());
+    _subscription = null;
+    return future;
+  }
+
+  /// Add block to queue.
+  Future<T> enqueue(AsyncQueueBlock<T> block) {
+    if (_subscription == null || _blockS.isClosed) {
+      throw StateError('AsyncQueue has been disposed!');
+    }
+
+    final completer = Completer<T>.sync();
+    _blockS.add(_AsyncQueueEntry(completer, block));
+    return completer.future;
+  }
+}

lib/src/async/async_queue.dart

@@ -4,7 +4,8 @@ import 'package:disposebag/disposebag.dart' hide Logger;
 import 'package:meta/meta.dart';
 import 'package:rxdart_ext/rxdart_ext.dart';
 
-import '../async_memoizer.dart';
+import '../async/async_memoizer.dart';
+import '../async/async_queue.dart';
 import '../interface/rx_storage.dart';
 import '../interface/storage.dart';
 import '../logger/event.dart';
@@ -18,11 +19,16 @@ import '../model/key_and_value.dart';
 /// Default [RxStorage] implementation.
 class RealRxStorage<Key extends Object, Options,
     S extends Storage<Key, Options>> implements RxStorage<Key, Options> {
-  static const _initialKeyValue =
-      KeyAndValue('rx_storage', 'Petrus Nguyen Thai Hoc <[email protected]>');
+  static const _initialKeyValue = KeyAndValue<Object, Object>(
+      'rx_storage', 'Petrus Nguyen Thai Hoc <[email protected]>', String);
 
   /// Trigger subject
-  final _keyValuesSubject = PublishSubject<Map<Key, Object?>>();
+  final _keyValuesSubject =
+      PublishSubject<Map<Key, KeyAndValue<Key, Object?>>>();
+
+  /// Write queue.
+  /// Basic lock mechanism to prevent concurrent access to asynchronous code.
+  final _writeQueue = AsyncQueue<Object?>();
 
   final _disposeMemo = AsyncMemoizer<void>();
   late final _bag =
@@ -140,7 +146,7 @@ class RealRxStorage<Key extends Object, Options,
 
   /// Add changed map to subject to trigger.
   @protected
-  void sendChange(Map<Key, Object?> map) {
+  void sendChange(Map<Key, KeyAndValue<Key, Object?>> map) {
     assert(_debugAssertNotDisposed());
     assert(map != null);
 
@@ -151,9 +157,9 @@ class RealRxStorage<Key extends Object, Options,
     }
   }
 
-  /// Log event.
+  /// Log event if logging is enabled.
   @protected
-  void log(LoggerEvent<Key, Options> event) {
+  void logIfEnabled(LoggerEvent<Key, Options> event) {
     assert(_debugAssertNotDisposed());
     assert(event != null);
 
@@ -162,6 +168,11 @@ class RealRxStorage<Key extends Object, Options,
     }
   }
 
+  /// Enqueue writing task to a queue.
+  @protected
+  Future<T> enqueueWritingTask<T>(AsyncQueueBlock<T> block) =>
+      _writeQueue.enqueue(block).then((value) => value as T);
+
   //
   // Get and set methods (implements [Storage])
   //
@@ -186,7 +197,7 @@ class RealRxStorage<Key extends Object, Options,
       (value, _) {
         if (_isLogEnabled) {
           _publishLog(
-              ReadValueSuccessEvent(KeyAndValue(key, value), T, options));
+              ReadValueSuccessEvent(KeyAndValue(key, value, T), options));
         }
       },
       (error, _) {
@@ -218,46 +229,52 @@ class RealRxStorage<Key extends Object, Options,
   }
 
   @override
-  Future<void> clear([Options? options]) async {
+  Future<void> clear([Options? options]) {
     assert(_debugAssertNotDisposed());
 
-    final keys = (await _useStorage((s) => s.readAll(options))).keys;
-
-    return await useStorageWithHandlers(
-      (s) => s.clear(options),
-      (_, __) {
-        sendChange({for (final k in keys) k: null});
-        if (_isLogEnabled) {
-          _publishLog(ClearSuccessEvent(options));
-        }
-      },
-      (error, _) {
-        if (_isLogEnabled) {
-          _publishLog(ClearFailureEvent(error, options));
-        }
-      },
-    );
+    return enqueueWritingTask(() async {
+      final keys = (await _useStorage((s) => s.readAll(options))).keys;
+
+      return await useStorageWithHandlers(
+        (s) => s.clear(options),
+        (_, __) {
+          sendChange({for (final k in keys) k: KeyAndValue(k, null, Null)});
+
+          if (_isLogEnabled) {
+            _publishLog(ClearSuccessEvent(options));
+          }
+        },
+        (error, _) {
+          if (_isLogEnabled) {
+            _publishLog(ClearFailureEvent(error, options));
+          }
+        },
+      );
+    });
   }
 
   @override
   Future<void> remove(Key key, [Options? options]) {
     assert(_debugAssertNotDisposed());
     assert(key != null);
 
-    return useStorageWithHandlers(
-      (s) => s.remove(key, options),
-      (_, __) {
-        sendChange({key: null});
-        if (_isLogEnabled) {
-          _publishLog(RemoveSuccessEvent(key, options));
-        }
-      },
-      (error, _) {
-        if (_isLogEnabled) {
-          _publishLog(RemoveFailureEvent(key, options, error));
-        }
-      },
-    );
+    return enqueueWritingTask(() {
+      return useStorageWithHandlers(
+        (s) => s.remove(key, options),
+        (_, __) {
+          sendChange({key: KeyAndValue(key, null, Null)});
+
+          if (_isLogEnabled) {
+            _publishLog(RemoveSuccessEvent(key, options));
+          }
+        },
+        (error, _) {
+          if (_isLogEnabled) {
+            _publishLog(RemoveFailureEvent(key, options, error));
+          }
+        },
+      );
+    });
   }
 
   @override
@@ -267,21 +284,26 @@ class RealRxStorage<Key extends Object, Options,
     assert(key != null);
     assert(encoder != null);
 
-    return useStorageWithHandlers(
-      (s) => s.write(key, value, encoder, options),
-      (_, __) {
-        sendChange({key: value});
-        if (_isLogEnabled) {
-          _publishLog(WriteSuccessEvent(KeyAndValue(key, value), T, options));
-        }
-      },
-      (error, __) {
-        if (_isLogEnabled) {
-          _publishLog(
-              WriteFailureEvent(KeyAndValue(key, value), T, options, error));
-        }
-      },
-    );
+    return enqueueWritingTask(() {
+      return useStorageWithHandlers(
+        (s) => s.write(key, value, encoder, options),
+        (_, __) {
+          final keyAndValue = KeyAndValue(key, value, T);
+
+          sendChange({key: keyAndValue});
+
+          if (_isLogEnabled) {
+            _publishLog(WriteSuccessEvent(keyAndValue, options));
+          }
+        },
+        (error, __) {
+          if (_isLogEnabled) {
+            _publishLog(
+                WriteFailureEvent(KeyAndValue(key, value, T), options, error));
+          }
+        },
+      );
+    });
   }
 
   //
@@ -294,22 +316,26 @@ class RealRxStorage<Key extends Object, Options,
     assert(_debugAssertNotDisposed());
     assert(key != null);
 
+    FutureOr<T?> convert(KeyAndValue<Object, Object?> entry) {
+      if (identical(entry, _initialKeyValue)) {
+        return _useStorage((s) => s.read<T>(key, decoder, options));
+      } else {
+        // ignore assertion if [entry.type] is `Null` or `dynamic`.
+        assert(entry.type == Null || entry.type == dynamic || entry.type == T);
+        return entry.value as FutureOr<T?>;
+      }
+    }
+
     final stream = _keyValuesSubject
         .toSingleSubscriptionStream()
-        .mapNotNull((map) => map.containsKey(key)
-            ? KeyAndValue<Object, Object?>(key, map[key])
-            : null)
+        .mapNotNull<KeyAndValue<Object, Object?>>((map) => map[key])
         .startWith(_initialKeyValue) // Dummy value to trigger initial load.
-        .asyncMap<T?>(
-          (entry) => identical(entry, _initialKeyValue)
-              ? _useStorage((s) => s.read<T>(key, decoder, options))
-              : entry.value as FutureOr<T?>,
-        );
+        .asyncMap<T?>(convert);
 
     return _isLogEnabled
         ? stream
             .doOnData((value) =>
-                _publishLog(OnDataStreamEvent(KeyAndValue(key, value))))
+                _publishLog(OnDataStreamEvent(KeyAndValue(key, value, T))))
             .doOnError((e, s) => _publishLog(
                 OnErrorStreamEvent(RxStorageError(e, s ?? StackTrace.empty))))
         : stream;
@@ -330,7 +356,9 @@ class RealRxStorage<Key extends Object, Options,
   Future<void> dispose() {
     assert(_debugAssertNotDisposed());
 
-    return _disposeMemo.runOnce(_bag.dispose).then((_) => _onDispose?.call());
+    return _disposeMemo
+        .runOnce(() => _writeQueue.dispose().then((_) => _bag.dispose()))
+        .then((_) => _onDispose?.call());
   }
 }