Merge remote-tracking branch 'remotes/open/master' into mnesbit-merge-20180307

# Conflicts:
#	node/src/main/kotlin/net/corda/node/services/messaging/Messaging.kt
#	node/src/main/kotlin/net/corda/node/services/messaging/P2PMessagingClient.kt
#	node/src/main/kotlin/net/corda/node/services/statemachine/StateMachineManagerImpl.kt

node-api/src/main/kotlin/net/corda/nodeapi/internal/ArtemisMessagingComponent.kt

@@ -17,6 +17,8 @@ import net.corda.core.messaging.MessageRecipients
 import net.corda.core.messaging.SingleMessageRecipient
 import net.corda.core.serialization.CordaSerializable
 import net.corda.core.utilities.NetworkHostAndPort
+import org.apache.activemq.artemis.api.core.Message
+import org.apache.activemq.artemis.api.core.SimpleString
 import java.security.PublicKey
 
 /**
@@ -38,6 +40,49 @@ class ArtemisMessagingComponent {
         const val BRIDGE_CONTROL = "${INTERNAL_PREFIX}bridge.control"
         const val BRIDGE_NOTIFY = "${INTERNAL_PREFIX}bridge.notify"
         const val NOTIFICATIONS_ADDRESS = "${INTERNAL_PREFIX}activemq.notifications"
+
+        /**
+         * In the operation mode where we have an out of process bridge we cannot correctly populate the Artemis validated user header
+         * as the TLS does not terminate directly onto Artemis. We therefore use this internal only header to forward
+         * the equivalent information from the Float.
+         */
+        val bridgedCertificateSubject = SimpleString("sender-subject-name")
+
+        object P2PMessagingHeaders {
+            // This is a "property" attached to an Artemis MQ message object, which contains our own notion of "topic".
+            // We should probably try to unify our notion of "topic" (really, just a string that identifies an endpoint
+            // that will handle messages, like a URL) with the terminology used by underlying MQ libraries, to avoid
+            // confusion.
+            val topicProperty = SimpleString("platform-topic")
+            val cordaVendorProperty = SimpleString("corda-vendor")
+            val releaseVersionProperty = SimpleString("release-version")
+            val platformVersionProperty = SimpleString("platform-version")
+            val senderUUID = SimpleString("sender-uuid")
+            val senderSeqNo = SimpleString("send-seq-no")
+            /**
+             * In the operation mode where we have an out of process bridge we cannot correctly populate the Artemis validated user header
+             * as the TLS does not terminate directly onto Artemis. We therefore use this internal only header to forward
+             * the equivalent information from the Float.
+             */
+            val bridgedCertificateSubject = SimpleString("sender-subject-name")
+
+
+            object Type {
+                const val KEY = "corda_p2p_message_type"
+                const val SESSION_INIT_VALUE = "session_init"
+            }
+
+            val whitelistedHeaders: Set<String> = setOf(topicProperty.toString(),
+                    cordaVendorProperty.toString(),
+                    releaseVersionProperty.toString(),
+                    platformVersionProperty.toString(),
+                    senderUUID.toString(),
+                    senderSeqNo.toString(),
+                    bridgedCertificateSubject.toString(),
+                    Type.KEY,
+                    Message.HDR_DUPLICATE_DETECTION_ID.toString(),
+                    Message.HDR_VALIDATED_USER.toString())
+        }
     }
 
     interface ArtemisAddress : MessageRecipients {

node-api/src/main/kotlin/net/corda/nodeapi/internal/bridging/AMQPBridgeManager.kt

@@ -20,6 +20,7 @@ import net.corda.core.utilities.debug
 import net.corda.nodeapi.internal.ArtemisMessagingClient
 import net.corda.nodeapi.internal.ArtemisMessagingComponent
 import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.NODE_USER
+import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.P2PMessagingHeaders
 import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.PEER_USER
 import net.corda.nodeapi.internal.ArtemisMessagingComponent.RemoteInboxAddress.Companion.translateLocalQueueToInboxAddress
 import net.corda.nodeapi.internal.ArtemisSessionProvider
@@ -138,12 +139,14 @@ class AMQPBridgeManager(config: NodeSSLConfiguration, val artemisMessageClientFa
         private fun clientArtemisMessageHandler(artemisMessage: ClientMessage) {
             val data = ByteArray(artemisMessage.bodySize).apply { artemisMessage.bodyBuffer.readBytes(this) }
             val properties = HashMap<Any?, Any?>()
-            for (key in artemisMessage.propertyNames) {
+            for (key in P2PMessagingHeaders.whitelistedHeaders) {
+                if (artemisMessage.containsProperty(key)) {
                     var value = artemisMessage.getObjectProperty(key)
                     if (value is SimpleString) {
                         value = value.toString()
                     }
-                properties[key.toString()] = value
+                    properties[key] = value
+                }
             }
             log.debug { "Bridged Send to ${legalNames.first()} uuid: ${artemisMessage.getObjectProperty("_AMQ_DUPL_ID")}" }
             val peerInbox = translateLocalQueueToInboxAddress(queueName)

node/src/integration-test/kotlin/net/corda/node/amqp/AMQPBridgeTest.kt

@@ -23,6 +23,7 @@ import net.corda.node.services.config.configureWithDevSSLCertificate
 import net.corda.node.services.messaging.ArtemisMessagingServer
 import net.corda.nodeapi.internal.ArtemisMessagingClient
 import net.corda.nodeapi.internal.ArtemisMessagingComponent
+import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.P2PMessagingHeaders
 import net.corda.nodeapi.internal.bridging.AMQPBridgeManager
 import net.corda.nodeapi.internal.bridging.BridgeManager
 import net.corda.nodeapi.internal.protonwrapper.netty.AMQPServer
@@ -73,7 +74,7 @@ class AMQPBridgeTest {
         val artemis = artemisClient.started!!
         for (i in 0 until 3) {
             val artemisMessage = artemis.session.createMessage(true).apply {
-                putIntProperty("CountProp", i)
+                putIntProperty(P2PMessagingHeaders.senderUUID, i)
                 writeBodyBufferBytes("Test$i".toByteArray())
                 // Use the magic deduplication property built into Artemis as our message identity too
                 putStringProperty(HDR_DUPLICATE_DETECTION_ID, SimpleString(UUID.randomUUID().toString()))
@@ -97,7 +98,7 @@ class AMQPBridgeTest {
         }
 
         val received1 = receive.next()
-        val messageID1 = received1.applicationProperties["CountProp"] as Int
+        val messageID1 = received1.applicationProperties[P2PMessagingHeaders.senderUUID.toString()] as Int
         assertArrayEquals("Test$messageID1".toByteArray(), received1.payload)
         assertEquals(0, messageID1)
         dedupeSet += received1.applicationProperties[HDR_DUPLICATE_DETECTION_ID.toString()] as String
@@ -106,7 +107,7 @@ class AMQPBridgeTest {
         atNodeSequence += messageID1
 
         val received2 = receive.next()
-        val messageID2 = received2.applicationProperties["CountProp"] as Int
+        val messageID2 = received2.applicationProperties[P2PMessagingHeaders.senderUUID.toString()] as Int
         assertArrayEquals("Test$messageID2".toByteArray(), received2.payload)
         assertEquals(1, messageID2, formatMessage("1", messageID2, receivedSequence))
         received2.complete(false) // Reject message and don't add to dedupe
@@ -115,7 +116,7 @@ class AMQPBridgeTest {
         // drop things until we get back to the replay
         while (true) {
             val received3 = receive.next()
-            val messageID3 = received3.applicationProperties["CountProp"] as Int
+            val messageID3 = received3.applicationProperties[P2PMessagingHeaders.senderUUID.toString()] as Int
             assertArrayEquals("Test$messageID3".toByteArray(), received3.payload)
             receivedSequence += messageID3
             if (messageID3 != 1) { // keep rejecting any batched items following rejection
@@ -134,7 +135,7 @@ class AMQPBridgeTest {
         // start receiving again, but discarding duplicates
         while (true) {
             val received4 = receive.next()
-            val messageID4 = received4.applicationProperties["CountProp"] as Int
+            val messageID4 = received4.applicationProperties[P2PMessagingHeaders.senderUUID.toString()] as Int
             assertArrayEquals("Test$messageID4".toByteArray(), received4.payload)
             receivedSequence += messageID4
             val messageId = received4.applicationProperties[HDR_DUPLICATE_DETECTION_ID.toString()] as String
@@ -150,7 +151,7 @@ class AMQPBridgeTest {
 
         // Send a fresh item and check receive
         val artemisMessage = artemis.session.createMessage(true).apply {
-            putIntProperty("CountProp", 3)
+            putIntProperty(P2PMessagingHeaders.senderUUID, 3)
             writeBodyBufferBytes("Test3".toByteArray())
             // Use the magic deduplication property built into Artemis as our message identity too
             putStringProperty(HDR_DUPLICATE_DETECTION_ID, SimpleString(UUID.randomUUID().toString()))
@@ -161,7 +162,7 @@ class AMQPBridgeTest {
         // start receiving again, discarding duplicates
         while (true) {
             val received5 = receive.next()
-            val messageID5 = received5.applicationProperties["CountProp"] as Int
+            val messageID5 = received5.applicationProperties[P2PMessagingHeaders.senderUUID.toString()] as Int
             assertArrayEquals("Test$messageID5".toByteArray(), received5.payload)
             receivedSequence += messageID5
             val messageId = received5.applicationProperties[HDR_DUPLICATE_DETECTION_ID.toString()] as String

node/src/main/kotlin/net/corda/node/services/messaging/Messaging.kt

@@ -176,10 +176,3 @@ interface AcknowledgeHandle {
 
 typealias MessageHandler = (ReceivedMessage, MessageHandlerRegistration, AcknowledgeHandle) -> Unit
 
-object P2PMessagingHeaders {
-
-    object Type {
-        const val KEY = "corda_p2p_message_type"
-        const val SESSION_INIT_VALUE = "session_init"
-    }
-}

node/src/main/kotlin/net/corda/node/services/messaging/P2PMessagingClient.kt

@@ -46,6 +46,7 @@ import net.corda.nodeapi.internal.ArtemisMessagingComponent
 import net.corda.nodeapi.internal.ArtemisMessagingComponent.ArtemisAddress
 import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.BRIDGE_CONTROL
 import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.BRIDGE_NOTIFY
+import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.P2PMessagingHeaders
 import net.corda.nodeapi.internal.ArtemisMessagingComponent.Companion.PEERS_PREFIX
 import net.corda.nodeapi.internal.ArtemisMessagingComponent.NodeAddress
 import net.corda.nodeapi.internal.ArtemisMessagingComponent.RemoteInboxAddress
@@ -124,17 +125,7 @@ class P2PMessagingClient(val config: NodeConfiguration,
 ) : SingletonSerializeAsToken(), MessagingService, AddressToArtemisQueueResolver, AutoCloseable {
     companion object {
         private val log = contextLogger()
-        // This is a "property" attached to an Artemis MQ message object, which contains our own notion of "topic".
+        private val amqDelayMillis = System.getProperty("amq.delivery.delay.ms", "0").toInt()
-        // We should probably try to unify our notion of "topic" (really, just a string that identifies an endpoint
-        // that will handle messages, like a URL) with the terminology used by underlying MQ libraries, to avoid
-        // confusion.
-        val topicProperty = SimpleString("platform-topic")
-        val cordaVendorProperty = SimpleString("corda-vendor")
-        val releaseVersionProperty = SimpleString("release-version")
-        val platformVersionProperty = SimpleString("platform-version")
-        val senderUUID = SimpleString("sender-uuid")
-        val senderSeqNo = SimpleString("send-seq-no")
-
         private const val messageMaxRetryCount: Int = 3
 
         fun createMessageToRedeliver(): PersistentMap<Long, Pair<Message, MessageRecipients>, RetryMessage, Long> {
@@ -403,9 +394,9 @@ class P2PMessagingClient(val config: NodeConfiguration,
 
     private fun artemisToCordaMessage(message: ClientMessage): ReceivedMessage? {
         try {
-            val topic = message.required(topicProperty) { getStringProperty(it) }
+            val topic = message.required(P2PMessagingHeaders.topicProperty) { getStringProperty(it) }
             val user = requireNotNull(message.getStringProperty(HDR_VALIDATED_USER)) { "Message is not authenticated" }
-            val platformVersion = message.required(platformVersionProperty) { getIntProperty(it) }
+            val platformVersion = message.required(P2PMessagingHeaders.platformVersionProperty) { getIntProperty(it) }
             // Use the magic deduplication property built into Artemis as our message identity too
             val uniqueMessageId = message.required(HDR_DUPLICATE_DETECTION_ID) { DeduplicationId(message.getStringProperty(it)) }
             val receivedSenderUUID = message.getStringProperty(senderUUID)