Add BATCH support

Closes #11

CHANGELOG

@@ -7,6 +7,9 @@ vNEXT (in development)
  * Events now have a `metadata` property instead of a `time` (and time is available in metadata)
    * IrcEvent.time is now deprecated but will remain until after v1.0.0.
    * Metadata now contains the event's batch ID, if any.
+ * Added support for batches
+   * All events in a batch are buffered until the batch is finished
+   * The events are then published together in a single `BatchReceived` event
  * (Internal) Introduced event mutators
    * Event mutators are now responsible for handling changing events in response to state
      e.g. ChannelFanOutMutator creates Channel* events for global quits/nick changes/etc

src/main/kotlin/com/dmdirc/ktirc/IrcClientImpl.kt

@@ -1,7 +1,6 @@
 package com.dmdirc.ktirc
 
 import com.dmdirc.ktirc.events.*
-import com.dmdirc.ktirc.events.handlers.EventHandler
 import com.dmdirc.ktirc.events.handlers.eventHandlers
 import com.dmdirc.ktirc.events.mutators.eventMutators
 import com.dmdirc.ktirc.io.KtorLineBufferedSocket
@@ -42,7 +41,7 @@ internal class IrcClientImpl(private val config: IrcClientConfig) : IrcClient, C
     override val channelState = ChannelStateMap { caseMapping }
     override val userState = UserState { caseMapping }
 
-    private val messageHandler = MessageHandler(messageProcessors, eventMutators, eventHandlers.toMutableList())
+    private val messageHandler = MessageHandler(messageProcessors, eventMutators, eventHandlers)
 
     private val parser = MessageParser()
     private var socket: LineBufferedSocket? = null
@@ -85,15 +84,9 @@ internal class IrcClientImpl(private val config: IrcClientConfig) : IrcClient, C
         socket?.disconnect()
     }
 
-    override fun onEvent(handler: (IrcEvent) -> Unit) {
-        messageHandler.handlers.add(object : EventHandler {
-            override fun processEvent(client: IrcClient, event: IrcEvent) {
-                handler(event)
-            }
-        })
-    }
+    override fun onEvent(handler: (IrcEvent) -> Unit) = messageHandler.addEmitter(handler)
 
-    private fun emitEvent(event: IrcEvent) = messageHandler.emitEvent(this, event)
+    private fun emitEvent(event: IrcEvent) = messageHandler.handleEvent(this, event)
     private fun sendPasswordIfPresent() = config.server.password?.let(this::sendPassword)
 
     internal fun reset() {

src/main/kotlin/com/dmdirc/ktirc/events/Events.kt

@@ -152,3 +152,6 @@ class BatchStarted(metadata: EventMetadata, val referenceId: String, val batchTy
 
 /** Indicates a batch of messages has finished. */
 class BatchFinished(metadata: EventMetadata, val referenceId: String) : IrcEvent(metadata)
+
+/** A batch of events that should be handled together. */
+class BatchReceived(metadata: EventMetadata, val type: String, val params: Array<String>, val events: List<IrcEvent>) : IrcEvent(metadata)

src/main/kotlin/com/dmdirc/ktirc/events/mutators/BatchMutator.kt

@@ -0,0 +1,49 @@
+package com.dmdirc.ktirc.events.mutators
+
+import com.dmdirc.ktirc.IrcClient
+import com.dmdirc.ktirc.events.BatchFinished
+import com.dmdirc.ktirc.events.BatchReceived
+import com.dmdirc.ktirc.events.BatchStarted
+import com.dmdirc.ktirc.events.IrcEvent
+import com.dmdirc.ktirc.io.MessageEmitter
+import com.dmdirc.ktirc.model.Batch
+
+internal class BatchMutator : EventMutator {
+
+    override fun mutateEvent(client: IrcClient, messageEmitter: MessageEmitter, event: IrcEvent): List<IrcEvent> {
+        when {
+            event is BatchStarted -> startBatch(client, event)
+            event is BatchFinished -> return finishBatch(client, event)
+            event.metadata.batchId != null -> addToBatch(client, messageEmitter, event)
+            else -> return listOf(event)
+        }
+        return emptyList()
+    }
+
+    private fun startBatch(client: IrcClient, event: BatchStarted) {
+        client.serverState.batches[event.referenceId] =
+                Batch(event.batchType, event.params.asList(), event.metadata.batchId, mutableListOf(event))
+    }
+
+    private fun finishBatch(client: IrcClient, event: BatchFinished): List<IrcEvent> {
+        client.serverState.batches.remove(event.referenceId)?.let {
+            it.events += event
+            val batch = BatchReceived(it.events[0].metadata, it.type, it.arguments.toTypedArray(), it.events)
+            if (it.parent == null) {
+                return listOf(batch)
+            } else {
+                client.serverState.batches[it.parent]?.events?.add(batch)
+            }
+        }
+
+        return emptyList()
+    }
+
+    private fun addToBatch(client: IrcClient, messageEmitter: MessageEmitter, event: IrcEvent) {
+        client.serverState.batches[event.metadata.batchId]?.let {
+            it.events += event
+            messageEmitter.handleEvent(client, event, true)
+        }
+    }
+
+}

src/main/kotlin/com/dmdirc/ktirc/events/mutators/ChannelFanOutMutator.kt

@@ -2,13 +2,14 @@ package com.dmdirc.ktirc.events.mutators
 
 import com.dmdirc.ktirc.IrcClient
 import com.dmdirc.ktirc.events.*
+import com.dmdirc.ktirc.io.MessageEmitter
 
 /**
  * "Fans out" global events such as quits and nick changes to each channel a user is in.
  */
 internal class ChannelFanOutMutator : EventMutator {
 
-    override fun mutateEvent(client: IrcClient, event: IrcEvent) = sequence<IrcEvent> {
+    override fun mutateEvent(client: IrcClient, messageEmitter: MessageEmitter, event: IrcEvent) = sequence<IrcEvent> {
         yield(event)
         when (event) {
             is UserQuit -> handleQuit(client, event)

src/main/kotlin/com/dmdirc/ktirc/events/mutators/EventMutator.kt

@@ -2,15 +2,17 @@ package com.dmdirc.ktirc.events.mutators
 
 import com.dmdirc.ktirc.IrcClient
 import com.dmdirc.ktirc.events.IrcEvent
+import com.dmdirc.ktirc.io.MessageEmitter
 
 @FunctionalInterface
 internal interface EventMutator {
 
-    fun mutateEvent(client: IrcClient, event: IrcEvent): List<IrcEvent>
+    fun mutateEvent(client: IrcClient, messageEmitter: MessageEmitter, event: IrcEvent): List<IrcEvent>
 
 }
 
 internal val eventMutators = listOf(
         ServerReadyMutator(),
-        ChannelFanOutMutator()
+        ChannelFanOutMutator(),
+        BatchMutator()
 )

src/main/kotlin/com/dmdirc/ktirc/events/mutators/ServerReadyMutator.kt

@@ -2,6 +2,7 @@ package com.dmdirc.ktirc.events.mutators
 
 import com.dmdirc.ktirc.IrcClient
 import com.dmdirc.ktirc.events.*
+import com.dmdirc.ktirc.io.MessageEmitter
 import com.dmdirc.ktirc.model.ServerStatus
 
 /**
@@ -24,7 +25,7 @@ internal class ServerReadyMutator : EventMutator {
             ServerCapabilitiesFinished::class
     )
 
-    override fun mutateEvent(client: IrcClient, event: IrcEvent): List<IrcEvent> = sequence {
+    override fun mutateEvent(client: IrcClient, messageEmitter: MessageEmitter, event: IrcEvent): List<IrcEvent> = sequence {
         if (client.serverState.receivedWelcome
                 && client.serverState.status == ServerStatus.Negotiating
                 && event::class !in excludedEvents) {

src/main/kotlin/com/dmdirc/ktirc/io/MessageHandler.kt

@@ -9,30 +9,50 @@ import com.dmdirc.ktirc.model.IrcMessage
 import com.dmdirc.ktirc.util.logger
 import kotlinx.coroutines.channels.ReceiveChannel
 
+internal interface MessageEmitter {
+
+    fun handleEvent(ircClient: IrcClient, ircEvent: IrcEvent, processOnly: Boolean = false) = handleEvents(ircClient, listOf(ircEvent), processOnly)
+
+    fun handleEvents(ircClient: IrcClient, ircEvents: List<IrcEvent>, processOnly: Boolean = false)
+
+}
+
 internal class MessageHandler(
         private val processors: List<MessageProcessor>,
         private val mutators: List<EventMutator>,
-        val handlers: MutableList<EventHandler>) {
+        private val handlers: List<EventHandler>) : MessageEmitter {
 
     private val log by logger()
 
+    private val emitters = mutableListOf<(IrcEvent) -> Unit>()
+
     suspend fun processMessages(ircClient: IrcClient, messages: ReceiveChannel<IrcMessage>) {
         for (message in messages) {
-            emitEvents(ircClient, message.toEvents())
+            handleEvents(ircClient, message.toEvents())
         }
     }
 
-    fun emitEvent(ircClient: IrcClient, ircEvent: IrcEvent) = emitEvents(ircClient, listOf(ircEvent))
-
-    fun emitEvents(ircClient: IrcClient, ircEvents: List<IrcEvent>) {
-        mutators.fold(ircEvents) { events, mutator ->
-            events.flatMap { mutator.mutateEvent(ircClient, it) }
-        }.forEach { event ->
-            log.fine { "Dispatching event of type ${event::class}" }
-            handlers.forEach { it.processEvent(ircClient, event) }
+    override fun handleEvents(ircClient: IrcClient, ircEvents: List<IrcEvent>, processOnly: Boolean) {
+        val events = if (processOnly) ircEvents else ircEvents.mutate(ircClient)
+        events.forEach { event ->
+            event.process(ircClient)
+            if (!processOnly) {
+                log.fine { "Dispatching event of type ${event::class}" }
+                emitters.forEach { it(event) }
+            }
         }
     }
 
+    fun addEmitter(emitter: (IrcEvent) -> Unit) {
+        emitters.add(emitter)
+    }
+
+    private fun IrcEvent.process(ircClient: IrcClient) = handlers.forEach { it.processEvent(ircClient, this) }
+
+    private fun List<IrcEvent>.mutate(ircClient: IrcClient) = mutators.fold(this) { events, mutator ->
+        events.flatMap { mutator.mutateEvent(ircClient, this@MessageHandler, it) }
+    }
+
     private fun IrcMessage.toEvents() = this.getProcessor()?.process(this) ?: emptyList()
     private fun IrcMessage.getProcessor() = processors.firstOrNull { it.commands.contains(command) } ?: run {
         log.warning { "No processor found for $command" }

src/main/kotlin/com/dmdirc/ktirc/messages/BatchProcessor.kt

@@ -1,5 +1,7 @@
 package com.dmdirc.ktirc.messages
 
+import com.dmdirc.ktirc.events.BatchFinished
+import com.dmdirc.ktirc.events.BatchStarted
 import com.dmdirc.ktirc.events.IrcEvent
 import com.dmdirc.ktirc.model.IrcMessage
 
@@ -8,7 +10,13 @@ internal class BatchProcessor : MessageProcessor {
     override val commands = arrayOf("BATCH")
 
     override fun process(message: IrcMessage): List<IrcEvent> {
-        TODO("not implemented") //To change body of created functions use File | Settings | File Templates.
+        val args = message.params.map { String(it) }
+        val id = args[0]
+        return when (id[0]) {
+            '+' -> listOf(BatchStarted(message.metadata, id.substring(1), args[1], args.subList(2, args.size).toTypedArray()))
+            '-' -> listOf(BatchFinished(message.metadata, id.substring(1)))
+            else -> emptyList()
+        }
     }
 
 }

src/main/kotlin/com/dmdirc/ktirc/messages/MessageProcessor.kt

@@ -20,6 +20,7 @@ internal interface MessageProcessor {
 internal val messageProcessors = listOf(
         AccountProcessor(),
         AuthenticationProcessor(),
+        BatchProcessor(),
         CapabilityProcessor(),
         ISupportProcessor(),
         JoinProcessor(),

src/main/kotlin/com/dmdirc/ktirc/model/ServerState.kt

@@ -1,6 +1,7 @@
 package com.dmdirc.ktirc.model
 
 import com.dmdirc.ktirc.SaslConfig
+import com.dmdirc.ktirc.events.IrcEvent
 import com.dmdirc.ktirc.io.CaseMapping
 import com.dmdirc.ktirc.util.logger
 import kotlin.reflect.KClass
@@ -63,6 +64,11 @@ class ServerState internal constructor(
     val channelTypes
         get() = features[ServerFeature.ChannelTypes] ?: throw IllegalStateException("lost channel types")
 
+    /**
+     * Batches that are currently in progress.
+     */
+    internal val batches = mutableMapOf<String, Batch>()
+
     /**
      * Determines if the given mode is one applied to a user of a channel, such as 'o' for operator.
      */
@@ -92,6 +98,7 @@ class ServerState internal constructor(
         features.clear()
         capabilities.reset()
         sasl.reset()
+        batches.clear()
     }
 
 }
@@ -176,6 +183,11 @@ enum class ServerStatus {
     Ready,
 }
 
+/**
+ * Represents an in-progress batch.
+ */
+internal data class Batch(val type: String, val arguments: List<String>, val parent: String? = null, val events: MutableList<IrcEvent> = mutableListOf())
+
 internal val serverFeatures: Map<String, ServerFeature<*>> by lazy {
     ServerFeature::class.nestedClasses.map { it.objectInstance as ServerFeature<*> }.associateBy { it.name }
 }

src/test/kotlin/com/dmdirc/ktirc/events/mutators/BatchMutatorTest.kt

@@ -0,0 +1,121 @@
+package com.dmdirc.ktirc.events.mutators
+
+import com.dmdirc.ktirc.IrcClient
+import com.dmdirc.ktirc.TestConstants
+import com.dmdirc.ktirc.events.*
+import com.dmdirc.ktirc.io.MessageEmitter
+import com.dmdirc.ktirc.model.Batch
+import com.dmdirc.ktirc.model.ServerState
+import com.dmdirc.ktirc.model.User
+import com.nhaarman.mockitokotlin2.*
+import org.junit.jupiter.api.Assertions.*
+import org.junit.jupiter.api.Test
+
+internal class BatchMutatorTest {
+
+    private val mutator = BatchMutator()
+
+    private val serverState = ServerState("", "")
+    private val messageEmitter = mock<MessageEmitter>()
+    private val ircClient = mock<IrcClient> {
+        on { serverState } doReturn serverState
+    }
+
+    @Test
+    fun `returns non-batched events unmodified`() {
+        val event = ServerWelcome(EventMetadata(TestConstants.time), "the.gibson", "acidBurn")
+        val events = mutator.mutateEvent(ircClient, messageEmitter, event)
+
+        assertEquals(1, events.size)
+        assertSame(event, events[0])
+    }
+
+    @Test
+    fun `starts a batch when BatchStarted event is received`() {
+        val event = BatchStarted(EventMetadata(TestConstants.time), "abcdef", "netsplit", arrayOf("foo", "bar"))
+        val events = mutator.mutateEvent(ircClient, messageEmitter, event)
+
+        assertTrue(events.isEmpty())
+        assertNotNull(serverState.batches["abcdef"])
+        serverState.batches["abcdef"]?.let {
+            assertEquals(listOf("foo", "bar"), it.arguments)
+            assertEquals("netsplit", it.type)
+            assertEquals(listOf(event), it.events)
+            assertNull(it.parent)
+        }
+    }
+
+    @Test
+    fun `adds to batch when event has a batch ID`() {
+        serverState.batches["abcdef"] = Batch("netsplit", emptyList())
+
+        val event = UserNickChanged(EventMetadata(TestConstants.time, "abcdef"), User("zeroCool"), "crashOverride")
+        mutator.mutateEvent(ircClient, messageEmitter, event)
+
+        assertEquals(listOf(event), serverState.batches["abcdef"]!!.events)
+    }
+
+    @Test
+    fun `suppresses event when it has a batch ID`() {
+        serverState.batches["abcdef"] = Batch("netsplit", emptyList())
+
+        val event = UserNickChanged(EventMetadata(TestConstants.time, "abcdef"), User("zeroCool"), "crashOverride")
+        val events = mutator.mutateEvent(ircClient, messageEmitter, event)
+
+        assertTrue(events.isEmpty())
+    }
+
+    @Test
+    fun `passes event for processing only when it has a batch ID`() {
+        serverState.batches["abcdef"] = Batch("netsplit", emptyList())
+
+        val event = UserNickChanged(EventMetadata(TestConstants.time, "abcdef"), User("zeroCool"), "crashOverride")
+        mutator.mutateEvent(ircClient, messageEmitter, event)
+
+        verify(messageEmitter).handleEvent(any(), same(event), eq(true))
+    }
+
+    @Test
+    fun `sends a batch when it finishes and the parent is null`() {
+        serverState.batches["abcdef"] = Batch("netsplit", listOf("p1", "p2"), events = mutableListOf(ServerConnected(EventMetadata(TestConstants.time, "abcdef"))))
+
+        val events = mutator.mutateEvent(ircClient, messageEmitter, BatchFinished(EventMetadata(TestConstants.time), "abcdef"))
+
+        assertEquals(1, events.size)
+        assertTrue(events[0] is BatchReceived)
+        val event = events[0] as BatchReceived
+        assertEquals("netsplit", event.type)
+        assertArrayEquals(arrayOf("p1", "p2"), event.params)
+        assertEquals(2, event.events.size)
+        assertTrue(event.events[0] is ServerConnected)
+        assertTrue(event.events[1] is BatchFinished)
+    }
+
+    @Test
+    fun `adds a batch to its parent when it finishes`() {
+        serverState.batches["12345"] = Batch("history", emptyList())
+        serverState.batches["abcdef"] = Batch("netsplit", emptyList(), "12345", mutableListOf(ServerConnected(EventMetadata(TestConstants.time, "abcdef"))))
+
+        val events = mutator.mutateEvent(ircClient, messageEmitter, BatchFinished(EventMetadata(TestConstants.time), "abcdef"))
+
+        assertEquals(0, events.size)
+
+        val parent = serverState.batches["12345"]?.events
+        assertEquals(1, parent?.size)
+
+        val event = parent?.get(0) as BatchReceived
+        assertEquals(2, event.events.size)
+        assertTrue(event.events[0] is ServerConnected)
+        assertTrue(event.events[1] is BatchFinished)
+    }
+
+    @Test
+    fun `deletes batch when it finishes`() {
+        serverState.batches["abcdef"] = Batch("netsplit", emptyList(), events = mutableListOf(ServerConnected(EventMetadata(TestConstants.time, "abcdef"))))
+
+        mutator.mutateEvent(ircClient, messageEmitter, BatchFinished(EventMetadata(TestConstants.time), "abcdef"))
+
+        assertNull(serverState.batches["abcdef"])
+    }
+
+}

src/test/kotlin/com/dmdirc/ktirc/events/mutators/ChannelFanOutMutatorTest.kt

@@ -5,6 +5,7 @@ import com.dmdirc.ktirc.IrcClient
 import com.dmdirc.ktirc.TestConstants
 import com.dmdirc.ktirc.events.*
 import com.dmdirc.ktirc.io.CaseMapping
+import com.dmdirc.ktirc.io.MessageEmitter
 import com.dmdirc.ktirc.model.*
 import com.nhaarman.mockitokotlin2.doReturn
 import com.nhaarman.mockitokotlin2.mock
@@ -17,6 +18,7 @@ internal class ChannelFanOutMutatorTest {
     private val channelStateMap = ChannelStateMap { CaseMapping.Rfc }
     private val serverState = ServerState("", "")
     private val behaviour = BehaviourConfig()
+    private val messageEmitter = mock<MessageEmitter>()
     private val ircClient = mock<IrcClient> {
         on { serverState } doReturn serverState
         on { channelState } doReturn channelStateMap
@@ -43,7 +45,7 @@ internal class ChannelFanOutMutatorTest {
         }
 
         val quitEvent = UserQuit(EventMetadata(TestConstants.time), User("zerocool", "dade", "root.localhost"), "Hack the planet!")
-        val events = mutator.mutateEvent(ircClient, quitEvent)
+        val events = mutator.mutateEvent(ircClient, messageEmitter, quitEvent)
 
         val names = mutableListOf<String>()
         Assertions.assertEquals(3, events.size)
@@ -79,7 +81,7 @@ internal class ChannelFanOutMutatorTest {
         }
 
         val nickEvent = UserNickChanged(EventMetadata(TestConstants.time), User("zerocool", "dade", "root.localhost"), "zer0c00l")
-        val events = mutator.mutateEvent(ircClient, nickEvent)
+        val events = mutator.mutateEvent(ircClient, messageEmitter, nickEvent)
 
         val names = mutableListOf<String>()
         Assertions.assertEquals(3, events.size)

src/test/kotlin/com/dmdirc/ktirc/events/mutators/ServerReadyMutatorTest.kt

@@ -3,6 +3,7 @@ package com.dmdirc.ktirc.events.mutators
 import com.dmdirc.ktirc.IrcClient
 import com.dmdirc.ktirc.TestConstants
 import com.dmdirc.ktirc.events.*
+import com.dmdirc.ktirc.io.MessageEmitter
 import com.dmdirc.ktirc.model.ServerState
 import com.dmdirc.ktirc.model.ServerStatus
 import com.dmdirc.ktirc.model.User
@@ -14,6 +15,7 @@ import org.junit.jupiter.api.Test
 internal class ServerReadyMutatorTest {
 
     private val serverState = ServerState("", "")
+    private val messageEmitter = mock<MessageEmitter>()
     private val ircClient = mock<IrcClient> {
         on { serverState } doReturn serverState
     }
@@ -33,11 +35,11 @@ internal class ServerReadyMutatorTest {
                 ServerCapabilitiesAcknowledged(EventMetadata(TestConstants.time), emptyMap()),
                 ServerCapabilitiesFinished(EventMetadata(TestConstants.time))
         ).forEach {
-            assertEquals(1, mutator.mutateEvent(ircClient, it).size)
+            assertEquals(1, mutator.mutateEvent(ircClient, messageEmitter, it).size)
         }
 
         val event = MessageReceived(EventMetadata(TestConstants.time), User("zeroCool"), "acidBurn", "Welcome!")
-        val events = mutator.mutateEvent(ircClient, event)
+        val events = mutator.mutateEvent(ircClient, messageEmitter, event)
         assertEquals(2, events.size)
         assertSame(event, events[1])
         assertTrue(events[0] is ServerReady)

src/test/kotlin/com/dmdirc/ktirc/io/MessageHandlerTest.kt

@@ -2,10 +2,7 @@ package com.dmdirc.ktirc.io
 
 import com.dmdirc.ktirc.IrcClient
 import com.dmdirc.ktirc.TestConstants
-import com.dmdirc.ktirc.events.EventMetadata
-import com.dmdirc.ktirc.events.ServerConnected
-import com.dmdirc.ktirc.events.ServerReady
-import com.dmdirc.ktirc.events.ServerWelcome
+import com.dmdirc.ktirc.events.*
 import com.dmdirc.ktirc.events.handlers.EventHandler
 import com.dmdirc.ktirc.events.mutators.EventMutator
 import com.dmdirc.ktirc.messages.MessageProcessor
@@ -83,31 +80,61 @@ internal class MessageHandlerTest {
     }
 
     @Test
-    fun `emits custom events to all handlers`() {
+    fun `sends custom events to all handlers`() {
         val eventHandler1 = mock<EventHandler>()
         val eventHandler2 = mock<EventHandler>()
         val handler = MessageHandler(emptyList(), emptyList(), mutableListOf(eventHandler1, eventHandler2))
-        handler.emitEvent(ircClient, ServerWelcome(EventMetadata(TestConstants.time), "the.gibson", "acidBurn"))
+        handler.handleEvent(ircClient, ServerWelcome(EventMetadata(TestConstants.time), "the.gibson", "acidBurn"))
 
         verify(eventHandler1).processEvent(same(ircClient), isA<ServerWelcome>())
         verify(eventHandler2).processEvent(same(ircClient), isA<ServerWelcome>())
     }
 
+    @Test
+    fun `sends custom events to all emitters`() {
+        val handler = MessageHandler(emptyList(), emptyList(), emptyList())
+        val emitter1 = mock<(IrcEvent) -> Unit>()
+        val emitter2 = mock<(IrcEvent) -> Unit>()
+        handler.addEmitter(emitter1)
+        handler.addEmitter(emitter2)
+        handler.handleEvent(ircClient, ServerWelcome(EventMetadata(TestConstants.time), "the.gibson", "acidBurn"))
+
+        verify(emitter1).invoke(isA<ServerWelcome>())
+        verify(emitter2).invoke(isA<ServerWelcome>())
+    }
+
+    @Test
+    fun `sends events to handlers but not mutators or emitters if process only is true`() {
+        val mutator = mock<EventMutator>()
+        val eventHandler1 = mock<EventHandler>()
+        val eventHandler2 = mock<EventHandler>()
+        val handler = MessageHandler(emptyList(), listOf(mutator), listOf(eventHandler1, eventHandler2))
+        val emitter = mock<(IrcEvent) -> Unit>()
+        handler.addEmitter(emitter)
+
+        handler.handleEvent(ircClient, ServerWelcome(EventMetadata(TestConstants.time), "the.gibson", "acidBurn"), true)
+
+        verify(eventHandler1).processEvent(same(ircClient), isA<ServerWelcome>())
+        verify(eventHandler2).processEvent(same(ircClient), isA<ServerWelcome>())
+        verify(emitter, never()).invoke(any())
+        verify(mutator, never()).mutateEvent(any(), any(), any())
+    }
+
     @Test
     fun `mutates events in order`() {
         val eventMutator1 = mock<EventMutator> {
-            on { mutateEvent(any(), isA<ServerWelcome>()) } doReturn listOf(ServerReady(EventMetadata(TestConstants.time)))
+            on { mutateEvent(any(), any(), isA<ServerWelcome>()) } doReturn listOf(ServerReady(EventMetadata(TestConstants.time)))
         }
         val eventMutator2 = mock<EventMutator> {
-            on { mutateEvent(any(), isA<ServerReady>()) } doReturn listOf(ServerConnected(EventMetadata(TestConstants.time)))
+            on { mutateEvent(any(), any(), isA<ServerReady>()) } doReturn listOf(ServerConnected(EventMetadata(TestConstants.time)))
         }
         val eventHandler = mock<EventHandler>()
 
         val handler = MessageHandler(emptyList(), listOf(eventMutator1, eventMutator2), mutableListOf(eventHandler))
-        handler.emitEvent(ircClient, ServerWelcome(EventMetadata(TestConstants.time), "the.gibson", "acidBurn"))
+        handler.handleEvent(ircClient, ServerWelcome(EventMetadata(TestConstants.time), "the.gibson", "acidBurn"))
 
-        verify(eventMutator1).mutateEvent(same(ircClient), isA<ServerWelcome>())
-        verify(eventMutator2).mutateEvent(same(ircClient), isA<ServerReady>())
+        verify(eventMutator1).mutateEvent(same(ircClient), same(handler), isA<ServerWelcome>())
+        verify(eventMutator2).mutateEvent(same(ircClient), same(handler), isA<ServerReady>())
         verify(eventHandler).processEvent(same(ircClient), isA<ServerConnected>())
         verifyNoMoreInteractions(eventHandler)
     }
@@ -115,23 +142,23 @@ internal class MessageHandlerTest {
     @Test
     fun `allows mutators to fan out events`() {
         val eventMutator1 = mock<EventMutator> {
-            on { mutateEvent(any(), isA<ServerWelcome>()) } doReturn listOf(
+            on { mutateEvent(any(), any(), isA<ServerWelcome>()) } doReturn listOf(
                     ServerReady(EventMetadata(TestConstants.time)),
                     ServerConnected(EventMetadata(TestConstants.time))
             )
         }
         val eventMutator2 = mock<EventMutator> {
-            on { mutateEvent(any(), isA<ServerReady>()) } doReturn listOf(ServerReady(EventMetadata(TestConstants.time)))
-            on { mutateEvent(any(), isA<ServerConnected>()) } doReturn listOf(ServerConnected(EventMetadata(TestConstants.time)))
+            on { mutateEvent(any(), any(), isA<ServerReady>()) } doReturn listOf(ServerReady(EventMetadata(TestConstants.time)))
+            on { mutateEvent(any(), any(), isA<ServerConnected>()) } doReturn listOf(ServerConnected(EventMetadata(TestConstants.time)))
         }
         val eventHandler = mock<EventHandler>()
 
         val handler = MessageHandler(emptyList(), listOf(eventMutator1, eventMutator2), mutableListOf(eventHandler))
-        handler.emitEvent(ircClient, ServerWelcome(EventMetadata(TestConstants.time), "the.gibson", "acidBurn"))
+        handler.handleEvent(ircClient, ServerWelcome(EventMetadata(TestConstants.time), "the.gibson", "acidBurn"))
 
-        with (inOrder(eventMutator2, eventHandler)) {
-            verify(eventMutator2).mutateEvent(same(ircClient), isA<ServerReady>())
-            verify(eventMutator2).mutateEvent(same(ircClient), isA<ServerConnected>())
+        with(inOrder(eventMutator2, eventHandler)) {
+            verify(eventMutator2).mutateEvent(same(ircClient), same(handler), isA<ServerReady>())
+            verify(eventMutator2).mutateEvent(same(ircClient), same(handler), isA<ServerConnected>())
             verify(eventHandler).processEvent(same(ircClient), isA<ServerReady>())
             verify(eventHandler).processEvent(same(ircClient), isA<ServerConnected>())
         }
@@ -140,15 +167,15 @@ internal class MessageHandlerTest {
     @Test
     fun `allows mutators to suppress events`() {
         val eventMutator1 = mock<EventMutator> {
-            on { mutateEvent(any(), isA<ServerWelcome>()) } doReturn emptyList()
+            on { mutateEvent(any(), any(), isA<ServerWelcome>()) } doReturn emptyList()
         }
         val eventMutator2 = mock<EventMutator>()
         val eventHandler = mock<EventHandler>()
 
         val handler = MessageHandler(emptyList(), listOf(eventMutator1, eventMutator2), mutableListOf(eventHandler))
-        handler.emitEvent(ircClient, ServerWelcome(EventMetadata(TestConstants.time), "the.gibson", "acidBurn"))
+        handler.handleEvent(ircClient, ServerWelcome(EventMetadata(TestConstants.time), "the.gibson", "acidBurn"))
 
-        verify(eventMutator2, never()).mutateEvent(any(), any())
+        verify(eventMutator2, never()).mutateEvent(any(), same(handler), any())
         verify(eventHandler, never()).processEvent(any(), any())
     }
 

src/test/kotlin/com/dmdirc/ktirc/messages/BatchProcessorTest.kt

@@ -0,0 +1,52 @@
+package com.dmdirc.ktirc.messages
+
+import com.dmdirc.ktirc.TestConstants
+import com.dmdirc.ktirc.events.BatchFinished
+import com.dmdirc.ktirc.events.BatchStarted
+import com.dmdirc.ktirc.model.IrcMessage
+import com.dmdirc.ktirc.params
+import com.dmdirc.ktirc.util.currentTimeProvider
+import org.junit.jupiter.api.Assertions.assertArrayEquals
+import org.junit.jupiter.api.Assertions.assertEquals
+import org.junit.jupiter.api.BeforeEach
+import org.junit.jupiter.api.Test
+
+internal class BatchProcessorTest {
+
+    private var processor = BatchProcessor()
+
+    @BeforeEach
+    fun setUp() {
+        currentTimeProvider = { TestConstants.time }
+    }
+
+    @Test
+    fun `raises batch finished event when reference starts with a -`() {
+        val events = processor.process(IrcMessage(emptyMap(), null, "BATCH", params("-mybatch")))
+
+        assertEquals(1, events.size)
+        val event = events[0] as BatchFinished
+        assertEquals(TestConstants.time, event.metadata.time)
+        assertEquals("mybatch", event.referenceId)
+    }
+
+    @Test
+    fun `raises batch started event when reference starts with a +`() {
+        val events = processor.process(IrcMessage(emptyMap(), null, "BATCH", params("+mybatch", "mytype", "arg1", "arg2")))
+
+        assertEquals(1, events.size)
+        val event = events[0] as BatchStarted
+        assertEquals(TestConstants.time, event.metadata.time)
+        assertEquals("mybatch", event.referenceId)
+        assertEquals("mytype", event.batchType)
+        assertArrayEquals(arrayOf("arg1", "arg2"), event.params)
+    }
+
+    @Test
+    fun `ignores batches with bad reference ids`() {
+        val events = processor.process(IrcMessage(emptyMap(), null, "BATCH", params("~mybatch")))
+
+        assertEquals(0, events.size)
+    }
+
+}

src/test/kotlin/com/dmdirc/ktirc/model/ServerStateTest.kt

@@ -67,6 +67,7 @@ internal class ServerStateTest {
         features[ServerFeature.Network] = "gibson"
         capabilities.advertisedCapabilities[Capability.SaslAuthentication] = "sure"
         sasl.saslBuffer = "in progress"
+        batches["batch"] = Batch("type", emptyList())
 
         reset()
 
@@ -77,6 +78,7 @@ internal class ServerStateTest {
         assertTrue(features.isEmpty())
         assertTrue(capabilities.advertisedCapabilities.isEmpty())
         assertEquals("", sasl.saslBuffer)
+        assertTrue(batches.isEmpty())
     }
 
 }