CORDA-1416: Upgrade version of Proton-J library (#3050)

* CORDA-1416: Upgrade version of Proton-J library

* CORDA-1416: Compilation fixes following Proton-J upgrade

Reflects:
https://issues.apache.org/jira/browse/PROTON-1712
and
https://issues.apache.org/jira/browse/PROTON-1672

* CORDA-1416: Add an integration test to prove that data saved by from previous version can be read.

* CORDA-1416: Add additional check validate serialized form.

finance/build.gradle

@@ -17,6 +17,9 @@ sourceSets {
             runtimeClasspath += main.output + test.output
             srcDir file('src/integration-test/kotlin')
         }
+        resources {
+            srcDir file('src/integration-test/resources')
+        }
     }
 }
 

finance/src/integration-test/kotlin/net/corda/finance/compat/CompatibilityTest.kt

@@ -0,0 +1,38 @@
+package net.corda.finance.compat
+
+import net.corda.core.serialization.SerializationDefaults
+import net.corda.core.serialization.deserialize
+import net.corda.core.serialization.serialize
+import net.corda.core.transactions.SignedTransaction
+import net.corda.finance.contracts.asset.Cash
+import net.corda.testing.core.SerializationEnvironmentRule
+import org.junit.Rule
+import org.junit.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertNotNull
+import kotlin.test.assertTrue
+
+// TODO: If this type of testing gets momentum, we can create a mini-framework that rides through list of files
+// and performs necessary validation on all of them.
+class CompatibilityTest {
+
+    @Rule
+    @JvmField
+    val testSerialization = SerializationEnvironmentRule()
+
+    @Test
+    fun issueCashTansactionReadTest() {
+        val inputStream = javaClass.classLoader.getResourceAsStream("compatibilityData/v3/node_transaction.dat")
+        assertNotNull(inputStream)
+        val inByteArray: ByteArray = inputStream.readBytes()
+        val transaction = inByteArray.deserialize<SignedTransaction>(context = SerializationDefaults.STORAGE_CONTEXT)
+        assertNotNull(transaction)
+        val commands = transaction.tx.commands
+        assertEquals(1, commands.size)
+        assertTrue(commands.first().value is Cash.Commands.Issue)
+
+        // Serialize back and check that representation is byte-to-byte identical to what it was originally.
+        val serializedForm = transaction.serialize(context = SerializationDefaults.STORAGE_CONTEXT)
+        assertTrue(inByteArray.contentEquals(serializedForm.bytes))
+    }
+}

node-api/build.gradle

@@ -35,7 +35,7 @@ dependencies {
     compile "de.javakaffee:kryo-serializers:0.41"
 
     // For AMQP serialisation.
-    compile "org.apache.qpid:proton-j:0.21.0"
+    compile "org.apache.qpid:proton-j:0.27.1"
 
     // FastClasspathScanner: classpath scanning - needed for the NetworkBootstraper
     compile 'io.github.lukehutch:fast-classpath-scanner:2.12.3'

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

@@ -130,7 +130,7 @@ 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?>()
+            val properties = HashMap<String, Any?>()
             for (key in P2PMessagingHeaders.whitelistedHeaders) {
                 if (artemisMessage.containsProperty(key)) {
                     var value = artemisMessage.getObjectProperty(key)

node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/engine/NettyWritable.kt

@@ -1,6 +1,7 @@
 package net.corda.nodeapi.internal.protonwrapper.engine
 
 import io.netty.buffer.ByteBuf
+import org.apache.qpid.proton.codec.ReadableBuffer
 import org.apache.qpid.proton.codec.WritableBuffer
 import java.nio.ByteBuffer
 
@@ -57,6 +58,10 @@ internal class NettyWritable(val nettyBuffer: ByteBuf) : WritableBuffer {
         nettyBuffer.writeBytes(payload)
     }
 
+    override fun put(payload: ReadableBuffer) {
+        nettyBuffer.writeBytes(payload.byteBuffer())
+    }
+
     override fun limit(): Int {
         return nettyBuffer.capacity()
     }

node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/messages/ApplicationMessage.kt

@@ -10,5 +10,5 @@ interface ApplicationMessage {
     val topic: String
     val destinationLegalName: String
     val destinationLink: NetworkHostAndPort
-    val applicationProperties: Map<Any?, Any?>
+    val applicationProperties: Map<String, Any?>
 }

node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/messages/impl/ReceivedMessageImpl.kt

@@ -16,7 +16,7 @@ internal class ReceivedMessageImpl(override val payload: ByteArray,
                                    override val sourceLink: NetworkHostAndPort,
                                    override val destinationLegalName: String,
                                    override val destinationLink: NetworkHostAndPort,
-                                   override val applicationProperties: Map<Any?, Any?>,
+                                   override val applicationProperties: Map<String, Any?>,
                                    private val channel: Channel,
                                    private val delivery: Delivery) : ReceivedMessage {
     data class MessageCompleter(val status: MessageStatus, val delivery: Delivery)

node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/messages/impl/SendableMessageImpl.kt

@@ -15,7 +15,7 @@ internal class SendableMessageImpl(override val payload: ByteArray,
                                    override val topic: String,
                                    override val destinationLegalName: String,
                                    override val destinationLink: NetworkHostAndPort,
-                                   override val applicationProperties: Map<Any?, Any?>) : SendableMessage {
+                                   override val applicationProperties: Map<String, Any?>) : SendableMessage {
     var buf: ByteBuf? = null
     @Volatile
     var status: MessageStatus = MessageStatus.Unsent

node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/netty/AMQPClient.kt

@@ -170,7 +170,7 @@ class AMQPClient(val targets: List<NetworkHostAndPort>,
     fun createMessage(payload: ByteArray,
                       topic: String,
                       destinationLegalName: String,
-                      properties: Map<Any?, Any?>): SendableMessage {
+                      properties: Map<String, Any?>): SendableMessage {
         return SendableMessageImpl(payload, topic, destinationLegalName, currentTarget, properties)
     }
 

node-api/src/main/kotlin/net/corda/nodeapi/internal/protonwrapper/netty/AMQPServer.kt

@@ -155,7 +155,7 @@ class AMQPServer(val hostName: String,
                       topic: String,
                       destinationLegalName: String,
                       destinationLink: NetworkHostAndPort,
-                      properties: Map<Any?, Any?>): SendableMessage {
+                      properties: Map<String, Any?>): SendableMessage {
         val dest = InetSocketAddress(destinationLink.host, destinationLink.port)
         require(dest in clientChannels.keys) {
             "Destination not available"

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

@@ -152,7 +152,7 @@ class ProtonWrapperTests {
         artemis.session.createQueue(sendAddress, RoutingType.ANYCAST, "queue", true)
         val consumer = artemis.session.createConsumer("queue")
         val testData = "Test".toByteArray()
-        val testProperty = mutableMapOf<Any?, Any?>()
+        val testProperty = mutableMapOf<String, Any?>()
         testProperty["TestProp"] = "1"
         val message = amqpClient.createMessage(testData, sendAddress, CHARLIE_NAME.toString(), testProperty)
         amqpClient.write(message)