wow-core/me.ahoo.wow.messaging/MessageReceiver

MessageReceiver

class MessageReceiver<E : Any>(messages: Flux<E>, val readiness: Mono<Void> = Mono.empty(), processingAdmission: () -> Unit = {}, processingQuiescence: () -> Unit = {})

A single message source with separate readiness and processing admission.

messages must be subscribed before readiness. The readiness signal is hot and replayable: it completes only after the subscribed source can retain messages without loss, or fails when that setup cannot complete. Once every runtime component is ready, openProcessing explicitly opens transport consumption. closeProcessing revokes that logical admission before physical cancellation, without inferring lifecycle state from reactive demand or subscription count.

Constructors

MessageReceiver

constructor(messages: Flux<E>, readiness: Mono<Void> = Mono.empty(), processingAdmission: () -> Unit = {}, processingQuiescence: () -> Unit = {})

Properties

messages

val messages: Flux<E>

readiness

val readiness: Mono<Void>

Functions

closeProcessing

fun closeProcessing()

Revokes processing admission synchronously without waiting for physical source cancellation. The callback must be prompt and idempotent.

mapMessages

fun <R : Any> mapMessages(transform: (Flux<E>) -> Flux<R>): MessageReceiver<R>

openedMessages

fun openedMessages(): Flux<E>

messages with processing opened on subscription: the stream a consumer that is ready at once reads, so a transport that gates consumption on openProcessing streams immediately. Like messages, it supports exactly one subscriber.

openProcessing

fun openProcessing()