refactor(quartz): switch project() builder from channelFlow to flow
channelFlow was overkill for a single-producer chain. The cost was
an extra Channel hop between every projection.snapshot() and the
downstream collector — pure overhead with no concurrency benefit.
flow { } needs the outer collector captured (because inside
changes.onSubscription { ... } and changes.collect { ... } the
implicit `this` is FlowCollector<StoreChange>, not
FlowCollector<ProjectionState<T>>). One `val outer = this` fixes
that.
Backpressure now flows directly: a slow collector suspends emit,
which suspends our changes.collect, which suspends the SharedFlow's
buffer drain — natural propagation, no decoupling Channel in the
middle.
17/17 projection tests pass.
https://claude.ai/code/session_01Jny85MTu1ynKgFBgysfWu5
This commit is contained in:
+9
-5
@@ -35,7 +35,7 @@ import com.vitorpamplona.quartz.nip62RequestToVanish.RequestToVanishEvent
|
|||||||
import com.vitorpamplona.quartz.utils.TimeUtils
|
import com.vitorpamplona.quartz.utils.TimeUtils
|
||||||
import kotlinx.coroutines.flow.Flow
|
import kotlinx.coroutines.flow.Flow
|
||||||
import kotlinx.coroutines.flow.MutableStateFlow
|
import kotlinx.coroutines.flow.MutableStateFlow
|
||||||
import kotlinx.coroutines.flow.channelFlow
|
import kotlinx.coroutines.flow.flow
|
||||||
import kotlinx.coroutines.flow.onSubscription
|
import kotlinx.coroutines.flow.onSubscription
|
||||||
import kotlinx.coroutines.yield
|
import kotlinx.coroutines.yield
|
||||||
|
|
||||||
@@ -410,9 +410,13 @@ class EventStoreProjection<T : Event>(
|
|||||||
* NIP-62 vanish handling is scoped by the inner store's `relay`.
|
* NIP-62 vanish handling is scoped by the inner store's `relay`.
|
||||||
*/
|
*/
|
||||||
fun <T : Event> ObservableEventStore.project(filters: List<Filter>): Flow<ProjectionState<T>> =
|
fun <T : Event> ObservableEventStore.project(filters: List<Filter>): Flow<ProjectionState<T>> =
|
||||||
channelFlow {
|
flow {
|
||||||
val projection = EventStoreProjection<T>(this@project, filters)
|
val projection = EventStoreProjection<T>(this@project, filters)
|
||||||
send(ProjectionState.Loading)
|
// Capture the outer collector so we can emit ProjectionState
|
||||||
|
// from inside `changes.onSubscription { }` and `collect { }`,
|
||||||
|
// where the implicit `this` is FlowCollector<StoreChange>.
|
||||||
|
val outer = this
|
||||||
|
emit(ProjectionState.Loading)
|
||||||
// `onSubscription` runs after the SharedFlow subscription is
|
// `onSubscription` runs after the SharedFlow subscription is
|
||||||
// active but before we pull events — the buffer absorbs
|
// active but before we pull events — the buffer absorbs
|
||||||
// emissions arriving during the seed query and we drain them
|
// emissions arriving during the seed query and we drain them
|
||||||
@@ -422,9 +426,9 @@ fun <T : Event> ObservableEventStore.project(filters: List<Filter>): Flow<Projec
|
|||||||
changes
|
changes
|
||||||
.onSubscription {
|
.onSubscription {
|
||||||
projection.seed()
|
projection.seed()
|
||||||
send(projection.snapshot())
|
outer.emit(projection.snapshot())
|
||||||
}.collect { change ->
|
}.collect { change ->
|
||||||
if (projection.apply(change)) send(projection.snapshot())
|
if (projection.apply(change)) outer.emit(projection.snapshot())
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user