CORDA-1098 - Fixed Artemis filters for flows draining mode. (corda#2612)

BevanH · Feb 22, 2018 · a4f6760 · a4f6760
1 parent de57790
commit a4f6760
Show file tree

Hide file tree

Showing 2 changed files with 8 additions and 7 deletions.
diff --git a/node/src/integration-test/kotlin/net/corda/node/modes/draining/P2PFlowsDrainingModeTest.kt b/node/src/integration-test/kotlin/net/corda/node/modes/draining/P2PFlowsDrainingModeTest.kt
@@ -18,8 +18,9 @@ import org.assertj.core.api.AssertionsForInterfaceTypes.assertThat
 import org.junit.After
 import org.junit.Before
 import org.junit.Test
-import java.util.concurrent.ExecutorService
 import java.util.concurrent.Executors
+import java.util.concurrent.ScheduledExecutorService
+import java.util.concurrent.TimeUnit
 import kotlin.test.fail
 
 class P2PFlowsDrainingModeTest {
@@ -28,15 +29,15 @@ class P2PFlowsDrainingModeTest {
     private val user = User("mark", "dadada", setOf(Permissions.all()))
     private val users = listOf(user)
 
-    private var executor: ExecutorService? = null
+    private var executor: ScheduledExecutorService? = null
 
     companion object {
         private val logger = loggerFor<P2PFlowsDrainingModeTest>()
     }
 
     @Before
     fun setup() {
-        executor = Executors.newSingleThreadExecutor()
+        executor = Executors.newSingleThreadScheduledExecutor()
     }
 
     @After
@@ -59,11 +60,11 @@ class P2PFlowsDrainingModeTest {
             initiating.apply {
                 val flow = startFlow(::InitiateSessionFlow, counterParty)
                 // this should be really fast, for the flow has already started, so 5 seconds should never be a problem
-                executor!!.submit({
+                executor!!.schedule({
                     logger.info("Now disabling flows draining mode for $counterParty.")
                     shouldFail = false
                     initiated.setFlowsDrainingModeEnabled(false)
-                })
+                }, 5, TimeUnit.SECONDS)
                 flow.returnValue.map { result ->
                     if (shouldFail) {
                         fail("Shouldn't happen until flows draining mode is switched off.")

diff --git a/node/src/main/kotlin/net/corda/node/services/messaging/P2PMessagingClient.kt b/node/src/main/kotlin/net/corda/node/services/messaging/P2PMessagingClient.kt
@@ -668,8 +668,8 @@ private class P2PMessagingConsumer(
         private val drainingModeWasChangedEvents: Observable<Pair<Boolean, Boolean>>) : LifecycleSupport {
 
     private companion object {
-        private const val initialSessionMessages = "${P2PMessagingHeaders.Type.KEY}=${P2PMessagingHeaders.Type.SESSION_INIT_VALUE}"
-        private const val existingSessionMessages = "${P2PMessagingHeaders.Type.KEY}<>${P2PMessagingHeaders.Type.SESSION_INIT_VALUE}"
+        private const val initialSessionMessages = "${P2PMessagingHeaders.Type.KEY}='${P2PMessagingHeaders.Type.SESSION_INIT_VALUE}'"
+        private const val existingSessionMessages = "${P2PMessagingHeaders.Type.KEY}<>'${P2PMessagingHeaders.Type.SESSION_INIT_VALUE}'"
     }
 
     private var startedFlag = false