*DO NOT MERGE* Topology changes - One consumer for multiple topics

applications/examples/sandbox-app/src/main/kotlin/net/corda/example/vnode/CordaVNode.kt

@@ -19,7 +19,7 @@ import net.corda.libs.packaging.core.CpkMetadata
 import net.corda.messaging.api.records.Record
 import net.corda.osgi.api.Application
 import net.corda.osgi.api.Shutdown
-import net.corda.schema.Schemas.Flow.FLOW_EVENT_TOPIC
+import net.corda.schema.Schemas.Flow.FLOW_START
 import net.corda.schema.configuration.ConfigKeys.FLOW_CONFIG
 import net.corda.schema.configuration.FlowConfig.PROCESSING_FLOW_CLEANUP_TIME
 import net.corda.schema.configuration.FlowConfig.PROCESSING_MAX_FLOW_SLEEP_DURATION
@@ -172,11 +172,11 @@ class CordaVNode @Activate constructor(
 
             val rpcStartFlow = createRPCStartFlow(clientId, vnodeInfo.toAvro())
             val flowId = generateRandomId()
-            val record = Record(FLOW_EVENT_TOPIC, flowId, FlowEvent(flowId, rpcStartFlow))
+            val record = Record(FLOW_START, flowId, FlowEvent(flowId, rpcStartFlow))
             flowEventProcessorFactory.create(mapOf(FLOW_CONFIG to smartConfig)).apply {
                 val result = onNext(null, record)
                 result.responseEvents.singleOrNull { evt ->
-                    evt.topic == FLOW_EVENT_TOPIC
+                    evt.topic == FLOW_START
                 }?.also { evt ->
                     @Suppress("unchecked_cast")
                     onNext(result.updatedState, evt as Record<String, FlowEvent>)

components/crypto/crypto-service-impl/src/test/kotlin/net/corda/crypto/service/impl/bus/CryptoFlowOpsBusProcessorTests.kt

@@ -257,7 +257,7 @@ import kotlin.test.assertTrue
                  )
              ).thenReturn(
                  Record(
-                     Schemas.Flow.FLOW_EVENT_TOPIC,
+                     Schemas.Flow.FLOW_SESSION,
                      flowExternalEventContexts.get(it).flowId,
                      FlowEvent()
                  )
@@ -269,7 +269,7 @@ import kotlin.test.assertTrue
                  )
              ).thenReturn(
                  Record(
-                     Schemas.Flow.FLOW_EVENT_TOPIC,
+                     Schemas.Flow.FLOW_SESSION,
                      flowExternalEventContexts.get(it).flowId,
                      FlowEvent()
                  )
@@ -281,7 +281,7 @@ import kotlin.test.assertTrue
                  )
              ).thenReturn(
                  Record(
-                     Schemas.Flow.FLOW_EVENT_TOPIC,
+                     Schemas.Flow.FLOW_SESSION,
                      flowExternalEventContexts.get(it).flowId,
                      FlowEvent()
                  )

components/flow/flow-mapper-impl/src/main/kotlin/net/corda/flow/mapper/impl/FlowMapperEventExecutorFactoryImpl.kt

@@ -17,7 +17,7 @@ import net.corda.flow.mapper.impl.executor.SessionEventExecutor
 import net.corda.flow.mapper.impl.executor.SessionInitProcessor
 import net.corda.flow.mapper.impl.executor.StartFlowExecutor
 import net.corda.libs.configuration.SmartConfig
-import net.corda.schema.Schemas.Flow.FLOW_EVENT_TOPIC
+import net.corda.schema.Schemas.Flow.FLOW_START
 import org.osgi.service.component.annotations.Activate
 import org.osgi.service.component.annotations.Component
 import org.osgi.service.component.annotations.Reference
@@ -66,7 +66,7 @@ class FlowMapperEventExecutorFactoryImpl @Activate constructor(
                 }
             }
 
-            is StartFlow -> StartFlowExecutor(eventKey, FLOW_EVENT_TOPIC, flowMapperEventPayload, state)
+            is StartFlow -> StartFlowExecutor(eventKey, FLOW_START, flowMapperEventPayload, state)
             is ExecuteCleanup -> ExecuteCleanupEventExecutor(eventKey)
             is ScheduleCleanup -> ScheduleCleanupEventExecutor(eventKey, flowMapperEventPayload, state)
 

components/flow/flow-mapper-impl/src/main/kotlin/net/corda/flow/mapper/impl/RecordFactoryImpl.kt

@@ -97,10 +97,10 @@ class RecordFactoryImpl @Activate constructor(
 
     private fun getSessionEventOutputTopic(sessionEvent: SessionEvent): String {
         return when (sessionEvent.messageDirection) {
-            MessageDirection.INBOUND -> Schemas.Flow.FLOW_EVENT_TOPIC
+            MessageDirection.INBOUND -> Schemas.Flow.FLOW_SESSION
             MessageDirection.OUTBOUND -> {
                 if (isLocalCluster(sessionEvent)) {
-                    Schemas.Flow.FLOW_MAPPER_EVENT_TOPIC
+                    Schemas.Flow.FLOW_MAPPER_SESSION_IN
                 } else {
                     Schemas.P2P.P2P_OUT_TOPIC
                 }
@@ -120,8 +120,8 @@ class RecordFactoryImpl @Activate constructor(
     ) : Record<*, *> {
         val outputTopic = getSessionEventOutputTopic(sourceEvent)
         val (newDirection, sessionId) = when (outputTopic) {
-            Schemas.Flow.FLOW_MAPPER_EVENT_TOPIC -> Pair(MessageDirection.INBOUND, toggleSessionId(sourceEvent.sessionId))
-            Schemas.Flow.FLOW_EVENT_TOPIC -> Pair(MessageDirection.INBOUND, sourceEvent.sessionId)
+            Schemas.Flow.FLOW_MAPPER_SESSION_IN -> Pair(MessageDirection.INBOUND, toggleSessionId(sourceEvent.sessionId))
+            Schemas.Flow.FLOW_SESSION -> Pair(MessageDirection.INBOUND, sourceEvent.sessionId)
             else -> Pair(MessageDirection.OUTBOUND, sourceEvent.sessionId)
         }
         val sequenceNumber = if (newPayload is SessionError) null else sourceEvent.sequenceNum
@@ -136,14 +136,14 @@ class RecordFactoryImpl @Activate constructor(
             sourceEvent.contextSessionProperties
         )
         return when (outputTopic) {
-            Schemas.Flow.FLOW_EVENT_TOPIC -> {
+            Schemas.Flow.FLOW_SESSION -> {
                 if (flowId == null) {
                     throw IllegalArgumentException("Flow ID is required to forward an event back to the flow event" +
                             "topic, but it was not provided.")
                 }
                 Record(outputTopic, flowId, FlowEvent(flowId, sessionEvent))
             }
-            Schemas.Flow.FLOW_MAPPER_EVENT_TOPIC -> {
+            Schemas.Flow.FLOW_MAPPER_SESSION_IN -> {
                 Record(outputTopic, sessionEvent.sessionId, FlowMapperEvent(sessionEvent))
             }
             Schemas.P2P.P2P_OUT_TOPIC -> {

components/flow/flow-mapper-impl/src/test/kotlin/net/corda/flow/mapper/impl/RecordFactoryImplTest.kt

@@ -27,7 +27,6 @@ import org.mockito.kotlin.anyOrNull
 import org.mockito.kotlin.mock
 import org.mockito.kotlin.verify
 import org.mockito.kotlin.whenever
-import java.lang.IllegalArgumentException
 import java.nio.ByteBuffer
 import java.time.Instant
 
@@ -98,7 +97,7 @@ internal class RecordFactoryImplTest {
             "my-flow-id"
         )
         assertThat(record).isNotNull
-        assertThat(record.topic).isEqualTo(Schemas.Flow.FLOW_MAPPER_EVENT_TOPIC)
+        assertThat(record.topic).isEqualTo(Schemas.Flow.FLOW_MAPPER_SESSION_IN)
         assertThat(record.value!!::class).isEqualTo(FlowMapperEvent::class)
         verify(locallyHostedIdentitiesServiceSameCluster).isHostedLocally(bobId.toCorda())
         val sessionOutput = (record.value as FlowMapperEvent).payload as SessionEvent
@@ -165,7 +164,7 @@ internal class RecordFactoryImplTest {
             flowConfig,
             FLOW_ID
         )
-        assertThat(record.topic).isEqualTo(Schemas.Flow.FLOW_EVENT_TOPIC)
+        assertThat(record.topic).isEqualTo(Schemas.Flow.FLOW_SESSION)
         assertThat(record.key).isEqualTo(FLOW_ID)
         assertThat(record.value!!::class.java).isEqualTo(FlowEvent::class.java)
         val sessionOutput = (record.value as FlowEvent).payload as SessionEvent
@@ -194,7 +193,7 @@ internal class RecordFactoryImplTest {
             FLOW_ID
         )
         assertThat(record).isNotNull
-        assertThat(record.topic).isEqualTo(Schemas.Flow.FLOW_MAPPER_EVENT_TOPIC)
+        assertThat(record.topic).isEqualTo(Schemas.Flow.FLOW_MAPPER_SESSION_IN)
         assertThat(record.value!!::class).isEqualTo(FlowMapperEvent::class)
         val sessionOutput = (record.value as FlowMapperEvent).payload as SessionEvent
         assertThat(sessionOutput.sessionId).isEqualTo("$SESSION_ID-INITIATED")
@@ -249,7 +248,7 @@ internal class RecordFactoryImplTest {
             flowConfig,
             FLOW_ID
         )
-        assertThat(record.topic).isEqualTo(Schemas.Flow.FLOW_EVENT_TOPIC)
+        assertThat(record.topic).isEqualTo(Schemas.Flow.FLOW_SESSION)
         assertThat(record.key).isEqualTo(FLOW_ID)
         assertThat(record.value!!::class.java).isEqualTo(FlowEvent::class.java)
         val sessionOutput = (record.value as FlowEvent).payload as SessionEvent
@@ -316,7 +315,7 @@ internal class RecordFactoryImplTest {
             timestamp,
             flowConfig,
         )
-        assertThat(record.topic).isEqualTo(Schemas.Flow.FLOW_MAPPER_EVENT_TOPIC)
+        assertThat(record.topic).isEqualTo(Schemas.Flow.FLOW_MAPPER_SESSION_IN)
         assertThat(record.key).isEqualTo("$SESSION_ID-INITIATED")
         assertThat(record.value!!::class).isEqualTo(FlowMapperEvent::class)
         val sessionOutput = (record.value as FlowMapperEvent).payload as SessionEvent

components/flow/flow-mapper-impl/src/test/kotlin/net/corda/flow/mapper/impl/executor/SessionInitProcessorTest.kt

@@ -29,7 +29,7 @@ class SessionInitProcessorTest {
             flowId: String
         ): Record<*, *> {
             return if (sourceEvent.messageDirection == MessageDirection.INBOUND) {
-                Record(Schemas.Flow.FLOW_EVENT_TOPIC, flowId, FlowEvent(flowId, sourceEvent))
+                Record(Schemas.Flow.FLOW_SESSION, flowId, FlowEvent(flowId, sourceEvent))
             } else {
                 Record(Schemas.P2P.P2P_OUT_TOPIC, "sessionId", "")
             }
@@ -79,7 +79,7 @@ class SessionInitProcessorTest {
 
         Assertions.assertThat(outboundEvents.size).isEqualTo(1)
         val outboundEvent = outboundEvents.first()
-        Assertions.assertThat(outboundEvent.topic).isEqualTo(Schemas.Flow.FLOW_EVENT_TOPIC)
+        Assertions.assertThat(outboundEvent.topic).isEqualTo(Schemas.Flow.FLOW_SESSION)
         Assertions.assertThat(outboundEvent.key::class).isEqualTo(String::class)
         Assertions.assertThat(outboundEvent.value!!::class).isEqualTo(FlowEvent::class)
         Assertions.assertThat(payload.sessionId).isEqualTo("sessionId-INITIATED")

components/flow/flow-mapper-service/src/integrationTest/kotlin/net/corda/session/mapper/service/integration/FlowMapperServiceIntegrationTest.kt

@@ -37,7 +37,9 @@ import net.corda.messaging.api.subscription.config.SubscriptionConfig
 import net.corda.messaging.api.subscription.factory.SubscriptionFactory
 import net.corda.schema.Schemas.Config.CONFIG_TOPIC
 import net.corda.schema.Schemas.Flow.FLOW_MAPPER_CLEANUP_TOPIC
-import net.corda.schema.Schemas.Flow.FLOW_MAPPER_EVENT_TOPIC
+import net.corda.schema.Schemas.Flow.FLOW_MAPPER_SESSION_IN
+import net.corda.schema.Schemas.Flow.FLOW_MAPPER_SESSION_OUT
+import net.corda.schema.Schemas.Flow.FLOW_MAPPER_START
 import net.corda.schema.Schemas.P2P.P2P_OUT_TOPIC
 import net.corda.schema.Schemas.ScheduledTask
 import net.corda.schema.configuration.BootConfig.BOOT_MAX_ALLOWED_MSG_SIZE
@@ -156,7 +158,7 @@ class FlowMapperServiceIntegrationTest {
 
         //send 2 session init
         val sessionDataAndInitEvent = Record<Any, Any>(
-            FLOW_MAPPER_EVENT_TOPIC, testSessionId, FlowMapperEvent(
+            FLOW_MAPPER_SESSION_OUT, testSessionId, FlowMapperEvent(
                 buildSessionEvent(
                     MessageDirection.OUTBOUND, testSessionId, 1, SessionData(ByteBuffer.wrap("bytes".toByteArray()), SessionInit(
                         testCpiId, testFlowId, emptyKeyValuePairList(), emptyKeyValuePairList()
@@ -181,7 +183,7 @@ class FlowMapperServiceIntegrationTest {
 
         //send data back
         val sessionDataEvent = Record<Any, Any>(
-            FLOW_MAPPER_EVENT_TOPIC, testSessionId, FlowMapperEvent(
+            FLOW_MAPPER_SESSION_IN, testSessionId, FlowMapperEvent(
                 buildSessionEvent(
                     MessageDirection.INBOUND,
                     testSessionId,
@@ -228,7 +230,7 @@ class FlowMapperServiceIntegrationTest {
         )
 
         val startRPCEvent = Record<Any, Any>(
-            FLOW_MAPPER_EVENT_TOPIC, testId, FlowMapperEvent(
+            FLOW_MAPPER_START, testId, FlowMapperEvent(
                 StartFlow(
                     context,
                     ""
@@ -284,7 +286,7 @@ class FlowMapperServiceIntegrationTest {
 
         //send data, no state
         val sessionDataEvent = Record<Any, Any>(
-            FLOW_MAPPER_EVENT_TOPIC, testId, FlowMapperEvent(
+            FLOW_MAPPER_SESSION_OUT, testId, FlowMapperEvent(
                 buildSessionEvent(
                     MessageDirection.OUTBOUND,
                     testId,
@@ -317,7 +319,7 @@ class FlowMapperServiceIntegrationTest {
 
         //send 2 session init, 1 is duplicate
         val sessionInitEvent = Record<Any, Any>(
-            FLOW_MAPPER_EVENT_TOPIC, testSessionId, FlowMapperEvent(
+            FLOW_MAPPER_SESSION_OUT, testSessionId, FlowMapperEvent(
                 buildSessionEvent(
                     MessageDirection.OUTBOUND, testSessionId, 1, SessionCounterpartyInfoRequest(SessionInit(
                         testCpiId, testFlowId, emptyKeyValuePairList(), emptyKeyValuePairList()
@@ -345,7 +347,7 @@ class FlowMapperServiceIntegrationTest {
 
         //send data back
         val sessionDataEvent = Record<Any, Any>(
-            FLOW_MAPPER_EVENT_TOPIC, testSessionId, FlowMapperEvent(
+            FLOW_MAPPER_SESSION_IN, testSessionId, FlowMapperEvent(
                 buildSessionEvent(
                     MessageDirection.INBOUND,
                     testSessionId,
@@ -380,7 +382,7 @@ class FlowMapperServiceIntegrationTest {
 
         //send data, no state
         val sessionDataEvent = Record<Any, Any>(
-            FLOW_MAPPER_EVENT_TOPIC, testId, FlowMapperEvent(
+            FLOW_MAPPER_SESSION_IN, testId, FlowMapperEvent(
                 buildSessionEvent(
                     MessageDirection.INBOUND,
                     testId,
@@ -396,7 +398,7 @@ class FlowMapperServiceIntegrationTest {
         val mapperLatch = CountDownLatch(2) // The initial message and the error back.
         val records = mutableListOf<SessionEvent>()
         val mapperSub = subscriptionFactory.createPubSubSubscription(
-            SubscriptionConfig("$testId-mapper", FLOW_MAPPER_EVENT_TOPIC),
+            SubscriptionConfig("$testId-mapper", FLOW_MAPPER_SESSION_IN),
             TestFlowMapperProcessor(mapperLatch, records),
             messagingConfig
         )

components/flow/flow-mapper-service/src/integrationTest/kotlin/net/corda/session/mapper/service/integration/TestFlowEventMediatorFactoryImpl.kt

@@ -14,8 +14,9 @@ import net.corda.messaging.api.mediator.factory.MessageRouterFactory
 import net.corda.messaging.api.mediator.factory.MessagingClientFactoryFactory
 import net.corda.messaging.api.mediator.factory.MultiSourceEventMediatorFactory
 import net.corda.messaging.api.processor.StateAndEventProcessor
-import net.corda.schema.Schemas.Flow.FLOW_EVENT_TOPIC
-import net.corda.schema.Schemas.Flow.FLOW_MAPPER_EVENT_TOPIC
+import net.corda.schema.Schemas.Flow.FLOW_MAPPER_SESSION_OUT
+import net.corda.schema.Schemas.Flow.FLOW_SESSION
+import net.corda.schema.Schemas.Flow.FLOW_START
 import net.corda.schema.configuration.MessagingConfig
 import org.osgi.service.component.annotations.Activate
 import org.osgi.service.component.annotations.Component
@@ -61,7 +62,12 @@ class TestFlowEventMediatorFactoryImpl @Activate constructor(
         .messagingConfig(messagingConfig)
         .consumerFactories(
             mediatorConsumerFactoryFactory.createMessageBusConsumerFactory(
-                FLOW_EVENT_TOPIC, CONSUMER_GROUP, messagingConfig
+                listOf(
+                    FLOW_START,
+                    FLOW_SESSION,
+                ),
+                CONSUMER_GROUP,
+                messagingConfig
             ),
         )
         .clientFactories(
@@ -81,7 +87,7 @@ class TestFlowEventMediatorFactoryImpl @Activate constructor(
 
         MessageRouter { message ->
             when (val event = message.payload) {
-                is FlowMapperEvent -> routeTo(messageBusClient, FLOW_MAPPER_EVENT_TOPIC)
+                is FlowMapperEvent -> routeTo(messageBusClient, FLOW_MAPPER_SESSION_OUT)
                 else -> {
                     val eventType = event?.let { it::class.java }
                     throw IllegalStateException("No route defined for event type [$eventType]")

components/flow/flow-mapper-service/src/main/kotlin/net/corda/session/mapper/messaging/mediator/FlowMapperEventMediatorFactoryImpl.kt

@@ -1,6 +1,7 @@
 package net.corda.session.mapper.messaging.mediator
 
 import net.corda.data.flow.event.FlowEvent
+import net.corda.data.flow.event.StartFlow
 import net.corda.data.flow.event.mapper.FlowMapperEvent
 import net.corda.data.flow.state.mapper.FlowMapperState
 import net.corda.data.p2p.app.AppMessage
@@ -15,8 +16,11 @@ import net.corda.messaging.api.mediator.factory.MessageRouterFactory
 import net.corda.messaging.api.mediator.factory.MessagingClientFactoryFactory
 import net.corda.messaging.api.mediator.factory.MultiSourceEventMediatorFactory
 import net.corda.messaging.api.processor.StateAndEventProcessor
-import net.corda.schema.Schemas.Flow.FLOW_EVENT_TOPIC
-import net.corda.schema.Schemas.Flow.FLOW_MAPPER_EVENT_TOPIC
+import net.corda.schema.Schemas.Flow.FLOW_MAPPER_SESSION_IN
+import net.corda.schema.Schemas.Flow.FLOW_MAPPER_SESSION_OUT
+import net.corda.schema.Schemas.Flow.FLOW_MAPPER_START
+import net.corda.schema.Schemas.Flow.FLOW_SESSION
+import net.corda.schema.Schemas.Flow.FLOW_START
 import net.corda.schema.Schemas.P2P.P2P_OUT_TOPIC
 import net.corda.schema.configuration.FlowConfig
 import net.corda.session.mapper.service.executor.FlowMapperMessageProcessor
@@ -63,7 +67,13 @@ class FlowMapperEventMediatorFactoryImpl @Activate constructor(
         .messagingConfig(messagingConfig)
         .consumerFactories(
             mediatorConsumerFactoryFactory.createMessageBusConsumerFactory(
-                FLOW_MAPPER_EVENT_TOPIC, CONSUMER_GROUP, messagingConfig
+                listOf(
+                    FLOW_MAPPER_START,
+                    FLOW_MAPPER_SESSION_IN,
+                    FLOW_MAPPER_SESSION_OUT,
+                ),
+                CONSUMER_GROUP,
+                messagingConfig
             ),
         )
         .clientFactories(
@@ -84,8 +94,14 @@ class FlowMapperEventMediatorFactoryImpl @Activate constructor(
         MessageRouter { message ->
             when (val event = message.payload) {
                 is AppMessage -> routeTo(messageBusClient, P2P_OUT_TOPIC)
-                is FlowEvent -> routeTo(messageBusClient, FLOW_EVENT_TOPIC)
-                is FlowMapperEvent -> routeTo(messageBusClient, FLOW_MAPPER_EVENT_TOPIC)
+                is FlowEvent -> {
+                    if (event.payload is StartFlow) {
+                        routeTo(messageBusClient, FLOW_START)
+                    } else {
+                        routeTo(messageBusClient, FLOW_SESSION)
+                    }
+                }
+                is FlowMapperEvent -> routeTo(messageBusClient, FLOW_MAPPER_SESSION_IN)
                 else -> {
                     val eventType = event?.let { it::class.java }
                     throw IllegalStateException("No route defined for event type [$eventType]")

components/flow/flow-mapper-service/src/main/kotlin/net/corda/session/mapper/service/executor/FlowMapperListener.kt

@@ -6,7 +6,7 @@ import net.corda.data.flow.state.mapper.FlowMapperState
 import net.corda.data.flow.state.mapper.FlowMapperStateType
 import net.corda.messaging.api.records.Record
 import net.corda.messaging.api.subscription.listener.StateAndEventListener
-import net.corda.schema.Schemas.Flow.FLOW_MAPPER_EVENT_TOPIC
+import net.corda.schema.Schemas.Flow.FLOW_MAPPER_SESSION_IN
 import net.corda.utilities.debug
 import net.corda.utilities.trace
 import org.slf4j.LoggerFactory
@@ -38,7 +38,7 @@ class FlowMapperListener(
                     publisher?.publish(
                         listOf(
                             Record(
-                                FLOW_MAPPER_EVENT_TOPIC, key, FlowMapperEvent(
+                                FLOW_MAPPER_SESSION_IN, key, FlowMapperEvent(
                                     ExecuteCleanup(listOf())
                                 )
                             )
@@ -81,7 +81,7 @@ class FlowMapperListener(
             executorService.schedule(
                 {
                     log.debug { "Clearing up mapper state for key $eventKey" }
-                    publisher?.publish(listOf(Record(FLOW_MAPPER_EVENT_TOPIC, eventKey, FlowMapperEvent(ExecuteCleanup(listOf())))))
+                    publisher?.publish(listOf(Record(FLOW_MAPPER_SESSION_IN, eventKey, FlowMapperEvent(ExecuteCleanup(listOf())))))
                 },
                 expiryTime - clock.millis(),
                 TimeUnit.MILLISECONDS