wow-core/me.ahoo.wow.messaging.handler/MessageExchange

MessageExchange

interface MessageExchange<SOURCE : MessageExchange<SOURCE, M>, out M : Message<*, *>> : ErrorAccessor, CommandResultAccessor

Inheritors

CommandExchange

Properties

abstract val attributes: MutableMap<String, Any>

abstract val message: M

Functions

open fun acknowledge(): Mono<Void>

acknowledgementWithheldBy

@InternalWowApi

fun MessageExchange<*, *>.acknowledgementWithheldBy(): String?

Who withheld the acknowledgement of this exchange (withholdAcknowledgement), or null.

open override fun clearError()

Clears the current error object.

confirmLocalDelivery

fun MessageExchange<*, *>.confirmLocalDelivery()

Confirms that this receiver synchronously admitted the local exchange.

extractDeclared

open fun <T : Any> extractDeclared(type: Class<T>): T?

open fun <T : Any> extractObject(type: KType): T?

getAggregateVersion

open fun getAggregateVersion(): Int?

open fun <T> getAttribute(key: String): T?

getCommandResult

open override fun getCommandResult(): Map<String, Any>

open override fun <R> getCommandResult(key: String): R?

open override fun getError(): Throwable?

Gets the current error object.

open fun getFunction(): FunctionInfo?

open fun <FUN : FunctionInfo> getFunctionAs(): FUN?

getServiceProvider

open fun getServiceProvider(): ServiceProvider?

isAcknowledgementWithheld

@InternalWowApi

fun MessageExchange<*, *>.isAcknowledgementWithheld(): Boolean

Whether withholdAcknowledgement was called on this exchange.

rejectLocalDelivery

fun MessageExchange<*, *>.rejectLocalDelivery()

Rejects local-only suppression when this exchange is filtered or cannot be admitted. The distributed copy then remains eligible for processing.

removeAttribute

open fun removeAttribute(key: String): SOURCE

open fun setAttribute(key: String, value: Any): SOURCE

setCommandResult

open override fun setCommandResult(key: String, value: Any)

open override fun setError(throwable: Throwable)

Sets the error object for this context.

open fun setFunction(functionInfo: FunctionInfo): SOURCE

setServiceProvider

open fun setServiceProvider(serviceProvider: ServiceProvider): SOURCE

withholdAcknowledgement

@InternalWowApi

fun MessageExchange<*, *>.withholdAcknowledgement(withheldBy: String = getFunction()?.let { "${it.processorName}.${it.name}" } ?: "unknown")

Leaves this exchange unacknowledged when its processing ends, so a bus that redelivers unacknowledged messages delivers it again. For a processing failure that was neither handled nor durably recorded, for example when recording an event-processing failure for compensation fails.