wow-kafka/me.ahoo.wow.kafka/KafkaServerCommandExchange

KafkaServerCommandExchange

class KafkaServerCommandExchange<C : Any>(val message: CommandMessage<C>, receiverOffset: ReceiverOffset, val attributes: MutableMap<String, Any> = ConcurrentHashMap()) : ServerCommandExchange<C>

Constructors

constructor(message: CommandMessage<C>, receiverOffset: ReceiverOffset, attributes: MutableMap<String, Any> = ConcurrentHashMap())

Properties

attributes

open override val attributes: MutableMap<String, Any>

message

open override val message: CommandMessage<C>

Functions

acknowledge

open override fun acknowledge(): Mono<Void>

clearError

open override fun clearError()

extractDeclared

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

extractObject

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

extractRequiredObject

open fun <T : Any> extractRequiredObject(type: Class<T>): T

getAggregateProcessor

open fun getAggregateProcessor(): AggregateProcessor<C>?

getAttribute

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

getCommandResult

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

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

getError

open override fun getError(): Throwable?

getEventStream

open fun getEventStream(): DomainEventStream?

getFunction

open fun getFunction(): FunctionInfo?

getFunctionAs

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

getServiceProvider

open fun getServiceProvider(): ServiceProvider?

removeAttribute

open fun removeAttribute(key: String): ServerCommandExchange<C>

setAggregateProcessor

open fun setAggregateProcessor(aggregateProcessor: AggregateProcessor<*>): ServerCommandExchange<C>

setAttribute

open fun setAttribute(key: String, value: Any): ServerCommandExchange<C>

setCommandResult

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

setError

open override fun setError(throwable: Throwable)

setEventStream

open fun setEventStream(eventStream: DomainEventStream): ServerCommandExchange<C>

setFunction

open fun setFunction(functionInfo: FunctionInfo): ServerCommandExchange<C>

setServiceProvider

open fun setServiceProvider(serviceProvider: ServiceProvider): ServerCommandExchange<C>