Fix the ToStringSerializer so it interacts with reference counting co… (#1385)

* Fix the ToStringSerializer so it interacts with reference counting correctly, and stop ref counting byte arrays since they are invariable wrapped in an object we do reference count.

* Pull out object reference condition into a helper function shared by serialization and deserialization

node-api/src/main/kotlin/net/corda/nodeapi/internal/serialization/amqp/CustomSerializer.kt

@@ -1,6 +1,5 @@
 package net.corda.nodeapi.internal.serialization.amqp
 
-import net.corda.core.serialization.SerializationDefaults
 import net.corda.nodeapi.internal.serialization.amqp.SerializerFactory.Companion.nameForType
 import org.apache.qpid.proton.codec.Data
 import java.lang.reflect.Type
@@ -160,11 +159,11 @@ abstract class CustomSerializer<T> : AMQPSerializer<T> {
                         descriptor, emptyList())))
 
         override fun writeDescribedObject(obj: T, data: Data, type: Type, output: SerializationOutput) {
-            data.putObject(unmaker(obj))
+            data.putString(unmaker(obj))
         }
 
         override fun readObject(obj: Any, schema: Schema, input: DeserializationInput): T {
-            val proxy = input.readObject(obj, schema, String::class.java) as String
+            val proxy = obj as String
             return maker(proxy)
         }
     }

node-api/src/main/kotlin/net/corda/nodeapi/internal/serialization/amqp/DeserializationInput.kt

@@ -142,7 +142,7 @@ class DeserializationInput(internal val serializerFactory: SerializerFactory) {
             }
             // Store the reference in case we need it later on.
             // Skip for primitive types as they are too small and overhead of referencing them will be much higher than their content
-            if (type.asClass()?.isPrimitive != true) objectHistory.add(objectRead)
+            if (suitableForObjectReference(type)) objectHistory.add(objectRead)
             objectRead
         }
 

node-api/src/main/kotlin/net/corda/nodeapi/internal/serialization/amqp/SerializationHelper.kt

@@ -219,4 +219,7 @@ internal fun Type.asParameterizedType(): ParameterizedType {
 
 internal fun Type.isSubClassOf(type: Type): Boolean {
     return TypeToken.of(this).isSubtypeOf(type)
-}
+}
+
+internal fun suitableForObjectReference(type: Type) =
+        type != ByteArray::class.java && type.asClass()?.isPrimitive != true

node-api/src/main/kotlin/net/corda/nodeapi/internal/serialization/amqp/SerializationOutput.kt

@@ -87,7 +87,7 @@ open class SerializationOutput(internal val serializerFactory: SerializerFactory
             // Important to do it after serialization such that dependent object will have preceding reference numbers
             // assigned to them first as they will be first read from the stream on receiving end.
             // Skip for primitive types as they are too small and overhead of referencing them will be much higher than their content
-            if (type.asClass()?.isPrimitive != true) objectHistory.put(obj, objectHistory.size)
+            if (suitableForObjectReference(type)) objectHistory.put(obj, objectHistory.size)
         }
         else {
             data.writeReferencedObject(ReferencedObject(retrievedRefCount))

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

@@ -21,6 +21,7 @@ import net.corda.testing.MEGA_CORP_PUBKEY
 import org.apache.qpid.proton.amqp.*
 import org.apache.qpid.proton.codec.DecoderImpl
 import org.apache.qpid.proton.codec.EncoderImpl
+import org.junit.Assert.assertNotSame
 import org.junit.Assert.assertSame
 import org.junit.Ignore
 import org.junit.Test
@@ -834,4 +835,35 @@ class SerializationOutputTests {
         val factory2 = SerializerFactory(AllWhitelist, ClassLoader.getSystemClassLoader())
         serdes(obj, factory, factory2)
     }
+
+    data class BigDecimals(val a: BigDecimal, val b: BigDecimal)
+
+    @Test
+    fun `test toString custom serializer`() {
+        val factory = SerializerFactory(AllWhitelist, ClassLoader.getSystemClassLoader())
+        factory.register(net.corda.nodeapi.internal.serialization.amqp.custom.BigDecimalSerializer)
+
+        val factory2 = SerializerFactory(AllWhitelist, ClassLoader.getSystemClassLoader())
+        factory2.register(net.corda.nodeapi.internal.serialization.amqp.custom.BigDecimalSerializer)
+
+        val obj = BigDecimals(BigDecimal.TEN, BigDecimal.TEN)
+        val objCopy = serdes(obj, factory, factory2)
+        assertEquals(objCopy.a, objCopy.b)
+    }
+
+    data class ByteArrays(val a: ByteArray, val b: ByteArray)
+
+    @Test
+    fun `test byte arrays not reference counted`() {
+        val factory = SerializerFactory(AllWhitelist, ClassLoader.getSystemClassLoader())
+        factory.register(net.corda.nodeapi.internal.serialization.amqp.custom.BigDecimalSerializer)
+
+        val factory2 = SerializerFactory(AllWhitelist, ClassLoader.getSystemClassLoader())
+        factory2.register(net.corda.nodeapi.internal.serialization.amqp.custom.BigDecimalSerializer)
+
+        val bytes = ByteArray(1)
+        val obj = ByteArrays(bytes, bytes)
+        val objCopy = serdes(obj, factory, factory2, false, false)
+        assertNotSame(objCopy.a, objCopy.b)
+    }
 }