Native kotlin suspending event gateway

Author: jnfeinstein

kotlin/src/main/kotlin/org/axonframework/extensions/kotlin/eventhandling/gateway/DefaultSuspendingEventGateway.kt

@@ -0,0 +1,72 @@
+package org.axonframework.extensions.kotlin.eventhandling.gateway
+
+import org.axonframework.common.BuilderUtils
+import org.axonframework.common.Registration
+import org.axonframework.eventhandling.EventBus
+import org.axonframework.eventhandling.EventMessage
+import org.axonframework.eventhandling.GenericEventMessage
+import org.axonframework.extensions.kotlin.messaging.SuspendingMessageDispatchInterceptor
+import java.util.*
+import java.util.concurrent.CopyOnWriteArrayList
+
+/**
+ * Implementation of the [SuspendingEventGateway].
+ *
+ * @author Joel Feinstein
+ * @since 0.2.0
+ */
+class DefaultSuspendingEventGateway(config: Builder.() -> Unit) : SuspendingEventGateway {
+    private val eventBus: EventBus
+    private val dispatchInterceptors: MutableList<SuspendingMessageDispatchInterceptor<EventMessage<*>>>
+
+    init {
+        Builder().apply(config).validate().let {
+            eventBus = it.eventBus
+            dispatchInterceptors = CopyOnWriteArrayList(it.dispatchInterceptors)
+        }
+    }
+
+    /**
+     * This implementation will process interceptors and dispatch each event before moving on to the next.
+     */
+    override suspend fun publish(events: Iterable<*>): List<EventMessage<*>> {
+        return events.map {
+            GenericEventMessage.asEventMessage<Any>(it).processEventInterceptors().apply { publish() }
+        }
+    }
+
+    override fun registerDispatchInterceptor(interceptor: SuspendingMessageDispatchInterceptor<EventMessage<*>>): Registration {
+        dispatchInterceptors += interceptor
+        return Registration { dispatchInterceptors.remove(interceptor) }
+    }
+
+    private suspend fun EventMessage<*>.processEventInterceptors(): EventMessage<*> {
+        return dispatchInterceptors.fold(this) { acc, interceptor -> interceptor.intercept(acc) }
+    }
+
+    private fun EventMessage<*>.publish() {
+        eventBus.publish(this)
+    }
+
+    /**
+     * Builder class to instantiate [DefaultSuspendingEventGateway].
+     *
+     * The `dispatchInterceptors` are defaulted to an empty list.
+     * The [EventBus] is a **hard requirement**
+     *
+     */
+    class Builder {
+        lateinit var eventBus: EventBus
+        val dispatchInterceptors: List<SuspendingMessageDispatchInterceptor<EventMessage<*>>> = CopyOnWriteArrayList()
+
+        /**
+         * Validate whether the fields contained in this Builder as set accordingly.
+         *
+         * @throws AxonConfigurationException if one field is asserted to be incorrect according to the Builder's
+         * specifications
+         */
+        fun validate() = apply {
+            BuilderUtils.assertThat(this::eventBus.isInitialized, { it == true }, "The EventBus is a hard requirement and should be provided")
+        }
+    }
+}

kotlin/src/main/kotlin/org/axonframework/extensions/kotlin/eventhandling/gateway/SuspendingEventGateway.kt

@@ -0,0 +1,45 @@
+package org.axonframework.extensions.kotlin.eventhandling.gateway
+
+import org.axonframework.eventhandling.EventMessage
+import org.axonframework.extensions.kotlin.messaging.SuspendingMessageDispatchInterceptor
+import org.axonframework.extensions.kotlin.messaging.SuspendingMessageDispatchInterceptorSupport
+import java.util.*
+
+/**
+ * Variation of the [EventGateway], wrapping a [EventBus] for a friendlier API.
+ * Provides support for Kotlin coroutines.
+ *
+ * @author Joel Feinstein
+ * @since 0.2.0
+ */
+interface SuspendingEventGateway : SuspendingMessageDispatchInterceptorSupport<EventMessage<*>> {
+    /**
+     * Publishes given `events`.
+     *
+     * Given `events` are wrapped as payloads of a [EventMessage] that are eventually published on the
+     * [EventBus], unless `event` already implements [Message]. In that case, a `EventMessage`
+     * is constructed from that message's payload and [org.axonframework.messaging.MetaData].
+     *
+     * @param events events to be published
+     * @return events that were published. DO NOTE: if there were some [SuspendingMessageDispatchInterceptor]s
+     * registered to this `gateway`, they will be processed first, before returning events to the caller. The
+     * order of returned events is the same as one provided as the input parameter.
+     */
+    suspend fun publish(vararg events: Any?): List<EventMessage<*>> {
+        return publish(events.toList())
+    }
+
+    /**
+     * Publishes the given `events`.
+     *
+     * Given `events` are wrapped as payloads of a [EventMessage] that are eventually published on the
+     * [EventBus], unless `event` already implements [Message]. In that case, a `EventMessage`
+     * is constructed from that message's payload and [org.axonframework.messaging.MetaData].
+     *
+     * @param events the list of events to be published
+     * @return events that were published. DO NOTE: if there were some [SuspendingMessageDispatchInterceptor]s
+     * registered to this `gateway`, they will be processed first, before returning events to the caller. The
+     * order of returned events is the same as one provided as the input parameter.
+     */
+    suspend fun publish(events: Iterable<*>): List<EventMessage<*>>
+}

kotlin/src/test/kotlin/org/axonframework/extensions/kotlin/eventhandling/gateway/DefaultSuspendingEventGatewayTests.kt

@@ -0,0 +1,61 @@
+package org.axonframework.extensions.kotlin.eventhandling.gateway
+
+import io.mockk.every
+import io.mockk.justRun
+import io.mockk.mockk
+import io.mockk.verify
+import kotlinx.coroutines.runBlocking
+import org.axonframework.eventhandling.EventBus
+import org.axonframework.eventhandling.EventMessage
+import org.axonframework.eventhandling.GenericEventMessage
+import org.axonframework.extensions.kotlin.messaging.andMetaData
+import org.axonframework.extensions.kotlin.messaging.registerDispatchInterceptor
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.assertThrows
+import kotlin.test.assertEquals
+
+class DefaultSuspendingEventGatewayTests {
+    private lateinit var eventBus: EventBus
+    private lateinit var gateway: SuspendingEventGateway
+
+    @BeforeEach
+    fun setUp() {
+        eventBus = mockk {
+            justRun { publish(any<EventMessage<*>>()) }
+        }
+
+        gateway = DefaultSuspendingEventGateway {
+            eventBus = this@DefaultSuspendingEventGatewayTests.eventBus
+        }
+    }
+
+    @Test
+    fun testPublish(): Unit = runBlocking {
+        gateway.publish("event", "event")
+        verify(exactly = 2) { eventBus.publish(any<EventMessage<*>>()) }
+    }
+
+    @Test
+    fun testPublishWithError(): Unit = runBlocking {
+        every { eventBus.publish(any<EventMessage<*>>()) } throws RuntimeException("oops")
+        assertThrows<RuntimeException> {
+            gateway.publish("event", "event")
+        }
+    }
+
+    @Test
+    fun testDispatchInterceptor(): Unit = runBlocking {
+        gateway.registerDispatchInterceptor {
+            GenericEventMessage.asEventMessage<Any?>(it.payload).andMetaData("key" to "value")
+        }
+        assertEquals("value", gateway.publish("event").first().metaData["key"])
+    }
+
+    @Test
+    fun testPublishOrder(): Unit = runBlocking {
+        val event1 = GenericEventMessage.asEventMessage<Any>("event1")
+        val event2 = GenericEventMessage.asEventMessage<Any>("event2")
+        assertEquals(listOf(event1, event2), gateway.publish(event1, event2))
+    }
+}