Utility classes and functions commonly used in DataSource integrations, such as efficiently observable Maps, and stream transformations.
Key components
Observable maps and shared flows
| Component | Description |
|---|
| / | A that is also observable via , , and per-key . Built with / , or collected from an event stream with . |
| / | A variant that also propagates completion and error events to subscribers. Built with . |
| Keyed cache of s, sharing one upstream collection per key and evicting a key once its upstream ends so the cache can't grow unbounded. Built with , or for a that propagates completion/errors. |
| / | Keyed cache of s, sharing one underlying collection per key. |
Stores
| Component | Description |
|---|
| / | A store-backed map exposing a delta-only stream plus a read-through, Caffeine-bounded cache. Built with / / . |
| / | Suspending views of the stores whose reads/writes dispatch the store I/O themselves. |
| / / | The SPI you implement to back a ; mutations enlist on the caller's transaction and publish on commit. |
| / | The transaction handle a write enlists on, and a value paired with the store-assigned version. |
Flow operators
| Operator | Description |
|---|
| Buffers elements until a quiet period elapses, then emits them as a . |
| Emits at most once per interval, keeping the latest value and dropping older ones. |
| Applies a function to the first element together with the entire upstream flow. |
| Groups elements by a key selector and processes each group's sub-flow. |
| Retries the upstream on error with an exponential delay between attempts. |
| Casts a to a . |
| / / | Errors / emits null / emits a default if no first element arrives within a timeout. |
| / (and ) | Convert between flow values and events. |
Event models and folds
| Component | Description |
|---|
| / / / | Sealed event types describing map and set mutations (with or without old values / versions). |
| / | Materialised stream events: initial-state-plus-deltas, and value-or-terminal-signal. |
| / | Fold a delta stream into a live of map / set snapshots. |
| / | Collapse per-key updates, and expand a collection stream into entry events. |
Serialization
| Component | Description |
|---|
| / | Register Fory serializers for the event types and persistent collections. |
| , / | Jackson 2 / Jackson 3 modules that serialize the event types without annotations. |
| / | implementations backed by a Jackson . |
DataSource and general utilities
| Component | Description |
|---|
| A matching subjects by Ant-style patterns, with path-variable extraction. |
| / | Build a from a simplified config for tests and examples. |
| An slf4j wrapper with lazily-evaluated message lambdas. |
| A non-reentrant suspending read/write lock. |
| A coroutine timeout that throws rather than a . |
Samples
val store =
mutableFlowStore(
JooqAccountStore(rootDsl),
Caffeine.newBuilder().maximumSize(10_000),
txContext = Configuration::asTxContext,
)
// Install the publishing listener once on the DSLContext; transactions opened from it run the
// store's buffered commit/rollback actions in their commit/rollback callbacks.
val dsl =
rootDsl
.configuration()
.derive(DefaultTransactionListenerProvider(FlowStorePublishingListener))
.dsl()
withContext(Dispatchers.IO) {
dsl.transaction { config ->
val alice = store.get("alice", config) ?: Account("alice", 0)
val bob = store.get("bob", config) ?: Account("bob", 0)
store.put("alice", alice.copy(balance = alice.balance - 10), config)
store.put("bob", bob.copy(balance = bob.balance + 10), config)
}
}