rx_storage

Reactive key-value storage for Dart and Flutter, powered by RxDart.

rx_shared_preferences is an extension of this package, check it out!

Author: Petrus Nguyễn Thái Học

RxDart logo

Liked some of my work? Buy me a coffee (or more likely a beer).

Contents

Getting started
Observing changes
Storage adapters and codecs
Ordering and disposal
Migrating to 4.0.0
Links and support

1. Getting started

Requires Dart SDK >=3.0.0 <4.0.0.

dart pub add rx_storage

For Flutter projects, use flutter pub add rx_storage.

The built-in InMemoryStorage adapter needs no additional setup:

import 'package:rx_storage/rx_storage.dart';

Future<void> main() async {
  final storage = RxStorage<String, void>(InMemoryStorage<String>());
  final subscription = storage
      .observe<String>('name', (raw) => raw as String?)
      .listen((value) => print('Name: $value'));

  try {
    await storage.write<String>('name', 'Ada', (value) => value);

    final name = await storage.read<String>('name', (raw) => raw as String?);
    print('Read: $name'); // Ada

    final values = await storage.readAll((_, raw) => raw);
    print('All: $values'); // {name: Ada}

    await storage.remove('name');
  } finally {
    await subscription.cancel();
    await storage.dispose();
  }
}

For multiple observers and bulk decoding, run the repository example:

dart run example/rx_storage_example.dart

2. Observing changes

Both methods return single-subscription streams. Create one stream per listener. Only successful mutations through the same wrapper trigger notifications. Direct adapter changes and changes through other wrappers are not observed.

2.1. Observe one key

observe(key, decoder) starts with a storage read, then receives changes for that key:

Trigger	Emitted value
Listener subscribes	The stored value, passed through the decoder
`write()` or `update()` succeeds	The value passed to the write, without decoding it again
`remove()` succeeds	`null`
`clear()` succeeds	`null` if the key appeared in the adapter's `readAll()` before clearing

The initial decoder receives null for both missing keys and stored nulls. It can supply a default. Later mutation events bypass the decoder, so removal still emits null.

Use containsKey() to distinguish a missing key from a stored null, if the adapter retains nulls.
Read and decoding failures become stream errors. Incompatible mutation values produce a TypeError. Null events are always valid.
Equal consecutive values are emitted. Use .distinct() to skip duplicates.

Stream timeline: subscribing to K1 triggers an initial read, followed by events for K1 changes only.

2.2. Observe all entries

observeAll(decoder) reads and decodes all entries on subscription and after each successful write(), update(), remove(), or clear():

final changes = storage.observeAll(
  (key, raw) => raw == null ? null : '$key=$raw',
);

A null decoding result keeps the key. Removed keys are omitted, and empty storage emits an empty map.
Reads and asynchronous decoding complete in stream order. Read or decoding failures become stream errors.
A read may reflect later mutations, rather than the exact state at the mutation that triggered it.

3. Storage adapters and codecs

Storage<Key, Options> adapters determine persistence, durability, null retention, and value copying. Options holds adapter-specific settings.

Encoders convert values for writing. Decoders convert them for reading. Both can be synchronous or asynchronous. Bulk decoders receive each existing key and stored value, allowing different codecs per key.

The built-in InMemoryStorage<Key>:

Starts empty, with no persistence between runs.
Retains encoded nulls until removal or clearing.
Takes a shallow entry snapshot for readAll() and decodes it into a new map.
Keeps objects by reference. Decoders can copy them. Direct object mutations do not trigger stream events.

4. Ordering and disposal

Within one wrapper, write(), remove(), and update() share a queue per key. An update() reads, decodes, transforms, encodes, then writes in that queue.

The queue excludes reads, clear(), other keys, direct adapter access, and other wrappers. It provides no multi-key transactions. update() is experimental.

Await required writes before dispose(). Pending tasks fail with StateError. Started operations can finish without further notifications or logs. Repeated disposal shares one result and runs external cleanup at most once. Stored data and adapter resources remain. Do not use the wrapper afterward.

5. Migrating to 4.0.0

readAll() and observeAll() now require an EntryDecoder<Key> and produce decoded Map<Key, Object?> values. To keep the previous raw-value behavior, pass an identity decoder:

final raw = await storage.readAll((_, raw) => raw);
final rawChanges = storage.observeAll((_, raw) => raw);

Custom adapters must support synchronous and asynchronous entry decoding in readAll(). Custom reactive implementations must also implement observeAll(). Both retain null decoding results and report failures through the future or stream.

Also replace executeUpdate() with named-argument update(). Dart 3 class modifiers now restrict inheritance and implementation. See the changelog for all breaking changes.