CORDA-973 Refactoring for serialization compression support (#2466)

* Use constant for empty byte array
* Less byte array copying
* Fix InputStreamSerializer trailing garbage
* More OO kryo streams
* Introduce SerializationMagic
* Introduce non-copying slice on ByteSequence

.ci/api-current.txt

@@ -3150,8 +3150,8 @@ public final class net.corda.core.utilities.ByteArrays extends java.lang.Object
   @org.jetbrains.annotations.NotNull public final net.corda.core.utilities.ByteSequence copy()
   public boolean equals(Object)
   @org.jetbrains.annotations.NotNull public abstract byte[] getBytes()
-  public abstract int getOffset()
+  public final int getOffset()
-  public abstract int getSize()
+  public final int getSize()
   public int hashCode()
   @kotlin.jvm.JvmStatic @org.jetbrains.annotations.NotNull public static final net.corda.core.utilities.ByteSequence of(byte[])
   @kotlin.jvm.JvmStatic @org.jetbrains.annotations.NotNull public static final net.corda.core.utilities.ByteSequence of(byte[], int)
@@ -3274,8 +3274,6 @@ public static final class net.corda.core.utilities.NonEmptySet$iterator$1 extend
 @net.corda.core.serialization.CordaSerializable public class net.corda.core.utilities.OpaqueBytes extends net.corda.core.utilities.ByteSequence
   public <init>(byte[])
   @org.jetbrains.annotations.NotNull public final byte[] getBytes()
-  public int getOffset()
-  public int getSize()
   public static final net.corda.core.utilities.OpaqueBytes$Companion Companion
 ##
 public static final class net.corda.core.utilities.OpaqueBytes$Companion extends java.lang.Object
@@ -3283,8 +3281,6 @@ public static final class net.corda.core.utilities.OpaqueBytes$Companion extends
 @net.corda.core.serialization.CordaSerializable public final class net.corda.core.utilities.OpaqueBytesSubSequence extends net.corda.core.utilities.ByteSequence
   public <init>(byte[], int, int)
   @org.jetbrains.annotations.NotNull public byte[] getBytes()
-  public int getOffset()
-  public int getSize()
 ##
 @net.corda.core.serialization.CordaSerializable public final class net.corda.core.utilities.ProgressTracker extends java.lang.Object
   public final void endWithError(Throwable)

client/rpc/src/main/kotlin/net/corda/client/rpc/internal/KryoClientSerializationScheme.kt

@@ -2,22 +2,22 @@ package net.corda.client.rpc.internal
 
 import com.esotericsoftware.kryo.pool.KryoPool
 import net.corda.core.serialization.SerializationContext
+import net.corda.nodeapi.internal.serialization.CordaSerializationMagic
 import net.corda.core.serialization.internal.SerializationEnvironment
 import net.corda.core.serialization.internal.SerializationEnvironmentImpl
 import net.corda.core.serialization.internal.nodeSerializationEnv
-import net.corda.core.utilities.ByteSequence
 import net.corda.nodeapi.internal.serialization.KRYO_P2P_CONTEXT
 import net.corda.nodeapi.internal.serialization.KRYO_RPC_CLIENT_CONTEXT
 import net.corda.nodeapi.internal.serialization.SerializationFactoryImpl
 import net.corda.nodeapi.internal.serialization.amqp.AMQPClientSerializationScheme
 import net.corda.nodeapi.internal.serialization.kryo.AbstractKryoSerializationScheme
 import net.corda.nodeapi.internal.serialization.kryo.DefaultKryoCustomizer
-import net.corda.nodeapi.internal.serialization.kryo.KryoHeaderV0_1
+import net.corda.nodeapi.internal.serialization.kryo.kryoMagic
 import net.corda.nodeapi.internal.serialization.kryo.RPCKryo
 
 class KryoClientSerializationScheme : AbstractKryoSerializationScheme() {
-    override fun canDeserializeVersion(byteSequence: ByteSequence, target: SerializationContext.UseCase): Boolean {
+    override fun canDeserializeVersion(magic: CordaSerializationMagic, target: SerializationContext.UseCase): Boolean {
-        return byteSequence == KryoHeaderV0_1 && (target == SerializationContext.UseCase.RPCClient || target == SerializationContext.UseCase.P2P)
+        return magic == kryoMagic && (target == SerializationContext.UseCase.RPCClient || target == SerializationContext.UseCase.P2P)
     }
 
     override fun rpcClientKryoPool(context: SerializationContext): KryoPool {

core/build.gradle

@@ -104,7 +104,7 @@ dependencies {
     // Apache JEXL: An embeddable expression evaluation library.
     // This may be temporary until we experiment with other ways to do on-the-fly contract specialisation via an API.
     compile "org.apache.commons:commons-jexl3:3.0"
-
+    compile 'commons-lang:commons-lang:2.6'
     // For JSON
     compile "com.fasterxml.jackson.core:jackson-databind:${jackson_version}"
 

core/src/main/kotlin/net/corda/core/contracts/Structures.kt

@@ -46,7 +46,7 @@ interface NamedByHash {
 @CordaSerializable
 data class Issued<out P : Any>(val issuer: PartyAndReference, val product: P) {
     init {
-        require(issuer.reference.bytes.size <= MAX_ISSUER_REF_SIZE) { "Maximum issuer reference size is $MAX_ISSUER_REF_SIZE." }
+        require(issuer.reference.size <= MAX_ISSUER_REF_SIZE) { "Maximum issuer reference size is $MAX_ISSUER_REF_SIZE." }
     }
     override fun toString() = "$product issued by $issuer"
 }

core/src/main/kotlin/net/corda/core/internal/InternalUtils.kt

@@ -26,6 +26,7 @@ import java.lang.reflect.Field
 import java.math.BigDecimal
 import java.net.HttpURLConnection
 import java.net.URL
+import java.nio.ByteBuffer
 import java.nio.charset.Charset
 import java.nio.charset.StandardCharsets.UTF_8
 import java.nio.file.*
@@ -372,3 +373,5 @@ inline fun <T : Any> SerializedBytes<T>.sign(signer: (SerializedBytes<T>) -> Dig
 inline fun <T : Any> SerializedBytes<T>.sign(keyPair: KeyPair): SignedData<T> {
     return SignedData(this, keyPair.sign(this.bytes))
 }
+
+fun ByteBuffer.copyBytes() = ByteArray(remaining()).also { get(it) }

core/src/main/kotlin/net/corda/core/serialization/SerializationAPI.kt

@@ -98,9 +98,7 @@ abstract class SerializationFactory {
         val currentFactory: SerializationFactory? get() = _currentFactory.get()
     }
 }
-
+typealias SerializationMagic = ByteSequence
-typealias VersionHeader = ByteSequence
-
 /**
  * Parameters to serialization and deserialization.
  */
@@ -108,7 +106,7 @@ interface SerializationContext {
     /**
      * When serializing, use the format this header sequence represents.
      */
-    val preferredSerializationVersion: VersionHeader
+    val preferredSerializationVersion: SerializationMagic
     /**
      * The class loader to use for deserialization.
      */
@@ -161,7 +159,7 @@ interface SerializationContext {
     /**
      * Helper method to return a new context based on this context but with serialization using the format this header sequence represents.
      */
-    fun withPreferredSerializationVersion(versionHeader: VersionHeader): SerializationContext
+    fun withPreferredSerializationVersion(magic: SerializationMagic): SerializationContext
 
     /**
      * The use case that we are serializing for, since it influences the implementations chosen.
@@ -225,6 +223,7 @@ fun <T : Any> T.serialize(serializationFactory: SerializationFactory = Serializa
  * A type safe wrapper around a byte array that contains a serialised object. You can call [SerializedBytes.deserialize]
  * to get the original object back.
  */
+@Suppress("unused")
 class SerializedBytes<T : Any>(bytes: ByteArray) : OpaqueBytes(bytes) {
     // It's OK to use lazy here because SerializedBytes is configured to use the ImmutableClassSerializer.
     val hash: SecureHash by lazy { bytes.sha256() }

core/src/main/kotlin/net/corda/core/utilities/ByteArrays.kt

@@ -4,59 +4,43 @@ package net.corda.core.utilities
 
 import net.corda.core.serialization.CordaSerializable
 import java.io.ByteArrayInputStream
+import java.io.OutputStream
+import java.lang.Math.max
+import java.lang.Math.min
+import java.nio.ByteBuffer
 import javax.xml.bind.DatatypeConverter
 
 /**
  * An abstraction of a byte array, with offset and size that does no copying of bytes unless asked to.
  *
  * The data of interest typically starts at position [offset] within the [bytes] and is [size] bytes long.
+ *
+ * @property offset The start position of the sequence within the byte array.
+ * @property size The number of bytes this sequence represents.
  */
 @CordaSerializable
-sealed class ByteSequence : Comparable<ByteSequence> {
+sealed class ByteSequence(private val _bytes: ByteArray, val offset: Int, val size: Int) : Comparable<ByteSequence> {
-    constructor() {
-        this._bytes = COPY_BYTES
-    }
-
-    /**
-     * This constructor allows to bypass calls to [bytes] for functions in this class if the implementation
-     * of [bytes] makes a copy of the underlying [ByteArray] (as [OpaqueBytes] does for safety).  This improves
-     * performance.  It is recommended to use this constructor rather than the default constructor.
-     */
-    constructor(uncopiedBytes: ByteArray) {
-        this._bytes = uncopiedBytes
-    }
-
     /**
      * The underlying bytes.  Some implementations may choose to make a copy of the underlying [ByteArray] for
      * security reasons.  For example, [OpaqueBytes].
      */
     abstract val bytes: ByteArray
-    /**
-     * The number of bytes this sequence represents.
-     */
-    abstract val size: Int
-    /**
-     * The start position of the sequence within the byte array.
-     */
-    abstract val offset: Int
-
-    private val _bytes: ByteArray
-        get() = if (field === COPY_BYTES) bytes else field
 
     /** Returns a [ByteArrayInputStream] of the bytes */
     fun open() = ByteArrayInputStream(_bytes, offset, size)
 
     /**
-     * Create a sub-sequence backed by the same array.
+     * Create a sub-sequence, that may be backed by a new byte array.
      *
      * @param offset The offset within this sequence to start the new sequence.  Note: not the offset within the backing array.
      * @param size The size of the intended sub sequence.
      */
+    @Suppress("MemberVisibilityCanPrivate")
     fun subSequence(offset: Int, size: Int): ByteSequence {
         require(offset >= 0)
         require(offset + size <= this.size)
         // Intentionally use bytes rather than _bytes, to mirror the copy-or-not behaviour of that property.
-        return if (offset == 0 && size == this.size) this else of(bytes, this.offset + offset, size)
+        return if (offset == 0 && size == this.size) this else OpaqueBytesSubSequence(bytes, this.offset + offset, size)
     }
 
     companion object {
@@ -69,23 +53,40 @@ sealed class ByteSequence : Comparable<ByteSequence> {
         fun of(bytes: ByteArray, offset: Int = 0, size: Int = bytes.size): ByteSequence {
             return OpaqueBytesSubSequence(bytes, offset, size)
         }
-
-        private val COPY_BYTES: ByteArray = ByteArray(0)
     }
 
     /**
      * Take the first n bytes of this sequence as a sub-sequence.  See [subSequence] for further semantics.
      */
-    fun take(n: Int): ByteSequence {
+    fun take(n: Int): ByteSequence = subSequence(0, n)
-        require(size >= n)
+
-        return subSequence(0, n)
+    /**
+     * A new read-only [ByteBuffer] view of this sequence or part of it.
+     * If [start] or [end] are negative then [IllegalArgumentException] is thrown, otherwise they are clamped if necessary.
+     * This method cannot be used to get bytes before [offset] or after [offset]+[size], and never makes a new array.
+     */
+    fun slice(start: Int = 0, end: Int = size): ByteBuffer {
+        require(start >= 0)
+        require(end >= 0)
+        val clampedStart = min(start, size)
+        val clampedEnd = min(end, size)
+        return ByteBuffer.wrap(_bytes, offset + clampedStart, max(0, clampedEnd - clampedStart)).asReadOnlyBuffer()
     }
 
+    /** Write this sequence to an [OutputStream]. */
+    fun writeTo(output: OutputStream) = output.write(_bytes, offset, size)
+
+    /** Write this sequence to a [ByteBuffer]. */
+    fun putTo(buffer: ByteBuffer): ByteBuffer = buffer.put(_bytes, offset, size)
+
     /**
      * Copy this sequence, complete with new backing array.  This can be helpful to break references to potentially
      * large backing arrays from small sub-sequences.
      */
-    fun copy(): ByteSequence = of(_bytes.copyOfRange(offset, offset + size))
+    fun copy(): ByteSequence = of(copyBytes())
+
+    /** Same as [copy] but returns just the new byte array. */
+    fun copyBytes(): ByteArray = _bytes.copyOfRange(offset, offset + size)
 
     /**
      * Compare byte arrays byte by byte.  Arrays that are shorter are deemed less than longer arrays if all the bytes
@@ -133,7 +134,7 @@ sealed class ByteSequence : Comparable<ByteSequence> {
         return result
     }
 
-    override fun toString(): String = "[${_bytes.copyOfRange(offset, offset + size).toHexString()}]"
+    override fun toString(): String = "[${copyBytes().toHexString()}]"
 }
 
 /**
@@ -141,7 +142,7 @@ sealed class ByteSequence : Comparable<ByteSequence> {
  * In an ideal JVM this would be a value type and be completely overhead free. Project Valhalla is adding such
  * functionality to Java, but it won't arrive for a few years yet!
  */
-open class OpaqueBytes(bytes: ByteArray) : ByteSequence(bytes) {
+open class OpaqueBytes(bytes: ByteArray) : ByteSequence(bytes, 0, bytes.size) {
     companion object {
         /**
          * Create [OpaqueBytes] from a sequence of [Byte] values.
@@ -156,8 +157,8 @@ open class OpaqueBytes(bytes: ByteArray) : ByteSequence(bytes) {
 
     /**
      * The bytes are always cloned so that this object becomes immutable. This has been done
-     * to prevent tampering with entities such as [SecureHash] and [PrivacySalt], as well as
+     * to prevent tampering with entities such as [net.corda.core.crypto.SecureHash] and [net.corda.core.contracts.PrivacySalt], as well as
-     * preserve the integrity of our hash constants [zeroHash] and [allOnesHash].
+     * preserve the integrity of our hash constants [net.corda.core.crypto.SecureHash.zeroHash] and [net.corda.core.crypto.SecureHash.allOnesHash].
      *
      * Cloning like this may become a performance issue, depending on whether or not the JIT
      * compiler is ever able to optimise away the clone. In which case we may need to revisit
@@ -165,8 +166,6 @@ open class OpaqueBytes(bytes: ByteArray) : ByteSequence(bytes) {
      */
     override final val bytes: ByteArray = bytes
         get() = field.clone()
-    override val size: Int = bytes.size
-    override val offset: Int = 0
 }
 
 /**
@@ -188,7 +187,7 @@ fun String.parseAsHex(): ByteArray = DatatypeConverter.parseHexBinary(this)
 /**
  * Class is public for serialization purposes
  */
-class OpaqueBytesSubSequence(override val bytes: ByteArray, override val offset: Int, override val size: Int) : ByteSequence(bytes) {
+class OpaqueBytesSubSequence(override val bytes: ByteArray, offset: Int, size: Int) : ByteSequence(bytes, offset, size) {
     init {
         require(offset >= 0 && offset < bytes.size)
         require(size >= 0 && size <= bytes.size)

core/src/test/kotlin/net/corda/core/crypto/Base58Test.kt

@@ -1,5 +1,6 @@
 package net.corda.core.crypto
 
+import org.apache.commons.lang.ArrayUtils.EMPTY_BYTE_ARRAY
 import org.junit.Test
 import java.math.BigInteger
 import java.util.*
@@ -26,7 +27,7 @@ class Base58Test {
         assertEquals("1111111", Base58.encode(zeroBytes7))
 
         // test empty encode
-        assertEquals("", Base58.encode(ByteArray(0)))
+        assertEquals("", Base58.encode(EMPTY_BYTE_ARRAY))
     }
 
     @Test

core/src/test/kotlin/net/corda/core/crypto/CryptoUtilsTest.kt

@@ -8,6 +8,7 @@ import net.i2p.crypto.eddsa.math.GroupElement
 import net.i2p.crypto.eddsa.spec.EdDSANamedCurveSpec
 import net.i2p.crypto.eddsa.spec.EdDSANamedCurveTable
 import net.i2p.crypto.eddsa.spec.EdDSAPublicKeySpec
+import org.apache.commons.lang.ArrayUtils.EMPTY_BYTE_ARRAY
 import org.bouncycastle.asn1.pkcs.PrivateKeyInfo
 import org.bouncycastle.asn1.x509.SubjectPublicKeyInfo
 import org.bouncycastle.jcajce.provider.asymmetric.ec.BCECPrivateKey
@@ -77,7 +78,7 @@ class CryptoUtilsTest {
 
         // test for empty data signing
         try {
-            Crypto.doSign(privKey, ByteArray(0))
+            Crypto.doSign(privKey, EMPTY_BYTE_ARRAY)
             fail()
         } catch (e: Exception) {
             // expected
@@ -85,7 +86,7 @@ class CryptoUtilsTest {
 
         // test for empty source data when verifying
         try {
-            Crypto.doVerify(pubKey, testBytes, ByteArray(0))
+            Crypto.doVerify(pubKey, testBytes, EMPTY_BYTE_ARRAY)
             fail()
         } catch (e: Exception) {
             // expected
@@ -93,7 +94,7 @@ class CryptoUtilsTest {
 
         // test for empty signed data when verifying
         try {
-            Crypto.doVerify(pubKey, ByteArray(0), testBytes)
+            Crypto.doVerify(pubKey, EMPTY_BYTE_ARRAY, testBytes)
             fail()
         } catch (e: Exception) {
             // expected
@@ -132,7 +133,7 @@ class CryptoUtilsTest {
 
         // test for empty data signing
         try {
-            Crypto.doSign(privKey, ByteArray(0))
+            Crypto.doSign(privKey, EMPTY_BYTE_ARRAY)
             fail()
         } catch (e: Exception) {
             // expected
@@ -140,7 +141,7 @@ class CryptoUtilsTest {
 
         // test for empty source data when verifying
         try {
-            Crypto.doVerify(pubKey, testBytes, ByteArray(0))
+            Crypto.doVerify(pubKey, testBytes, EMPTY_BYTE_ARRAY)
             fail()
         } catch (e: Exception) {
             // expected
@@ -148,7 +149,7 @@ class CryptoUtilsTest {
 
         // test for empty signed data when verifying
         try {
-            Crypto.doVerify(pubKey, ByteArray(0), testBytes)
+            Crypto.doVerify(pubKey, EMPTY_BYTE_ARRAY, testBytes)
             fail()
         } catch (e: Exception) {
             // expected
@@ -187,7 +188,7 @@ class CryptoUtilsTest {
 
         // test for empty data signing
         try {
-            Crypto.doSign(privKey, ByteArray(0))
+            Crypto.doSign(privKey, EMPTY_BYTE_ARRAY)
             fail()
         } catch (e: Exception) {
             // expected
@@ -195,7 +196,7 @@ class CryptoUtilsTest {
 
         // test for empty source data when verifying
         try {
-            Crypto.doVerify(pubKey, testBytes, ByteArray(0))
+            Crypto.doVerify(pubKey, testBytes, EMPTY_BYTE_ARRAY)
             fail()
         } catch (e: Exception) {
             // expected
@@ -203,7 +204,7 @@ class CryptoUtilsTest {
 
         // test for empty signed data when verifying
         try {
-            Crypto.doVerify(pubKey, ByteArray(0), testBytes)
+            Crypto.doVerify(pubKey, EMPTY_BYTE_ARRAY, testBytes)
             fail()
         } catch (e: Exception) {
             // expected
@@ -242,7 +243,7 @@ class CryptoUtilsTest {
 
         // test for empty data signing
         try {
-            Crypto.doSign(privKey, ByteArray(0))
+            Crypto.doSign(privKey, EMPTY_BYTE_ARRAY)
             fail()
         } catch (e: Exception) {
             // expected
@@ -250,7 +251,7 @@ class CryptoUtilsTest {
 
         // test for empty source data when verifying
         try {
-            Crypto.doVerify(pubKey, testBytes, ByteArray(0))
+            Crypto.doVerify(pubKey, testBytes, EMPTY_BYTE_ARRAY)
             fail()
         } catch (e: Exception) {
             // expected
@@ -258,7 +259,7 @@ class CryptoUtilsTest {
 
         // test for empty signed data when verifying
         try {
-            Crypto.doVerify(pubKey, ByteArray(0), testBytes)
+            Crypto.doVerify(pubKey, EMPTY_BYTE_ARRAY, testBytes)
             fail()
         } catch (e: Exception) {
             // expected
@@ -297,7 +298,7 @@ class CryptoUtilsTest {
 
         // test for empty data signing
         try {
-            Crypto.doSign(privKey, ByteArray(0))
+            Crypto.doSign(privKey, EMPTY_BYTE_ARRAY)
             fail()
         } catch (e: Exception) {
             // expected
@@ -305,7 +306,7 @@ class CryptoUtilsTest {
 
         // test for empty source data when verifying
         try {
-            Crypto.doVerify(pubKey, testBytes, ByteArray(0))
+            Crypto.doVerify(pubKey, testBytes, EMPTY_BYTE_ARRAY)
             fail()
         } catch (e: Exception) {
             // expected
@@ -313,7 +314,7 @@ class CryptoUtilsTest {
 
         // test for empty signed data when verifying
         try {
-            Crypto.doVerify(pubKey, ByteArray(0), testBytes)
+            Crypto.doVerify(pubKey, EMPTY_BYTE_ARRAY, testBytes)
             fail()
         } catch (e: Exception) {
             // expected

core/src/test/kotlin/net/corda/core/utilities/ByteArraysTest.kt

@@ -0,0 +1,45 @@
+package net.corda.core.utilities
+
+import net.corda.core.internal.declaredField
+import org.assertj.core.api.Assertions.catchThrowable
+import org.junit.Assert.assertSame
+import org.junit.Test
+import java.nio.ByteBuffer
+import java.nio.ReadOnlyBufferException
+import kotlin.test.assertEquals
+
+class ByteArraysTest {
+    @Test
+    fun `slice works`() {
+        byteArrayOf(9, 9, 0, 1, 2, 3, 4, 9, 9).let {
+            sliceWorksImpl(it, OpaqueBytesSubSequence(it, 2, 5))
+        }
+        byteArrayOf(0, 1, 2, 3, 4).let {
+            sliceWorksImpl(it, OpaqueBytes(it))
+        }
+    }
+
+    private fun sliceWorksImpl(array: ByteArray, seq: ByteSequence) {
+        // Python-style negative indices can be implemented later if needed:
+        assertSame(IllegalArgumentException::class.java, catchThrowable { seq.slice(-1) }.javaClass)
+        assertSame(IllegalArgumentException::class.java, catchThrowable { seq.slice(end = -1) }.javaClass)
+        fun check(expected: ByteArray, actual: ByteBuffer) {
+            assertEquals(ByteBuffer.wrap(expected), actual)
+            assertSame(ReadOnlyBufferException::class.java, catchThrowable { actual.array() }.javaClass)
+            assertSame(array, actual.declaredField<ByteArray>(ByteBuffer::class, "hb").value)
+        }
+        check(byteArrayOf(0, 1, 2, 3, 4), seq.slice())
+        check(byteArrayOf(0, 1, 2, 3, 4), seq.slice(0, 5))
+        check(byteArrayOf(0, 1, 2, 3, 4), seq.slice(0, 6))
+        check(byteArrayOf(0, 1, 2, 3), seq.slice(0, 4))
+        check(byteArrayOf(1, 2, 3), seq.slice(1, 4))
+        check(byteArrayOf(1, 2, 3, 4), seq.slice(1, 5))
+        check(byteArrayOf(1, 2, 3, 4), seq.slice(1, 6))
+        check(byteArrayOf(4), seq.slice(4))
+        check(byteArrayOf(), seq.slice(5))
+        check(byteArrayOf(), seq.slice(6))
+        check(byteArrayOf(2), seq.slice(2, 3))
+        check(byteArrayOf(), seq.slice(2, 2))
+        check(byteArrayOf(), seq.slice(2, 1))
+    }
+}

core/src/test/kotlin/net/corda/core/utilities/EncodingUtilsTest.kt

@@ -1,6 +1,7 @@
 package net.corda.core.utilities
 
 import net.corda.core.crypto.AddressFormatException
+import org.apache.commons.lang.ArrayUtils.EMPTY_BYTE_ARRAY
 import org.junit.Test
 import kotlin.test.assertEquals
 import kotlin.test.fail
@@ -23,10 +24,9 @@ class EncodingUtilsTest {
 
     @Test
     fun `empty encoding`() {
-        val emptyByteArray = ByteArray(0)
+        assertEquals("", EMPTY_BYTE_ARRAY.toBase58())
-        assertEquals("", emptyByteArray.toBase58())
+        assertEquals("", EMPTY_BYTE_ARRAY.toBase64())
-        assertEquals("", emptyByteArray.toBase64())
+        assertEquals("", EMPTY_BYTE_ARRAY.toHex())
-        assertEquals("", emptyByteArray.toHex())
     }
 
     @Test

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

@@ -10,9 +10,9 @@ import net.corda.core.node.NodeInfo
 import net.corda.core.node.NotaryInfo
 import net.corda.core.serialization.SerializationContext
 import net.corda.core.serialization.deserialize
+import net.corda.nodeapi.internal.serialization.CordaSerializationMagic
 import net.corda.core.serialization.internal.SerializationEnvironmentImpl
 import net.corda.core.serialization.internal._contextSerializationEnv
-import net.corda.core.utilities.ByteSequence
 import net.corda.core.utilities.getOrThrow
 import net.corda.core.utilities.seconds
 import net.corda.nodeapi.internal.SignedNodeInfo
@@ -20,7 +20,7 @@ import net.corda.nodeapi.internal.serialization.AMQP_P2P_CONTEXT
 import net.corda.nodeapi.internal.serialization.SerializationFactoryImpl
 import net.corda.nodeapi.internal.serialization.amqp.AMQPServerSerializationScheme
 import net.corda.nodeapi.internal.serialization.kryo.AbstractKryoSerializationScheme
-import net.corda.nodeapi.internal.serialization.kryo.KryoHeaderV0_1
+import net.corda.nodeapi.internal.serialization.kryo.kryoMagic
 import java.nio.file.Files
 import java.nio.file.Path
 import java.nio.file.Paths
@@ -198,8 +198,8 @@ class NetworkBootstrapper {
     }
 
     private object KryoParametersSerializationScheme : AbstractKryoSerializationScheme() {
-        override fun canDeserializeVersion(byteSequence: ByteSequence, target: SerializationContext.UseCase): Boolean {
+        override fun canDeserializeVersion(magic: CordaSerializationMagic, target: SerializationContext.UseCase): Boolean {
-            return byteSequence == KryoHeaderV0_1 && target == SerializationContext.UseCase.P2P
+            return magic == kryoMagic && target == SerializationContext.UseCase.P2P
         }
 
         override fun rpcClientKryoPool(context: SerializationContext) = throw UnsupportedOperationException()

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

@@ -4,8 +4,8 @@ package net.corda.nodeapi.internal.serialization
 
 import net.corda.core.serialization.SerializationContext
 import net.corda.core.serialization.SerializationDefaults
-import net.corda.nodeapi.internal.serialization.amqp.AmqpHeaderV1_0
+import net.corda.nodeapi.internal.serialization.amqp.amqpMagic
-import net.corda.nodeapi.internal.serialization.kryo.KryoHeaderV0_1
+import net.corda.nodeapi.internal.serialization.kryo.kryoMagic
 
 /*
  * Serialisation contexts for the client.
@@ -13,14 +13,13 @@ import net.corda.nodeapi.internal.serialization.kryo.KryoHeaderV0_1
  * servers from trying to instantiate any of them.
  */
 
-val KRYO_RPC_CLIENT_CONTEXT = SerializationContextImpl(KryoHeaderV0_1,
+val KRYO_RPC_CLIENT_CONTEXT = SerializationContextImpl(kryoMagic,
         SerializationDefaults.javaClass.classLoader,
         GlobalTransientClassWhiteList(BuiltInExceptionsWhitelist()),
         emptyMap(),
         true,
         SerializationContext.UseCase.RPCClient)
-
+val AMQP_RPC_CLIENT_CONTEXT = SerializationContextImpl(amqpMagic,
-val AMQP_RPC_CLIENT_CONTEXT = SerializationContextImpl(AmqpHeaderV1_0,
         SerializationDefaults.javaClass.classLoader,
         GlobalTransientClassWhiteList(BuiltInExceptionsWhitelist()),
         emptyMap(),

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

@@ -0,0 +1,12 @@
+package net.corda.nodeapi.internal.serialization
+
+import net.corda.core.utilities.ByteSequence
+import net.corda.core.utilities.OpaqueBytes
+import java.nio.ByteBuffer
+
+class CordaSerializationMagic(bytes: ByteArray) : OpaqueBytes(bytes) {
+    private val bufferView = slice()
+    fun consume(data: ByteSequence): ByteBuffer? {
+        return if (data.slice(end = size) == bufferView) data.slice(size) else null
+    }
+}

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

@@ -4,9 +4,12 @@ import com.google.common.cache.Cache
 import com.google.common.cache.CacheBuilder
 import net.corda.core.contracts.Attachment
 import net.corda.core.crypto.SecureHash
+import net.corda.core.internal.copyBytes
 import net.corda.core.serialization.*
 import net.corda.core.utilities.ByteSequence
 import net.corda.nodeapi.internal.AttachmentsClassLoader
+import net.corda.nodeapi.internal.serialization.amqp.amqpMagic
+import net.corda.nodeapi.internal.serialization.kryo.kryoMagic
 import org.slf4j.LoggerFactory
 import java.io.NotSerializableException
 import java.util.*
@@ -15,17 +18,7 @@ import java.util.concurrent.ExecutionException
 
 val attachmentsClassLoaderEnabledPropertyName = "attachments.class.loader.enabled"
 
-object NotSupportedSerializationScheme : SerializationScheme {
+data class SerializationContextImpl(override val preferredSerializationVersion: SerializationMagic,
-    private fun doThrow(): Nothing = throw UnsupportedOperationException("Serialization scheme not supported.")
-
-    override fun canDeserializeVersion(byteSequence: ByteSequence, target: SerializationContext.UseCase): Boolean = doThrow()
-
-    override fun <T : Any> deserialize(byteSequence: ByteSequence, clazz: Class<T>, context: SerializationContext): T = doThrow()
-
-    override fun <T : Any> serialize(obj: T, context: SerializationContext): SerializedBytes<T> = doThrow()
-}
-
-data class SerializationContextImpl(override val preferredSerializationVersion: VersionHeader,
                                     override val deserializationClassLoader: ClassLoader,
                                     override val whitelist: ClassWhitelist,
                                     override val properties: Map<Any, Any>,
@@ -76,14 +69,14 @@ data class SerializationContextImpl(override val preferredSerializationVersion:
         })
     }
 
-    override fun withPreferredSerializationVersion(versionHeader: VersionHeader) = copy(preferredSerializationVersion = versionHeader)
+    override fun withPreferredSerializationVersion(magic: SerializationMagic) = copy(preferredSerializationVersion = magic)
 }
 
-private const val HEADER_SIZE: Int = 8
-
-fun ByteSequence.obtainHeaderSignature(): VersionHeader = take(HEADER_SIZE).copy()
-
 open class SerializationFactoryImpl : SerializationFactory() {
+    companion object {
+        private val magicSize = sequenceOf(kryoMagic, amqpMagic).map { it.size }.distinct().single()
+    }
+
     private val creator: List<StackTraceElement> = Exception().stackTrace.asList()
 
     private val registeredSchemes: MutableCollection<SerializationScheme> = Collections.synchronizedCollection(mutableListOf())
@@ -91,19 +84,17 @@ open class SerializationFactoryImpl : SerializationFactory() {
     private val logger = LoggerFactory.getLogger(javaClass)
 
     // TODO: This is read-mostly. Probably a faster implementation to be found.
-    private val schemes: ConcurrentHashMap<Pair<ByteSequence, SerializationContext.UseCase>, SerializationScheme> = ConcurrentHashMap()
+    private val schemes: ConcurrentHashMap<Pair<CordaSerializationMagic, SerializationContext.UseCase>, SerializationScheme> = ConcurrentHashMap()
 
-    private fun schemeFor(byteSequence: ByteSequence, target: SerializationContext.UseCase): Pair<SerializationScheme, VersionHeader> {
+    private fun schemeFor(byteSequence: ByteSequence, target: SerializationContext.UseCase): Pair<SerializationScheme, CordaSerializationMagic> {
-        // truncate sequence to 8 bytes, and make sure it's a copy to avoid holding onto large ByteArrays
+        // truncate sequence to at most magicSize, and make sure it's a copy to avoid holding onto large ByteArrays
-        val lookupKey = byteSequence.obtainHeaderSignature() to target
+        val magic = CordaSerializationMagic(byteSequence.slice(end = magicSize).copyBytes())
-        val scheme = schemes.computeIfAbsent(lookupKey) {
+        val lookupKey = magic to target
-            registeredSchemes
+        return schemes.computeIfAbsent(lookupKey) {
-                    .filter { scheme -> scheme.canDeserializeVersion(it.first, it.second) }
+            registeredSchemes.filter { it.canDeserializeVersion(magic, target) }.forEach { return@computeIfAbsent it } // XXX: Not single?
-                    .forEach { return@computeIfAbsent it }
             logger.warn("Cannot find serialization scheme for: $lookupKey, registeredSchemes are: $registeredSchemes")
-            NotSupportedSerializationScheme
+            throw UnsupportedOperationException("Serialization scheme not supported.")
-        }
+        } to magic
-        return scheme to lookupKey.first
     }
 
     @Throws(NotSerializableException::class)
@@ -115,9 +106,9 @@ open class SerializationFactoryImpl : SerializationFactory() {
     override fun <T : Any> deserializeWithCompatibleContext(byteSequence: ByteSequence, clazz: Class<T>, context: SerializationContext): ObjectWithCompatibleContext<T> {
         return asCurrent {
             withCurrentContext(context) {
-                val (scheme, versionHeader) = schemeFor(byteSequence, context.useCase)
+                val (scheme, magic) = schemeFor(byteSequence, context.useCase)
                 val deserializedObject = scheme.deserialize(byteSequence, clazz, context)
-                ObjectWithCompatibleContext(deserializedObject, context.withPreferredSerializationVersion(versionHeader))
+                ObjectWithCompatibleContext(deserializedObject, context.withPreferredSerializationVersion(magic))
             }
         }
     }
@@ -149,9 +140,7 @@ open class SerializationFactoryImpl : SerializationFactory() {
 
 
 interface SerializationScheme {
-    // byteSequence expected to just be the 8 bytes necessary for versioning
+    fun canDeserializeVersion(magic: CordaSerializationMagic, target: SerializationContext.UseCase): Boolean
-    fun canDeserializeVersion(byteSequence: ByteSequence, target: SerializationContext.UseCase): Boolean
-
     @Throws(NotSerializableException::class)
     fun <T : Any> deserialize(byteSequence: ByteSequence, clazz: Class<T>, context: SerializationContext): T
 

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

@@ -5,8 +5,8 @@ package net.corda.nodeapi.internal.serialization
 import net.corda.core.serialization.ClassWhitelist
 import net.corda.core.serialization.SerializationContext
 import net.corda.core.serialization.SerializationDefaults
-import net.corda.nodeapi.internal.serialization.amqp.AmqpHeaderV1_0
+import net.corda.nodeapi.internal.serialization.amqp.amqpMagic
-import net.corda.nodeapi.internal.serialization.kryo.KryoHeaderV0_1
+import net.corda.nodeapi.internal.serialization.kryo.kryoMagic
 
 object QuasarWhitelist : ClassWhitelist {
     override fun hasListed(type: Class<*>): Boolean = true
@@ -22,29 +22,25 @@ object QuasarWhitelist : ClassWhitelist {
  * MUST be kept separate!
  */
 
-val KRYO_RPC_SERVER_CONTEXT = SerializationContextImpl(KryoHeaderV0_1,
+val KRYO_RPC_SERVER_CONTEXT = SerializationContextImpl(kryoMagic,
         SerializationDefaults.javaClass.classLoader,
         GlobalTransientClassWhiteList(BuiltInExceptionsWhitelist()),
         emptyMap(),
         true,
         SerializationContext.UseCase.RPCServer)
-
+val KRYO_STORAGE_CONTEXT = SerializationContextImpl(kryoMagic,
-val KRYO_STORAGE_CONTEXT = SerializationContextImpl(KryoHeaderV0_1,
         SerializationDefaults.javaClass.classLoader,
         AllButBlacklisted,
         emptyMap(),
         true,
         SerializationContext.UseCase.Storage)
-
+val AMQP_STORAGE_CONTEXT = SerializationContextImpl(amqpMagic,
-
-val AMQP_STORAGE_CONTEXT = SerializationContextImpl(AmqpHeaderV1_0,
         SerializationDefaults.javaClass.classLoader,
         AllButBlacklisted,
         emptyMap(),
         true,
         SerializationContext.UseCase.Storage)
-
+val AMQP_RPC_SERVER_CONTEXT = SerializationContextImpl(amqpMagic,
-val AMQP_RPC_SERVER_CONTEXT = SerializationContextImpl(AmqpHeaderV1_0,
         SerializationDefaults.javaClass.classLoader,
         GlobalTransientClassWhiteList(BuiltInExceptionsWhitelist()),
         emptyMap(),

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

@@ -4,8 +4,8 @@ package net.corda.nodeapi.internal.serialization
 
 import net.corda.core.serialization.SerializationContext
 import net.corda.core.serialization.SerializationDefaults
-import net.corda.nodeapi.internal.serialization.amqp.AmqpHeaderV1_0
+import net.corda.nodeapi.internal.serialization.amqp.amqpMagic
-import net.corda.nodeapi.internal.serialization.kryo.KryoHeaderV0_1
+import net.corda.nodeapi.internal.serialization.kryo.kryoMagic
 
 /*
  * Serialisation contexts shared by the server and client.
@@ -15,21 +15,19 @@ import net.corda.nodeapi.internal.serialization.kryo.KryoHeaderV0_1
  * MUST be kept separate from these ones!
  */
 
-val KRYO_P2P_CONTEXT = SerializationContextImpl(KryoHeaderV0_1,
+val KRYO_P2P_CONTEXT = SerializationContextImpl(kryoMagic,
         SerializationDefaults.javaClass.classLoader,
         GlobalTransientClassWhiteList(BuiltInExceptionsWhitelist()),
         emptyMap(),
         true,
         SerializationContext.UseCase.P2P)
-
+val KRYO_CHECKPOINT_CONTEXT = SerializationContextImpl(kryoMagic,
-val KRYO_CHECKPOINT_CONTEXT = SerializationContextImpl(KryoHeaderV0_1,
         SerializationDefaults.javaClass.classLoader,
         QuasarWhitelist,
         emptyMap(),
         true,
         SerializationContext.UseCase.Checkpoint)
-
+val AMQP_P2P_CONTEXT = SerializationContextImpl(amqpMagic,
-val AMQP_P2P_CONTEXT = SerializationContextImpl(AmqpHeaderV1_0,
         SerializationDefaults.javaClass.classLoader,
         GlobalTransientClassWhiteList(BuiltInExceptionsWhitelist()),
         emptyMap(),

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

@@ -4,6 +4,7 @@ package net.corda.nodeapi.internal.serialization.amqp
 
 import net.corda.core.cordapp.Cordapp
 import net.corda.core.serialization.*
+import net.corda.nodeapi.internal.serialization.CordaSerializationMagic
 import net.corda.core.utilities.ByteSequence
 import net.corda.nodeapi.internal.serialization.DefaultWhitelist
 import net.corda.nodeapi.internal.serialization.MutableClassWhitelist
@@ -12,7 +13,7 @@ import java.security.PublicKey
 import java.util.*
 import java.util.concurrent.ConcurrentHashMap
 
-val AMQP_ENABLED get() = SerializationDefaults.P2P_CONTEXT.preferredSerializationVersion == AmqpHeaderV1_0
+val AMQP_ENABLED get() = SerializationDefaults.P2P_CONTEXT.preferredSerializationVersion == amqpMagic
 
 fun SerializerFactory.addToWhitelist(vararg types: Class<*>) {
     require(types.toSet().size == types.size) {
@@ -106,7 +107,7 @@ abstract class AbstractAMQPSerializationScheme(val cordappLoader: List<Cordapp>)
         return SerializationOutput(serializerFactory).serialize(obj)
     }
 
-    protected fun canDeserializeVersion(byteSequence: ByteSequence): Boolean = byteSequence == AmqpHeaderV1_0
+    protected fun canDeserializeVersion(magic: CordaSerializationMagic) = magic == amqpMagic
 }
 
 // TODO: This will eventually cover server RPC as well and move to node module, but for now this is not implemented
@@ -119,9 +120,9 @@ class AMQPServerSerializationScheme(cordapps: List<Cordapp> = emptyList()) : Abs
         TODO("not implemented") //To change body of created functions use File | Settings | File Templates.
     }
 
-    override fun canDeserializeVersion(byteSequence: ByteSequence, target: SerializationContext.UseCase): Boolean {
+    override fun canDeserializeVersion(magic: CordaSerializationMagic, target: SerializationContext.UseCase): Boolean {
-        return (canDeserializeVersion(byteSequence) &&
+        return canDeserializeVersion(magic) &&
-                (target == SerializationContext.UseCase.P2P || target == SerializationContext.UseCase.Storage))
+                (target == SerializationContext.UseCase.P2P || target == SerializationContext.UseCase.Storage)
     }
 
 }
@@ -136,9 +137,9 @@ class AMQPClientSerializationScheme(cordapps: List<Cordapp> = emptyList()) : Abs
         throw UnsupportedOperationException()
     }
 
-    override fun canDeserializeVersion(byteSequence: ByteSequence, target: SerializationContext.UseCase): Boolean {
+    override fun canDeserializeVersion(magic: CordaSerializationMagic, target: SerializationContext.UseCase): Boolean {
-        return (canDeserializeVersion(byteSequence) &&
+        return canDeserializeVersion(magic) &&
-                (target == SerializationContext.UseCase.P2P || target == SerializationContext.UseCase.Storage))
+                (target == SerializationContext.UseCase.P2P || target == SerializationContext.UseCase.Storage)
     }
 
 }

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

@@ -13,7 +13,6 @@ import java.lang.reflect.ParameterizedType
 import java.lang.reflect.Type
 import java.lang.reflect.TypeVariable
 import java.lang.reflect.WildcardType
-import java.nio.ByteBuffer
 
 data class ObjectAndEnvelope<out T>(val obj: T, val envelope: Envelope)
 
@@ -58,19 +57,12 @@ class DeserializationInput(internal val serializerFactory: SerializerFactory) {
             deserializeAndReturnEnvelope(bytes, T::class.java)
 
     @Throws(NotSerializableException::class)
-    internal fun getEnvelope(bytes: ByteSequence): Envelope {
+    internal fun getEnvelope(byteSequence: ByteSequence): Envelope {
         // Check that the lead bytes match expected header
-        val headerSize = AmqpHeaderV1_0.size
+        val dataBytes = amqpMagic.consume(byteSequence) ?: throw NotSerializableException("Serialization header does not match.")
-        if (bytes.take(headerSize) != AmqpHeaderV1_0) {
-            throw NotSerializableException("Serialization header does not match.")
-        }
-
         val data = Data.Factory.create()
-        val size = data.decode(ByteBuffer.wrap(bytes.bytes, bytes.offset + headerSize, bytes.size - headerSize))
+        val expectedSize = dataBytes.remaining()
-        if (size.toInt() != bytes.size - headerSize) {
+        if (data.decode(dataBytes) != expectedSize.toLong()) throw NotSerializableException("Unexpected size of data")
-            throw NotSerializableException("Unexpected size of data")
-        }
-
         return Envelope.get(data)
     }
 

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

@@ -3,7 +3,7 @@ package net.corda.nodeapi.internal.serialization.amqp
 import com.google.common.hash.Hasher
 import com.google.common.hash.Hashing
 import net.corda.core.internal.uncheckedCast
-import net.corda.core.utilities.OpaqueBytes
+import net.corda.nodeapi.internal.serialization.CordaSerializationMagic
 import net.corda.core.utilities.loggerFor
 import net.corda.core.utilities.toBase64
 import org.apache.qpid.proton.amqp.DescribedType
@@ -18,9 +18,7 @@ import net.corda.nodeapi.internal.serialization.carpenter.Field as CarpenterFiel
 import net.corda.nodeapi.internal.serialization.carpenter.Schema as CarpenterSchema
 
 const val DESCRIPTOR_DOMAIN: String = "net.corda"
-
+val amqpMagic = CordaSerializationMagic("corda".toByteArray() + byteArrayOf(1, 0, 0))
-// "corda" + majorVersionByte + minorVersionMSB + minorVersionLSB
-val AmqpHeaderV1_0: OpaqueBytes = OpaqueBytes("corda\u0001\u0000\u0000".toByteArray())
 
 /**
  * This and the classes below are OO representations of the AMQP XML schema described in the specification. Their

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

@@ -69,7 +69,7 @@ open class SerializationOutput(internal val serializerFactory: SerializerFactory
         }
         val bytes = ByteArray(data.encodedSize().toInt() + 8)
         val buf = ByteBuffer.wrap(bytes)
-        buf.put(AmqpHeaderV1_0.bytes)
+        amqpMagic.putTo(buf)
         data.encode(buf)
         return SerializedBytes(bytes)
     }

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

@@ -13,8 +13,6 @@ import net.corda.nodeapi.internal.serialization.amqp.SerializerFactory
 class OpaqueBytesSubSequenceSerializer(factory: SerializerFactory) :
         CustomSerializer.Proxy<OpaqueBytesSubSequence, OpaqueBytes>(OpaqueBytesSubSequence::class.java, OpaqueBytes::class.java, factory) {
     override val additionalSerializers: Iterable<CustomSerializer<out Any>> = emptyList()
-
+    override fun toProxy(obj: OpaqueBytesSubSequence): OpaqueBytes = OpaqueBytes(obj.copyBytes())
-    override fun toProxy(obj: OpaqueBytesSubSequence): OpaqueBytes = OpaqueBytes(obj.copy().bytes)
-
     override fun fromProxy(proxy: OpaqueBytes): OpaqueBytesSubSequence = OpaqueBytesSubSequence(proxy.bytes, proxy.offset, proxy.size)
 }

node-api/src/main/kotlin/net/corda/nodeapi/internal/serialization/kryo/DefaultKryoCustomizer.kt

@@ -44,7 +44,6 @@ import org.objenesis.strategy.StdInstantiatorStrategy
 import org.slf4j.Logger
 import sun.security.ec.ECPublicKeyImpl
 import sun.security.provider.certpath.X509CertPath
-import sun.security.x509.X509CertImpl
 import java.io.BufferedInputStream
 import java.io.ByteArrayOutputStream
 import java.io.FileInputStream
@@ -200,7 +199,7 @@ object DefaultKryoCustomizer {
     private object ContractAttachmentSerializer : Serializer<ContractAttachment>() {
         override fun write(kryo: Kryo, output: Output, obj: ContractAttachment) {
             if (kryo.serializationContext() != null) {
-                output.writeBytes(obj.attachment.id.bytes)
+                obj.attachment.id.writeTo(output)
             } else {
                 val buffer = ByteArrayOutputStream()
                 obj.attachment.open().use { it.copyTo(buffer) }

node-api/src/main/kotlin/net/corda/nodeapi/internal/serialization/kryo/Kryo.kt

@@ -84,8 +84,8 @@ import kotlin.reflect.jvm.javaType
  */
 object SerializedBytesSerializer : Serializer<SerializedBytes<Any>>() {
     override fun write(kryo: Kryo, output: Output, obj: SerializedBytes<Any>) {
-        output.writeVarInt(obj.bytes.size, true)
+        output.writeVarInt(obj.size, true)
-        output.writeBytes(obj.bytes)
+        obj.writeTo(output)
     }
 
     override fun read(kryo: Kryo, input: Input, type: Class<SerializedBytes<Any>>): SerializedBytes<Any> {
@@ -186,7 +186,7 @@ object InputStreamSerializer : Serializer<InputStream>() {
                 output.writeInt(numberOfBytesRead, true)
                 output.writeBytes(buffer, 0, numberOfBytesRead)
             } else {
-                output.writeInt(0)
+                output.writeInt(0, true)
                 break
             }
         }

node-api/src/main/kotlin/net/corda/nodeapi/internal/serialization/kryo/KryoSerializationScheme.kt

@@ -1,27 +1,25 @@
 package net.corda.nodeapi.internal.serialization.kryo
 
 import java.util.concurrent.ConcurrentHashMap
-import java.io.ByteArrayOutputStream
 import co.paralleluniverse.fibers.Fiber
 import co.paralleluniverse.io.serialization.kryo.KryoSerializer
 import com.esotericsoftware.kryo.Kryo
 import com.esotericsoftware.kryo.KryoException
 import com.esotericsoftware.kryo.Serializer
+import com.esotericsoftware.kryo.io.ByteBufferInputStream
 import com.esotericsoftware.kryo.io.Input
 import com.esotericsoftware.kryo.io.Output
 import com.esotericsoftware.kryo.pool.KryoPool
 import com.esotericsoftware.kryo.serializers.ClosureSerializer
 import net.corda.core.internal.uncheckedCast
-import net.corda.core.utilities.OpaqueBytes
 import net.corda.core.utilities.ByteSequence
 import net.corda.core.serialization.*
-import net.corda.core.internal.LazyPool
+import net.corda.nodeapi.internal.serialization.CordaSerializationMagic
 import net.corda.nodeapi.internal.serialization.CordaClassResolver
 import net.corda.nodeapi.internal.serialization.SerializationScheme
 import java.security.PublicKey
 
-// "corda" + majorVersionByte + minorVersionMSB + minorVersionLSB
+val kryoMagic = CordaSerializationMagic("corda".toByteArray() + byteArrayOf(0, 0, 1))
-val KryoHeaderV0_1: OpaqueBytes = OpaqueBytes("corda\u0000\u0000\u0001".toByteArray(Charsets.UTF_8))
 
 private object AutoCloseableSerialisationDetector : Serializer<AutoCloseable>() {
     override fun write(kryo: Kryo, output: Output, closeable: AutoCloseable) {
@@ -73,65 +71,41 @@ abstract class AbstractKryoSerializationScheme : SerializationScheme {
         }
     }
 
-    private fun <T : Any> withContext(kryo: Kryo, context: SerializationContext, block: (Kryo) -> T): T {
+    private fun <T : Any> SerializationContext.kryo(task: Kryo.() -> T): T {
-        kryo.context.ensureCapacity(context.properties.size)
+        return getPool(this).run { kryo ->
-        context.properties.forEach { kryo.context.put(it.key, it.value) }
+            kryo.context.ensureCapacity(properties.size)
+            properties.forEach { kryo.context.put(it.key, it.value) }
             try {
-            return block(kryo)
+                kryo.task()
             } finally {
                 kryo.context.clear()
             }
         }
+    }
 
     override fun <T : Any> deserialize(byteSequence: ByteSequence, clazz: Class<T>, context: SerializationContext): T {
-        val pool = getPool(context)
+        val dataBytes = kryoMagic.consume(byteSequence) ?: throw KryoException("Serialized bytes header does not match expected format.")
-        val headerSize = KryoHeaderV0_1.size
+        return context.kryo {
-        val header = byteSequence.take(headerSize)
+            kryoInput(ByteBufferInputStream(dataBytes)) {
-        if (header != KryoHeaderV0_1) {
-            throw KryoException("Serialized bytes header does not match expected format.")
-        }
-        Input(byteSequence.bytes, byteSequence.offset + headerSize, byteSequence.size - headerSize).use { input ->
-            return pool.run { kryo ->
-                withContext(kryo, context) {
                 if (context.objectReferencesEnabled) {
-                        uncheckedCast(kryo.readClassAndObject(input))
+                    uncheckedCast(readClassAndObject(this))
                 } else {
-                        kryo.withoutReferences { uncheckedCast<Any?, T>(kryo.readClassAndObject(input)) }
+                    withoutReferences { uncheckedCast<Any?, T>(readClassAndObject(this)) }
-                    }
                 }
             }
         }
     }
 
     override fun <T : Any> serialize(obj: T, context: SerializationContext): SerializedBytes<T> {
-        val pool = getPool(context)
+        return context.kryo {
-        return pool.run { kryo ->
+            SerializedBytes(kryoOutput {
-            withContext(kryo, context) {
+                kryoMagic.writeTo(this)
-                serializeOutputStreamPool.run { stream ->
-                    serializeBufferPool.run { buffer ->
-                        Output(buffer).use {
-                            it.outputStream = stream
-                            it.writeBytes(KryoHeaderV0_1.bytes)
                 if (context.objectReferencesEnabled) {
-                                kryo.writeClassAndObject(it, obj)
+                    writeClassAndObject(this, obj)
                 } else {
-                                kryo.withoutReferences { kryo.writeClassAndObject(it, obj) }
+                    withoutReferences { writeClassAndObject(this, obj) }
-                            }
-                        }
-                        SerializedBytes(stream.toByteArray())
-                    }
-                }
                 }
+            })
         }
     }
 }
-
-private val serializeBufferPool = LazyPool(
-        newInstance = { ByteArray(64 * 1024) }
-)
-
-private val serializeOutputStreamPool = LazyPool(
-        clear = ByteArrayOutputStream::reset,
-        shouldReturnToPool = { it.size() < 256 * 1024 }, // Discard if it grew too large
-        newInstance = { ByteArrayOutputStream(64 * 1024) }
-)

node-api/src/main/kotlin/net/corda/nodeapi/internal/serialization/kryo/KryoStreams.kt

@@ -0,0 +1,43 @@
+package net.corda.nodeapi.internal.serialization.kryo
+
+import com.esotericsoftware.kryo.io.Input
+import com.esotericsoftware.kryo.io.Output
+import net.corda.core.internal.LazyPool
+import java.io.*
+
+private val serializationBufferPool = LazyPool(
+        newInstance = { ByteArray(64 * 1024) })
+private val serializeOutputStreamPool = LazyPool(
+        clear = ByteArrayOutputStream::reset,
+        shouldReturnToPool = { it.size() < 256 * 1024 }, // Discard if it grew too large
+        newInstance = { ByteArrayOutputStream(64 * 1024) })
+
+internal fun <T> kryoInput(underlying: InputStream, task: Input.() -> T): T {
+    return serializationBufferPool.run {
+        Input(it).use { input ->
+            input.inputStream = underlying
+            input.task()
+        }
+    }
+}
+
+internal fun <T> kryoOutput(task: Output.() -> T): ByteArray {
+    return serializeOutputStreamPool.run { underlying ->
+        serializationBufferPool.run {
+            Output(it).use { output ->
+                output.outputStream = underlying
+                output.task()
+            }
+        }
+        underlying.toByteArray() // Must happen after close, to allow ZIP footer to be written for example.
+    }
+}
+
+internal fun Output.substitute(transform: (OutputStream) -> OutputStream) {
+    flush()
+    outputStream = transform(outputStream)
+}
+
+internal fun Input.substitute(transform: (InputStream) -> InputStream) {
+    inputStream = transform(SequenceInputStream(ByteArrayInputStream(buffer.copyOfRange(position(), limit())), inputStream))
+}

node-api/src/test/java/net/corda/nodeapi/internal/serialization/ForbiddenLambdaSerializationTests.java

@@ -33,8 +33,7 @@ public final class ForbiddenLambdaSerializationTests {
         EnumSet<SerializationContext.UseCase> contexts = EnumSet.complementOf(EnumSet.of(SerializationContext.UseCase.Checkpoint));
 
         contexts.forEach(ctx -> {
-            SerializationContext context = new SerializationContextImpl(KryoSerializationSchemeKt.getKryoHeaderV0_1(), this.getClass().getClassLoader(), AllWhitelist.INSTANCE, Maps.newHashMap(), true, ctx);
+            SerializationContext context = new SerializationContextImpl(KryoSerializationSchemeKt.getKryoMagic(), this.getClass().getClassLoader(), AllWhitelist.INSTANCE, Maps.newHashMap(), true, ctx);
-
             String value = "Hey";
             Callable<String> target = (Callable<String> & Serializable) () -> value;
 
@@ -56,8 +55,7 @@ public final class ForbiddenLambdaSerializationTests {
         EnumSet<SerializationContext.UseCase> contexts = EnumSet.complementOf(EnumSet.of(SerializationContext.UseCase.Checkpoint));
 
         contexts.forEach(ctx -> {
-            SerializationContext context = new SerializationContextImpl(KryoSerializationSchemeKt.getKryoHeaderV0_1(), this.getClass().getClassLoader(), AllWhitelist.INSTANCE, Maps.newHashMap(), true, ctx);
+            SerializationContext context = new SerializationContextImpl(KryoSerializationSchemeKt.getKryoMagic(), this.getClass().getClassLoader(), AllWhitelist.INSTANCE, Maps.newHashMap(), true, ctx);
-
             String value = "Hey";
             Callable<String> target = () -> value;
 

node-api/src/test/java/net/corda/nodeapi/internal/serialization/LambdaCheckpointSerializationTest.java

@@ -26,7 +26,7 @@ public final class LambdaCheckpointSerializationTest {
     @Before
     public void setup() {
         factory = testSerialization.getSerializationFactory();
-        context = new SerializationContextImpl(KryoSerializationSchemeKt.getKryoHeaderV0_1(), this.getClass().getClassLoader(), AllWhitelist.INSTANCE, Maps.newHashMap(), true, SerializationContext.UseCase.Checkpoint);
+        context = new SerializationContextImpl(KryoSerializationSchemeKt.getKryoMagic(), this.getClass().getClassLoader(), AllWhitelist.INSTANCE, Maps.newHashMap(), true, SerializationContext.UseCase.Checkpoint);
     }
 
     @Test

node-api/src/test/kotlin/net/corda/nodeapi/internal/crypto/X509UtilitiesTest.kt

@@ -14,7 +14,7 @@ import net.corda.nodeapi.internal.createDevKeyStores
 import net.corda.nodeapi.internal.serialization.AllWhitelist
 import net.corda.nodeapi.internal.serialization.SerializationContextImpl
 import net.corda.nodeapi.internal.serialization.SerializationFactoryImpl
-import net.corda.nodeapi.internal.serialization.kryo.KryoHeaderV0_1
+import net.corda.nodeapi.internal.serialization.kryo.kryoMagic
 import net.corda.testing.core.ALICE_NAME
 import net.corda.testing.core.BOB_NAME
 import net.corda.testing.core.TestIdentity
@@ -314,7 +314,7 @@ class X509UtilitiesTest {
     @Test
     fun `serialize - deserialize X509Certififcate`() {
         val factory = SerializationFactoryImpl().apply { registerScheme(KryoServerSerializationScheme()) }
-        val context = SerializationContextImpl(KryoHeaderV0_1,
+        val context = SerializationContextImpl(kryoMagic,
                 javaClass.classLoader,
                 AllWhitelist,
                 emptyMap(),
@@ -329,7 +329,7 @@ class X509UtilitiesTest {
     @Test
     fun `serialize - deserialize X509CertPath`() {
         val factory = SerializationFactoryImpl().apply { registerScheme(KryoServerSerializationScheme()) }
-        val context = SerializationContextImpl(KryoHeaderV0_1,
+        val context = SerializationContextImpl(kryoMagic,
                 javaClass.classLoader,
                 AllWhitelist,
                 emptyMap(),

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

@@ -7,6 +7,7 @@ import net.corda.testing.contracts.DummyContract
 import net.corda.testing.core.SerializationEnvironmentRule
 import net.corda.testing.internal.rigorousMock
 import net.corda.testing.node.MockServices
+import org.apache.commons.lang.ArrayUtils.EMPTY_BYTE_ARRAY
 import org.assertj.core.api.Assertions.assertThat
 import org.assertj.core.api.Assertions.assertThatThrownBy
 import org.junit.Assert.assertArrayEquals
@@ -35,7 +36,7 @@ class ContractAttachmentSerializerTest {
 
     @Test
     fun `write contract attachment and read it back`() {
-        val contractAttachment = ContractAttachment(GeneratedAttachment(ByteArray(0)), DummyContract.PROGRAM_ID)
+        val contractAttachment = ContractAttachment(GeneratedAttachment(EMPTY_BYTE_ARRAY), DummyContract.PROGRAM_ID)
         // no token context so will serialize the whole attachment
         val serialized = contractAttachment.serialize(factory, context)
         val deserialized = serialized.deserialize(factory, context)
@@ -88,8 +89,7 @@ class ContractAttachmentSerializerTest {
 
     @Test
     fun `check attachment in deserialize is lazy loaded when using token context`() {
-        val attachment = GeneratedAttachment(ByteArray(0))
+        val attachment = GeneratedAttachment(EMPTY_BYTE_ARRAY)
-
         // don't importAttachment in mockService
 
         val contractAttachment = ContractAttachment(attachment, DummyContract.PROGRAM_ID)

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

@@ -11,7 +11,7 @@ import net.corda.core.serialization.*
 import net.corda.nodeapi.internal.AttachmentsClassLoader
 import net.corda.nodeapi.internal.AttachmentsClassLoaderTests
 import net.corda.nodeapi.internal.serialization.kryo.CordaKryo
-import net.corda.nodeapi.internal.serialization.kryo.KryoHeaderV0_1
+import net.corda.nodeapi.internal.serialization.kryo.kryoMagic
 import net.corda.testing.services.MockAttachmentStorage
 import net.corda.testing.internal.rigorousMock
 import org.junit.Rule
@@ -108,9 +108,8 @@ class CordaClassResolverTests {
         val emptyMapClass = mapOf<Any, Any>().javaClass
     }
 
-    private val emptyWhitelistContext: SerializationContext = SerializationContextImpl(KryoHeaderV0_1, this.javaClass.classLoader, EmptyWhitelist, emptyMap(), true, SerializationContext.UseCase.P2P)
+    private val emptyWhitelistContext: SerializationContext = SerializationContextImpl(kryoMagic, this.javaClass.classLoader, EmptyWhitelist, emptyMap(), true, SerializationContext.UseCase.P2P)
-    private val allButBlacklistedContext: SerializationContext = SerializationContextImpl(KryoHeaderV0_1, this.javaClass.classLoader, AllButBlacklisted, emptyMap(), true, SerializationContext.UseCase.P2P)
+    private val allButBlacklistedContext: SerializationContext = SerializationContextImpl(kryoMagic, this.javaClass.classLoader, AllButBlacklisted, emptyMap(), true, SerializationContext.UseCase.P2P)
-
     @Test
     fun `Annotation on enum works for specialised entries`() {
         CordaClassResolver(emptyWhitelistContext).getRegistration(Foo.Bar::class.java)

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

@@ -13,14 +13,13 @@ import net.corda.core.utilities.ProgressTracker
 import net.corda.core.utilities.sequence
 import net.corda.node.serialization.KryoServerSerializationScheme
 import net.corda.node.services.persistence.NodeAttachmentService
-import net.corda.nodeapi.internal.serialization.kryo.KryoHeaderV0_1
+import net.corda.nodeapi.internal.serialization.kryo.kryoMagic
 import net.corda.testing.core.ALICE_NAME
-import net.corda.testing.core.SerializationEnvironmentRule
 import net.corda.testing.core.TestIdentity
 import org.assertj.core.api.Assertions.assertThat
 import org.assertj.core.api.Assertions.assertThatThrownBy
+import org.junit.Assert.assertArrayEquals
 import org.junit.Before
-import org.junit.Rule
 import org.junit.Test
 import org.slf4j.LoggerFactory
 import java.io.ByteArrayInputStream
@@ -28,6 +27,7 @@ import java.io.InputStream
 import java.time.Instant
 import java.util.*
 import kotlin.test.assertEquals
+import kotlin.test.assertFalse
 import kotlin.test.assertNotNull
 import kotlin.test.assertTrue
 
@@ -36,16 +36,13 @@ class KryoTests {
         private val ALICE_PUBKEY = TestIdentity(ALICE_NAME, 70).publicKey
     }
 
-    @Rule
-    @JvmField
-    val testSerialization = SerializationEnvironmentRule()
     private lateinit var factory: SerializationFactory
     private lateinit var context: SerializationContext
 
     @Before
     fun setup() {
         factory = SerializationFactoryImpl().apply { registerScheme(KryoServerSerializationScheme()) }
-        context = SerializationContextImpl(KryoHeaderV0_1,
+        context = SerializationContextImpl(kryoMagic,
                 javaClass.classLoader,
                 AllWhitelist,
                 emptyMap(),
@@ -150,6 +147,14 @@ class KryoTests {
         assertEquals(-1, readRubbishStream.read())
     }
 
+    @Test
+    fun `InputStream serialisation does not write trailing garbage`() {
+        val byteArrays = listOf("123", "456").map { it.toByteArray() }
+        val streams = byteArrays.map { it.inputStream() }.serialize(factory, context).deserialize(factory, context).iterator()
+        byteArrays.forEach { assertArrayEquals(it, streams.next().readBytes()) }
+        assertFalse(streams.hasNext())
+    }
+
     @Test
     fun `serialize - deserialize SignableData`() {
         val testString = "Hello World"
@@ -249,7 +254,7 @@ class KryoTests {
         }
         Tmp()
         val factory = SerializationFactoryImpl().apply { registerScheme(KryoServerSerializationScheme()) }
-        val context = SerializationContextImpl(KryoHeaderV0_1,
+        val context = SerializationContextImpl(kryoMagic,
                 javaClass.classLoader,
                 AllWhitelist,
                 emptyMap(),

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

@@ -7,7 +7,7 @@ import net.corda.node.services.statemachine.SessionData
 import net.corda.nodeapi.internal.serialization.amqp.DeserializationInput
 import net.corda.nodeapi.internal.serialization.amqp.Envelope
 import net.corda.nodeapi.internal.serialization.amqp.SerializerFactory
-import net.corda.nodeapi.internal.serialization.kryo.KryoHeaderV0_1
+import net.corda.nodeapi.internal.serialization.kryo.kryoMagic
 import net.corda.testing.internal.amqpSpecific
 import net.corda.testing.internal.kryoSpecific
 import net.corda.testing.core.SerializationEnvironmentRule
@@ -68,7 +68,7 @@ class ListsSerializationTest {
         val nameID = 0
         val serializedForm = emptyList<Int>().serialize()
         val output = ByteArrayOutputStream().apply {
-            write(KryoHeaderV0_1.bytes)
+            kryoMagic.writeTo(this)
             write(DefaultClassResolver.NAME + 2)
             write(nameID)
             write(javaEmptyListClass.name.toAscii())

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

@@ -7,7 +7,7 @@ import net.corda.core.serialization.CordaSerializable
 import net.corda.core.serialization.deserialize
 import net.corda.core.serialization.serialize
 import net.corda.node.services.statemachine.SessionData
-import net.corda.nodeapi.internal.serialization.kryo.KryoHeaderV0_1
+import net.corda.nodeapi.internal.serialization.kryo.kryoMagic
 import net.corda.testing.core.SerializationEnvironmentRule
 import net.corda.testing.internal.amqpSpecific
 import net.corda.testing.internal.kryoSpecific
@@ -78,7 +78,7 @@ class MapsSerializationTest {
             val nameID = 0
             val serializedForm = emptyMap<Int, Int>().serialize()
             val output = ByteArrayOutputStream().apply {
-                write(KryoHeaderV0_1.bytes)
+                kryoMagic.writeTo(this)
                 write(DefaultClassResolver.NAME + 2)
                 write(nameID)
                 write(javaEmptyMapClass.name.toAscii())

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

@@ -7,7 +7,7 @@ import net.corda.core.serialization.*
 import net.corda.core.utilities.OpaqueBytes
 import net.corda.nodeapi.internal.serialization.kryo.CordaKryo
 import net.corda.nodeapi.internal.serialization.kryo.DefaultKryoCustomizer
-import net.corda.nodeapi.internal.serialization.kryo.KryoHeaderV0_1
+import net.corda.nodeapi.internal.serialization.kryo.kryoMagic
 import net.corda.testing.internal.rigorousMock
 import net.corda.testing.core.SerializationEnvironmentRule
 import org.assertj.core.api.Assertions.assertThat
@@ -98,7 +98,7 @@ class SerializationTokenTest {
         val kryo: Kryo = DefaultKryoCustomizer.customize(CordaKryo(CordaClassResolver(this.context)))
         val stream = ByteArrayOutputStream()
         Output(stream).use {
-            it.write(KryoHeaderV0_1.bytes)
+            kryoMagic.writeTo(it)
             kryo.writeClass(it, SingletonSerializeAsToken::class.java)
             kryo.writeObject(it, emptyList<Any>())
         }

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

@@ -5,7 +5,7 @@ import com.esotericsoftware.kryo.util.DefaultClassResolver
 import net.corda.core.serialization.deserialize
 import net.corda.core.serialization.serialize
 import net.corda.node.services.statemachine.SessionData
-import net.corda.nodeapi.internal.serialization.kryo.KryoHeaderV0_1
+import net.corda.nodeapi.internal.serialization.kryo.kryoMagic
 import net.corda.testing.internal.kryoSpecific
 import net.corda.testing.core.SerializationEnvironmentRule
 import org.junit.Assert.assertArrayEquals
@@ -55,7 +55,7 @@ class SetsSerializationTest {
         val nameID = 0
         val serializedForm = emptySet<Int>().serialize()
         val output = ByteArrayOutputStream().apply {
-            write(KryoHeaderV0_1.bytes)
+            kryoMagic.writeTo(this)
             write(DefaultClassResolver.NAME + 2)
             write(nameID)
             write(javaEmptySetClass.name.toAscii())

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

@@ -1,7 +1,7 @@
 package net.corda.nodeapi.internal.serialization.amqp
 
 import net.corda.core.serialization.SerializationContext
-import net.corda.core.utilities.ByteSequence
+import net.corda.nodeapi.internal.serialization.CordaSerializationMagic
 import net.corda.nodeapi.internal.serialization.AMQP_P2P_CONTEXT
 import org.apache.qpid.proton.codec.Data
 import org.assertj.core.api.Assertions
@@ -30,8 +30,7 @@ class OverridePKSerializerTest {
             TODO("not implemented") //To change body of created functions use File | Settings | File Templates.
         }
 
-        override fun canDeserializeVersion(byteSequence: ByteSequence, target: SerializationContext.UseCase): Boolean = true
+        override fun canDeserializeVersion(magic: CordaSerializationMagic, target: SerializationContext.UseCase) = true
-
         override fun rpcClientSerializerFactory(context: SerializationContext): SerializerFactory {
             TODO("not implemented") //To change body of created functions use File | Settings | File Templates.
         }

node-api/src/test/kotlin/net/corda/nodeapi/internal/serialization/kryo/KryoStreamsTest.kt

@@ -0,0 +1,60 @@
+package net.corda.nodeapi.internal.serialization.kryo
+
+import org.junit.Assert.assertArrayEquals
+import org.junit.Test
+import java.io.*
+import java.util.*
+import java.util.zip.DeflaterOutputStream
+import java.util.zip.InflaterInputStream
+import kotlin.test.assertEquals
+
+class KryoStreamsTest {
+    class NegOutputStream(private val stream: OutputStream) : OutputStream() {
+        override fun write(b: Int) = stream.write(-b)
+    }
+
+    class NegInputStream(private val stream: InputStream) : InputStream() {
+        override fun read() = stream.read().let {
+            if (it != -1) 0xff and -it else -1
+        }
+    }
+
+    @Test
+    fun `substitute output works`() {
+        assertArrayEquals(byteArrayOf(100, -101), kryoOutput {
+            write(100)
+            substitute(::NegOutputStream)
+            write(101)
+        })
+    }
+
+    @Test
+    fun `substitute input works`() {
+        kryoInput(byteArrayOf(100, 101).inputStream()) {
+            assertEquals(100, read())
+            substitute(::NegInputStream)
+            assertEquals(-101, read().toByte())
+            assertEquals(-1, read())
+        }
+    }
+
+    @Test
+    fun `zip round-trip`() {
+        val data = ByteArray(12345).also { Random(0).nextBytes(it) }
+        val encoded = kryoOutput {
+            write(data)
+            substitute(::DeflaterOutputStream)
+            write(data)
+            substitute(::DeflaterOutputStream) // Potentially useful if a different codec.
+            write(data)
+        }
+        kryoInput(encoded.inputStream()) {
+            assertArrayEquals(data, readBytes(data.size))
+            substitute(::InflaterInputStream)
+            assertArrayEquals(data, readBytes(data.size))
+            substitute(::InflaterInputStream)
+            assertArrayEquals(data, readBytes(data.size))
+            assertEquals(-1, read())
+        }
+    }
+}

node/src/main/kotlin/net/corda/node/serialization/KryoServerSerializationScheme.kt

@@ -2,16 +2,16 @@ package net.corda.node.serialization
 
 import com.esotericsoftware.kryo.pool.KryoPool
 import net.corda.core.serialization.SerializationContext
-import net.corda.core.utilities.ByteSequence
+import net.corda.nodeapi.internal.serialization.CordaSerializationMagic
 import net.corda.node.services.messaging.RpcServerObservableSerializer
 import net.corda.nodeapi.internal.serialization.kryo.AbstractKryoSerializationScheme
 import net.corda.nodeapi.internal.serialization.kryo.DefaultKryoCustomizer
-import net.corda.nodeapi.internal.serialization.kryo.KryoHeaderV0_1
+import net.corda.nodeapi.internal.serialization.kryo.kryoMagic
 import net.corda.nodeapi.internal.serialization.kryo.RPCKryo
 
 class KryoServerSerializationScheme : AbstractKryoSerializationScheme() {
-    override fun canDeserializeVersion(byteSequence: ByteSequence, target: SerializationContext.UseCase): Boolean {
+    override fun canDeserializeVersion(magic: CordaSerializationMagic, target: SerializationContext.UseCase): Boolean {
-        return byteSequence == KryoHeaderV0_1 && target != SerializationContext.UseCase.RPCClient
+        return magic == kryoMagic && target != SerializationContext.UseCase.RPCClient
     }
 
     override fun rpcClientKryoPool(context: SerializationContext): KryoPool = throw UnsupportedOperationException()

node/src/main/kotlin/net/corda/node/services/identity/PersistentIdentityService.kt

@@ -16,6 +16,7 @@ import net.corda.nodeapi.internal.crypto.X509CertificateFactory
 import net.corda.nodeapi.internal.crypto.X509Utilities
 import net.corda.nodeapi.internal.crypto.x509Certificates
 import net.corda.nodeapi.internal.persistence.NODE_DATABASE_PREFIX
+import org.apache.commons.lang.ArrayUtils.EMPTY_BYTE_ARRAY
 import java.security.InvalidAlgorithmParameterException
 import java.security.PublicKey
 import java.security.cert.*
@@ -73,7 +74,7 @@ class PersistentIdentityService(override val trustRoot: X509Certificate,
 
             @Lob
             @Column(name = "identity_value")
-            var identity: ByteArray = ByteArray(0)
+            var identity: ByteArray = EMPTY_BYTE_ARRAY
     )
 
     @Entity

node/src/main/kotlin/net/corda/node/services/keys/PersistentKeyManagementService.kt

@@ -8,6 +8,7 @@ import net.corda.core.utilities.MAX_HASH_HEX_SIZE
 import net.corda.node.services.api.IdentityServiceInternal
 import net.corda.node.utilities.AppendOnlyPersistentMap
 import net.corda.nodeapi.internal.persistence.NODE_DATABASE_PREFIX
+import org.apache.commons.lang.ArrayUtils.EMPTY_BYTE_ARRAY
 import org.bouncycastle.operator.ContentSigner
 import java.security.KeyPair
 import java.security.PrivateKey
@@ -37,11 +38,10 @@ class PersistentKeyManagementService(val identityService: IdentityServiceInterna
 
             @Lob
             @Column(name = "public_key")
-            var publicKey: ByteArray = ByteArray(0),
+            var publicKey: ByteArray = EMPTY_BYTE_ARRAY,
-
             @Lob
             @Column(name = "private_key")
-            var privateKey: ByteArray = ByteArray(0)
+            var privateKey: ByteArray = EMPTY_BYTE_ARRAY
     ) {
         constructor(publicKey: PublicKey, privateKey: PrivateKey)
             : this(publicKey.toStringShort(), publicKey.encoded, privateKey.encoded)

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

@@ -40,6 +40,7 @@ import org.apache.activemq.artemis.api.core.SimpleString
 import org.apache.activemq.artemis.api.core.client.ClientConsumer
 import org.apache.activemq.artemis.api.core.client.ClientMessage
 import org.apache.activemq.artemis.api.core.client.ClientSession
+import org.apache.commons.lang.ArrayUtils.EMPTY_BYTE_ARRAY
 import rx.Subscription
 import java.security.PublicKey
 import java.time.Instant
@@ -196,11 +197,10 @@ class P2PMessagingClient(config: NodeConfiguration,
 
             @Lob
             @Column
-            var message: ByteArray = ByteArray(0),
+            var message: ByteArray = EMPTY_BYTE_ARRAY,
-
             @Lob
             @Column
-            var recipients: ByteArray = ByteArray(0)
+            var recipients: ByteArray = EMPTY_BYTE_ARRAY
     )
 
     fun start() {

node/src/main/kotlin/net/corda/node/services/persistence/DBCheckpointStorage.kt

@@ -5,6 +5,7 @@ import net.corda.node.services.api.Checkpoint
 import net.corda.node.services.api.CheckpointStorage
 import net.corda.nodeapi.internal.persistence.NODE_DATABASE_PREFIX
 import net.corda.nodeapi.internal.persistence.currentDBSession
+import org.apache.commons.lang.ArrayUtils.EMPTY_BYTE_ARRAY
 import javax.persistence.Column
 import javax.persistence.Entity
 import javax.persistence.Id
@@ -24,13 +25,13 @@ class DBCheckpointStorage : CheckpointStorage {
 
             @Lob
             @Column(name = "checkpoint_value")
-            var checkpoint: ByteArray = ByteArray(0)
+            var checkpoint: ByteArray = EMPTY_BYTE_ARRAY
     )
 
     override fun addCheckpoint(checkpoint: Checkpoint) {
         currentDBSession().save(DBCheckpoint().apply {
             checkpointId = checkpoint.id.toString()
-            this.checkpoint = checkpoint.serializedFiber.bytes
+            this.checkpoint = checkpoint.serializedFiber.bytes // XXX: Is copying the byte array necessary?
         })
     }
 

node/src/main/kotlin/net/corda/node/services/persistence/DBTransactionStorage.kt

@@ -13,6 +13,7 @@ import net.corda.node.utilities.*
 import net.corda.nodeapi.internal.persistence.NODE_DATABASE_PREFIX
 import net.corda.nodeapi.internal.persistence.bufferUntilDatabaseCommit
 import net.corda.nodeapi.internal.persistence.wrapWithDatabaseTransaction
+import org.apache.commons.lang.ArrayUtils.EMPTY_BYTE_ARRAY
 import rx.Observable
 import rx.subjects.PublishSubject
 import java.util.*
@@ -34,7 +35,7 @@ class DBTransactionStorage(cacheSizeBytes: Long) : WritableTransactionStorage, S
 
             @Lob
             @Column(name = "transaction_value")
-            var transaction: ByteArray = ByteArray(0)
+            var transaction: ByteArray = EMPTY_BYTE_ARRAY
     )
 
     private companion object {

node/src/main/kotlin/net/corda/node/services/transactions/PersistentUniquenessProvider.kt

@@ -13,6 +13,7 @@ import net.corda.core.serialization.SingletonSerializeAsToken
 import net.corda.core.utilities.contextLogger
 import net.corda.node.utilities.AppendOnlyPersistentMap
 import net.corda.nodeapi.internal.persistence.NODE_DATABASE_PREFIX
+import org.apache.commons.lang.ArrayUtils.EMPTY_BYTE_ARRAY
 import org.hibernate.annotations.Type
 import java.io.Serializable
 import java.util.*
@@ -45,7 +46,7 @@ class PersistentUniquenessProvider : UniquenessProvider, SingletonSerializeAsTok
 
             @Column(name = "requesting_party_key", length = 255)
             @Type(type = "corda-wrapper-binary")
-            var owningKey: ByteArray = ByteArray(0)
+            var owningKey: ByteArray = EMPTY_BYTE_ARRAY
     ) : Serializable
 
     @Entity

node/src/main/kotlin/net/corda/node/services/transactions/RaftUniquenessProvider.kt

@@ -33,6 +33,7 @@ import net.corda.nodeapi.internal.config.NodeSSLConfiguration
 import net.corda.nodeapi.internal.config.SSLConfiguration
 import net.corda.nodeapi.internal.persistence.CordaPersistence
 import net.corda.nodeapi.internal.persistence.NODE_DATABASE_PREFIX
+import org.apache.commons.lang.ArrayUtils.EMPTY_BYTE_ARRAY
 import java.nio.file.Path
 import java.util.concurrent.CompletableFuture
 import javax.annotation.concurrent.ThreadSafe
@@ -76,8 +77,7 @@ class RaftUniquenessProvider(private val transportConfiguration: NodeSSLConfigur
 
             @Lob
             @Column(name = "state_value")
-            var value: ByteArray = ByteArray(0),
+            var value: ByteArray = EMPTY_BYTE_ARRAY,
-
             @Column(name = "state_index")
             var index: Long = 0
     )

node/src/test/kotlin/net/corda/node/messaging/InMemoryMessagingTests.kt

@@ -6,6 +6,7 @@ import net.corda.node.services.messaging.TopicStringValidator
 import net.corda.node.services.messaging.createMessage
 import net.corda.testing.internal.rigorousMock
 import net.corda.testing.node.MockNetwork
+import org.apache.commons.lang.ArrayUtils.EMPTY_BYTE_ARRAY
 import org.junit.After
 import org.junit.Before
 import org.junit.Test
@@ -93,9 +94,8 @@ class InMemoryMessagingTests {
         node1.network.addMessageHandler("valid_message") { _, _ ->
             received++
         }
-
+        val invalidMessage = node2.network.createMessage("invalid_message", data = EMPTY_BYTE_ARRAY)
-        val invalidMessage = node2.network.createMessage("invalid_message", data = ByteArray(0))
+        val validMessage = node2.network.createMessage("valid_message", data = EMPTY_BYTE_ARRAY)
-        val validMessage = node2.network.createMessage("valid_message", data = ByteArray(0))
         node2.network.send(invalidMessage, node1.network.myAddress)
         mockNet.runNetwork()
         assertEquals(0, received)
@@ -106,8 +106,8 @@ class InMemoryMessagingTests {
 
         // Here's the core of the test; previously the unhandled message would cause runNetwork() to abort early, so
         // this would fail. Make fresh messages to stop duplicate uniqueMessageId causing drops
-        val invalidMessage2 = node2.network.createMessage("invalid_message", data = ByteArray(0))
+        val invalidMessage2 = node2.network.createMessage("invalid_message", data = EMPTY_BYTE_ARRAY)
-        val validMessage2 = node2.network.createMessage("valid_message", data = ByteArray(0))
+        val validMessage2 = node2.network.createMessage("valid_message", data = EMPTY_BYTE_ARRAY)
         node2.network.send(invalidMessage2, node1.network.myAddress)
         node2.network.send(validMessage2, node1.network.myAddress)
         mockNet.runNetwork()

tools/explorer/build.gradle

@@ -50,7 +50,6 @@ dependencies {
 
     // Controls FX: more java FX components http://fxexperience.com/controlsfx/
     compile 'org.controlsfx:controlsfx:8.40.12'
-    compile 'commons-lang:commons-lang:2.6'
     // This provide com.apple.eawt stub for non-mac system.
     compile 'com.yuvimasory:orange-extensions:1.3.0'
 }

verifier/src/integration-test/kotlin/net/corda/verifier/GeneratedLedger.kt

@@ -12,6 +12,7 @@ import net.corda.core.transactions.LedgerTransaction
 import net.corda.core.transactions.WireTransaction
 import net.corda.nodeapi.internal.serialization.GeneratedAttachment
 import net.corda.testing.contracts.DummyContract
+import org.apache.commons.lang.ArrayUtils.EMPTY_BYTE_ARRAY
 import java.math.BigInteger
 import java.security.PublicKey
 import java.util.*
@@ -37,7 +38,7 @@ data class GeneratedLedger(
 
     companion object {
         val empty = GeneratedLedger(emptyList(), emptyMap(), emptySet(), emptySet())
-        val contractAttachment = ContractAttachment(GeneratedAttachment(ByteArray(0) { 0 }), DummyContract.PROGRAM_ID)
+        val contractAttachment = ContractAttachment(GeneratedAttachment(EMPTY_BYTE_ARRAY), DummyContract.PROGRAM_ID)
     }
 
     fun resolveWireTransaction(transaction: WireTransaction): LedgerTransaction {

verifier/src/main/kotlin/net/corda/verifier/Verifier.kt

@@ -5,6 +5,7 @@ import com.typesafe.config.ConfigFactory
 import com.typesafe.config.ConfigParseOptions
 import net.corda.core.internal.div
 import net.corda.core.serialization.SerializationContext
+import net.corda.nodeapi.internal.serialization.CordaSerializationMagic
 import net.corda.core.serialization.internal.SerializationEnvironmentImpl
 import net.corda.core.serialization.internal.nodeSerializationEnv
 import net.corda.core.utilities.*
@@ -17,10 +18,10 @@ import net.corda.nodeapi.internal.config.getValue
 import net.corda.nodeapi.internal.addShutdownHook
 import net.corda.nodeapi.internal.serialization.*
 import net.corda.nodeapi.internal.serialization.amqp.AbstractAMQPSerializationScheme
-import net.corda.nodeapi.internal.serialization.amqp.AmqpHeaderV1_0
+import net.corda.nodeapi.internal.serialization.amqp.amqpMagic
 import net.corda.nodeapi.internal.serialization.amqp.SerializerFactory
 import net.corda.nodeapi.internal.serialization.kryo.AbstractKryoSerializationScheme
-import net.corda.nodeapi.internal.serialization.kryo.KryoHeaderV0_1
+import net.corda.nodeapi.internal.serialization.kryo.kryoMagic
 import org.apache.activemq.artemis.api.core.client.ActiveMQClient
 import java.nio.file.Path
 import java.nio.file.Paths
@@ -96,8 +97,8 @@ class Verifier {
     }
 
     private object KryoVerifierSerializationScheme : AbstractKryoSerializationScheme() {
-        override fun canDeserializeVersion(byteSequence: ByteSequence, target: SerializationContext.UseCase): Boolean {
+        override fun canDeserializeVersion(magic: CordaSerializationMagic, target: SerializationContext.UseCase): Boolean {
-            return byteSequence == KryoHeaderV0_1 && target == SerializationContext.UseCase.P2P
+            return magic == kryoMagic && target == SerializationContext.UseCase.P2P
         }
 
         override fun rpcClientKryoPool(context: SerializationContext) = throw UnsupportedOperationException()
@@ -105,8 +106,8 @@ class Verifier {
     }
 
     private object AMQPVerifierSerializationScheme : AbstractAMQPSerializationScheme(emptyList()) {
-        override fun canDeserializeVersion(byteSequence: ByteSequence, target: SerializationContext.UseCase): Boolean {
+        override fun canDeserializeVersion(magic: CordaSerializationMagic, target: SerializationContext.UseCase): Boolean {
-            return (byteSequence == AmqpHeaderV1_0 && (target == SerializationContext.UseCase.P2P))
+            return magic == amqpMagic && target == SerializationContext.UseCase.P2P
         }
 
         override fun rpcClientSerializerFactory(context: SerializationContext): SerializerFactory = throw UnsupportedOperationException()