Make integration tests pass in AMQP mode, part 1 (#1855)

docs/source/serialization.rst

@@ -63,7 +63,7 @@ The long term goal is to migrate the current serialization format for everything
     #.  A desire to support open-ended polymorphism, where the number of subclasses of a superclass can expand over time
         and do not need to be defined in the schema *upfront*, which is key to many Corda concepts, such as contract states.
     #.  Increased security from deserialized objects being constructed through supported constructors rather than having
-        data poked directy into their fields without an opportunity to validate consistency or intercept attempts to manipulate
+        data poked directly into their fields without an opportunity to validate consistency or intercept attempts to manipulate
         supposed invariants.
 
 Documentation on that format, and how JVM classes are translated to AMQP, will be linked here when it is available.
@@ -259,9 +259,10 @@ Kotlin Objects
 ``````````````
 
     #.  Kotlin ``object`` s are singletons and treated differently.  They are recorded into the stream with no properties
-        and deserialize back to the singleton instance.
+        and deserialize back to the singleton instance. Currently, the same is not true of Java singletons,
-
+        and they will deserialize to new instances of the class.
-Currently, the same is not true of Java singletons, and they will deserialize to new instances of the class.
+    #.  Kotlin's anonymous ``object`` s are not currently supported. I.e. constructs like:
+        ``object : Contract {...}`` will not serialize correctly and need to be re-written as an explicit class declaration.
 
 The Carpenter
 `````````````

node-api/src/test/kotlin/net/corda/nodeapi/internal/serialization/amqp/SerializationOutputTests.kt

@@ -534,8 +534,22 @@ class SerializationOutputTests {
         }
     }
 
-    val FOO_PROGRAM_ID = "net.corda.nodeapi.internal.serialization.amqp.SerializationOutputTests.FooContract"
+    @Test
+    fun `test custom object`() {
+        serdes(FooContract)
+    }
 
+    @Test
+    @Ignore("Cannot serialize due to known Kotlin/serialization limitation")
+    fun `test custom anonymous object`() {
+        val anonymous: Contract = object : Contract {
+            override fun verify(tx: LedgerTransaction) {
+            }
+        }
+        serdes(anonymous)
+    }
+
+    private val FOO_PROGRAM_ID = "net.corda.nodeapi.internal.serialization.amqp.SerializationOutputTests.FooContract"
     class FooState : ContractState {
         override val participants: List<AbstractParty> = emptyList()
     }

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

@@ -213,6 +213,7 @@ data class TopicSession(val topic: String, val sessionID: Long = MessagingServic
  * These IDs and timestamps should not be assumed to be globally unique, although due to the nanosecond precision of
  * the timestamp field they probably will be, even if an implementation just uses a hash prefix as the message id.
  */
+@CordaSerializable
 interface Message {
     val topicSession: TopicSession
     val data: ByteArray

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

@@ -133,6 +133,11 @@ class NodeMessagingClient(override val config: NodeConfiguration,
                     persistentEntityClass = RetryMessage::class.java
             )
         }
+
+        private class NodeClientMessage(override val topicSession: TopicSession, override val data: ByteArray, override val uniqueMessageId: UUID) : Message {
+            override val debugTimestamp: Instant = Instant.now()
+            override fun toString() = "$topicSession#${String(data)}"
+        }
     }
 
     private class InnerState {
@@ -599,13 +604,7 @@ class NodeMessagingClient(override val config: NodeConfiguration,
 
     override fun createMessage(topicSession: TopicSession, data: ByteArray, uuid: UUID): Message {
         // TODO: We could write an object that proxies directly to an underlying MQ message here and avoid copying.
-        return object : Message {
+        return NodeClientMessage(topicSession, data, uuid)
-            override val topicSession: TopicSession = topicSession
-            override val data: ByteArray = data
-            override val debugTimestamp: Instant = Instant.now()
-            override val uniqueMessageId: UUID = uuid
-            override fun toString() = "$topicSession#${String(data)}"
-        }
     }
 
     private fun createOutOfProcessVerifierService(): TransactionVerifierService {

testing/node-driver/src/main/kotlin/net/corda/testing/node/InMemoryMessagingNetwork.kt

@@ -279,7 +279,6 @@ class InMemoryMessagingNetwork(
         _sentMessages.onNext(transfer)
     }
 
-    @CordaSerializable
     private data class InMemoryMessage(override val topicSession: TopicSession,
                                        override val data: ByteArray,
                                        override val uniqueMessageId: UUID,
@@ -287,7 +286,6 @@ class InMemoryMessagingNetwork(
         override fun toString() = "$topicSession#${String(data)}"
     }
 
-    @CordaSerializable
     private data class InMemoryReceivedMessage(override val topicSession: TopicSession,
                                                override val data: ByteArray,
                                                override val platformVersion: Int,