Messaging

This section describes the extension functions available when dispatching commands and queries, emitting query updates, and working with message metadata. All of them live in the org.axonframework.extension.kotlin.messaging package.

Commands

`CommandGateway`

The CommandGateway offers two reified extension functions. Both take the expected result type as a type argument, instead of a Class parameter.

sendWithResult<R>(command): Sends the command and returns a CompletableFuture<R>. This is the reified equivalent of CommandGateway#send(Object, Class).
sendAndWait<R>(command): Sends the command, blocks until it is handled, and returns the result as R. This is the reified equivalent of CommandGateway#sendAndWait(Object, Class). Unlike the Java method, the Kotlin function never returns null: it throws an IllegalStateException when the handler returns no result. Use sendWithResult if a null result is a valid outcome.

import org.axonframework.extension.kotlin.messaging.sendAndWait
import org.axonframework.extension.kotlin.messaging.sendWithResult
import org.axonframework.messaging.commandhandling.gateway.CommandGateway
import java.util.concurrent.CompletableFuture

class CardService(private val commandGateway: CommandGateway) {

    // Non-blocking, the result is converted to a String once handling completes
    fun issueCard(command: IssueCardCommand): CompletableFuture<String> =
        commandGateway.sendWithResult<String>(command)

    // Blocking, returns the converted result directly
    fun issueCardAndWait(command: IssueCardCommand): String =
        commandGateway.sendAndWait<String>(command)
}

The extension functions only cover commands that return a result. To dispatch a command without a result type, use the CommandGateway methods directly.

Queries

`QueryGateway`

The QueryGateway offers reified extension functions that take the response type as a type argument, instead of a Class parameter.

Extension function Returns Equivalent QueryGateway method

Extension function	Returns	Equivalent `QueryGateway` method
`query<R>(query)`	`CompletableFuture<R>`	`query(Object, Class)`
`queryMany<R>(query)`	`CompletableFuture<List<R>>`	`queryMany(Object, Class)`
`streamingQuery<R>(query)`	`Publisher<R>`	`streamingQuery(Object, Class)`
`subscriptionQuery<R>(query)`	`Publisher<R>`	`subscriptionQuery(Object, Class)`

query<R>(query)

CompletableFuture<R>

query(Object, Class)

queryMany<R>(query)

CompletableFuture<List<R>>

queryMany(Object, Class)

streamingQuery<R>(query)

Publisher<R>

streamingQuery(Object, Class)

subscriptionQuery<R>(query)

Publisher<R>

subscriptionQuery(Object, Class)

import org.axonframework.extension.kotlin.messaging.query
import org.axonframework.extension.kotlin.messaging.queryMany
import org.axonframework.messaging.queryhandling.gateway.QueryGateway
import java.util.concurrent.CompletableFuture

class CardQueryService(private val queryGateway: QueryGateway) {

    fun totalNumberOfCards(): CompletableFuture<Int> =
        queryGateway.query<Int>(CountCardSummariesQuery())

    fun cardSummaries(): CompletableFuture<List<CardSummary>> =
        queryGateway.queryMany<CardSummary>(FetchCardSummariesQuery())
}

Kotlin can often infer the type argument from the expected type, so the following is equivalent to the first function above:

fun totalNumberOfCards(): CompletableFuture<Int> =
    queryGateway.query(CountCardSummariesQuery())

`QueryUpdateEmitter`

The emit<Q, U>(update) { filter } function on the QueryUpdateEmitter takes the query type and update type as type arguments. Its filter is a Kotlin lambda, which can be written outside the parentheses. Only subscription queries of type Q for which the filter returns true receive the update. The QueryUpdateEmitter is available as a parameter of a message handler, such as the event handler below.

import org.axonframework.extension.kotlin.messaging.emit
import org.axonframework.messaging.eventhandling.annotation.EventHandler
import org.axonframework.messaging.queryhandling.QueryUpdateEmitter

class CardSummaryProjection {

    @EventHandler
    fun on(event: CardIssuedEvent, queryUpdateEmitter: QueryUpdateEmitter) {
        // Update the projection here

        // Then emit the update to the subscribers whose filter matches
        queryUpdateEmitter.emit<CountCardSummariesQuery, CountChangedUpdate>(CountChangedUpdate()) { query ->
            event.id.startsWith(query.idFilter)
        }
    }
}

Metadata

Every message carries metadata next to its payload: an immutable map of string keys to string values. Metadata typically holds contextual information, such as a correlation id, a tenant, or the user that triggered the message. The helpers below make reading and building that metadata feel natural in Kotlin:

AxonMetadata: A type alias for org.axonframework.messaging.core.Metadata. It avoids a clash with Kotlin’s own kotlin.Metadata annotation.
Message.metadata: A property that reads the metadata of any Message, replacing the metadata() call.
AxonMetadata.and(Pair<String, String?>): Adds an entry using Kotlin’s to function.
AxonMetadata.contains(AxonMetadata): Returns true if every entry of the given metadata is present, with an equal value, in the receiver. An empty metadata is therefore contained in any metadata.

import org.axonframework.extension.kotlin.messaging.AxonMetadata
import org.axonframework.extension.kotlin.messaging.and
import org.axonframework.extension.kotlin.messaging.contains
import org.axonframework.extension.kotlin.messaging.metadata
import org.axonframework.messaging.core.Message

fun isFromGame(message: Message, gameId: String): Boolean {
    val expected = AxonMetadata.with("gameId", gameId)
    return message.metadata.contains(expected)
}

val gameMetadata = AxonMetadata.with("gameId", "game-1")
    .and("playerId" to "player-1")