Introduced a few helper methods around InputStream.readBytes (#2928)

core/src/main/kotlin/net/corda/core/flows/SendTransactionFlow.kt

@@ -3,6 +3,7 @@ package net.corda.core.flows
 import co.paralleluniverse.fibers.Suspendable
 import net.corda.core.contracts.StateAndRef
 import net.corda.core.internal.FetchDataFlow
+import net.corda.core.internal.readFully
 import net.corda.core.transactions.SignedTransaction
 import net.corda.core.utilities.unwrap
 
@@ -60,7 +61,7 @@ open class DataVendingFlow(val otherSideSession: FlowSession, val payload: Any)
                     serviceHub.validatedTransactions.getTransaction(it) ?: throw FetchDataFlow.HashNotFound(it)
                 }
                 FetchDataFlow.DataType.ATTACHMENT -> dataRequest.hashes.map {
-                    serviceHub.attachments.openAttachment(it)?.open()?.readBytes() ?: throw FetchDataFlow.HashNotFound(it)
+                    serviceHub.attachments.openAttachment(it)?.open()?.readFully() ?: throw FetchDataFlow.HashNotFound(it)
                 }
             }
         }

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

@@ -28,7 +28,7 @@ abstract class AbstractAttachment(dataLoader: () -> ByteArray) : Attachment {
         fun SerializeAsTokenContext.attachmentDataLoader(id: SecureHash): () -> ByteArray {
             return {
                 val a = serviceHub.attachments.openAttachment(id) ?: throw MissingAttachmentsException(listOf(id))
-                (a as? AbstractAttachment)?.attachmentData ?: a.open().use { it.readBytes() }
+                (a as? AbstractAttachment)?.attachmentData ?: a.open().readFully()
             }
         }
 

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

@@ -2,6 +2,8 @@
 
 package net.corda.core.internal
 
+import com.google.common.hash.Hashing
+import com.google.common.hash.HashingInputStream
 import net.corda.core.cordapp.Cordapp
 import net.corda.core.cordapp.CordappConfig
 import net.corda.core.cordapp.CordappContext
@@ -150,8 +152,31 @@ fun Path.writeLines(lines: Iterable<CharSequence>, charset: Charset = UTF_8, var
 
 inline fun <reified T : Any> Path.readObject(): T = readAll().deserialize()
 
+/** Calculate the hash of the contents of this file. */
+val Path.hash: SecureHash get() = read { it.hash() }
+
 fun InputStream.copyTo(target: Path, vararg options: CopyOption): Long = Files.copy(this, target, *options)
 
+/** Same as [InputStream.readBytes] but also closes the stream. */
+fun InputStream.readFully(): ByteArray = use { it.readBytes() }
+
+/** Calculate the hash of the remaining bytes in this input stream. The stream is closed at the end. */
+fun InputStream.hash(): SecureHash {
+    return use {
+        val his = HashingInputStream(Hashing.sha256(), it)
+        his.copyTo(NullOutputStream)  // To avoid reading in the entire stream into memory just write out the bytes to /dev/null
+        SecureHash.SHA256(his.hash().asBytes())
+    }
+}
+
+inline fun <reified T : Any> InputStream.readObject(): T = readFully().deserialize()
+
+object NullOutputStream : OutputStream() {
+    override fun write(b: Int) = Unit
+    override fun write(b: ByteArray) = Unit
+    override fun write(b: ByteArray, off: Int, len: Int) = Unit
+}
+
 fun String.abbreviate(maxWidth: Int): String = if (length <= maxWidth) this else take(maxWidth - 1) + "…"
 
 /** Return the sum of an Iterable of [BigDecimal]s. */
@@ -333,7 +358,7 @@ fun URL.post(serializedData: OpaqueBytes): ByteArray {
         setRequestProperty("Content-Type", "application/octet-stream")
         outputStream.use { serializedData.open().copyTo(it) }
         checkOkResponse()
-        inputStream.use { it.readBytes() }
+        inputStream.readFully()
     }
 }
 
@@ -346,7 +371,7 @@ fun HttpURLConnection.checkOkResponse() {
 
 inline fun <reified T : Any> HttpURLConnection.responseAs(): T {
     checkOkResponse()
-    return inputStream.use { it.readBytes() }.deserialize()
+    return inputStream.readObject()
 }
 
 /** Analogous to [Thread.join]. */

core/src/test/kotlin/net/corda/core/flows/AttachmentTests.kt

@@ -7,6 +7,7 @@ import net.corda.core.crypto.sha256
 import net.corda.core.identity.Party
 import net.corda.core.internal.FetchAttachmentsFlow
 import net.corda.core.internal.FetchDataFlow
+import net.corda.core.internal.hash
 import net.corda.core.utilities.getOrThrow
 import net.corda.node.internal.StartedNode
 import net.corda.node.services.persistence.NodeAttachmentService
@@ -72,7 +73,7 @@ class AttachmentTests {
             bobNode.attachments.openAttachment(id)!!
         }
 
-        assertEquals(id, attachment.open().readBytes().sha256())
+        assertEquals(id, attachment.open().hash())
 
         // Shut down node zero and ensure node one can still resolve the attachment.
         aliceNode.dispose()

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

@@ -1,11 +1,7 @@
 package net.corda.nodeapi.internal.network
 
-import com.google.common.hash.Hashing
-import com.google.common.hash.HashingInputStream
 import com.typesafe.config.ConfigFactory
 import net.corda.cordform.CordformNode
-import net.corda.core.contracts.ContractClassName
-import net.corda.core.crypto.SecureHash
 import net.corda.core.crypto.SecureHash.Companion.parse
 import net.corda.core.identity.Party
 import net.corda.core.internal.*
@@ -27,7 +23,6 @@ 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.kryoMagic
-import java.io.File
 import java.io.PrintStream
 import java.nio.file.Files
 import java.nio.file.Path
@@ -191,7 +186,7 @@ class NetworkBootstrapper {
     private fun generateWhitelist(whitelistFile: Path, excludeWhitelistFile: Path, cordapps: List<String>?): Map<String, List<AttachmentId>> {
         val existingWhitelist = if (whitelistFile.exists()) readContractWhitelist(whitelistFile) else emptyMap()
 
-        println(if (existingWhitelist.isEmpty()) "No existing whitelist file found." else "Found existing whitelist: ${whitelistFile}")
+        println(if (existingWhitelist.isEmpty()) "No existing whitelist file found." else "Found existing whitelist: $whitelistFile")
 
         val excludeContracts = if (excludeWhitelistFile.exists()) readExcludeWhitelist(excludeWhitelistFile) else emptyList()
         if (excludeContracts.isNotEmpty()) {
@@ -199,7 +194,7 @@ class NetworkBootstrapper {
         }
 
         val newWhiteList = cordapps?.flatMap { cordappJarPath ->
-            val jarHash = getJarHash(cordappJarPath)
+            val jarHash = Paths.get(cordappJarPath).hash
             scanJarForContracts(cordappJarPath).map { contract ->
                 contract to jarHash
             }
@@ -213,30 +208,26 @@ class NetworkBootstrapper {
             contractClassName to (if (newHash == null || newHash in existing) existing else existing + newHash)
         }.toMap()
 
-        println("CorDapp whitelist " + (if (existingWhitelist.isEmpty()) "generated" else "updated") + " in ${whitelistFile}")
+        println("CorDapp whitelist " + (if (existingWhitelist.isEmpty()) "generated" else "updated") + " in $whitelistFile")
         return merged
     }
 
     private fun overwriteWhitelist(whitelistFile: Path, mergedWhiteList: Map<String, List<AttachmentId>>) {
         PrintStream(whitelistFile.toFile().outputStream()).use { out ->
             mergedWhiteList.forEach { (contract, attachments) ->
-                out.println("${contract}:${attachments.joinToString(",")}")
+                out.println("$contract:${attachments.joinToString(",")}")
             }
         }
     }
 
-    private fun getJarHash(cordappPath: String): AttachmentId = File(cordappPath).inputStream().use { jar ->
-        val hs = HashingInputStream(Hashing.sha256(), jar)
-        hs.readBytes()
-        SecureHash.SHA256(hs.hash().asBytes())
+    private fun readContractWhitelist(file: Path): Map<String, List<AttachmentId>> {
+        return file.readAllLines()
+                .map { line -> line.split(":") }
+                .map { (contract, attachmentIds) ->
+                    contract to (attachmentIds.split(",").map(::parse))
+                }.toMap()
     }
 
-    private fun readContractWhitelist(file: Path): Map<String, List<AttachmentId>> = file.readAllLines()
-            .map { line -> line.split(":") }
-            .map { (contract, attachmentIds) ->
-                contract to (attachmentIds.split(",").map(::parse))
-            }.toMap()
-
     private fun readExcludeWhitelist(file: Path): List<String> = file.readAllLines().map(String::trim)
 
     private fun NotaryInfo.prettyPrint(): String = "${identity.name} (${if (validating) "" else "non-"}validating)"

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

@@ -3,6 +3,7 @@ package net.corda.nodeapi.internal.serialization.amqp.custom
 import net.corda.core.contracts.Attachment
 import net.corda.core.contracts.ContractAttachment
 import net.corda.core.contracts.ContractClassName
+import net.corda.core.internal.readFully
 import net.corda.core.serialization.MissingAttachmentsException
 import net.corda.nodeapi.internal.serialization.GeneratedAttachment
 import net.corda.nodeapi.internal.serialization.amqp.CustomSerializer
@@ -17,7 +18,7 @@ class ContractAttachmentSerializer(factory: SerializerFactory) : CustomSerialize
         ContractAttachmentProxy::class.java, factory) {
     override fun toProxy(obj: ContractAttachment): ContractAttachmentProxy {
         val bytes = try {
-            obj.attachment.open().readBytes()
+            obj.attachment.open().readFully()
         } catch (e: Exception) {
             throw MissingAttachmentsException(listOf(obj.id))
         }

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

@@ -18,6 +18,7 @@ import net.corda.core.crypto.CompositeKey
 import net.corda.core.crypto.SecureHash
 import net.corda.core.identity.PartyAndCertificate
 import net.corda.core.internal.AbstractAttachment
+import net.corda.core.internal.readFully
 import net.corda.core.serialization.MissingAttachmentsException
 import net.corda.core.serialization.SerializationWhitelist
 import net.corda.core.serialization.SerializeAsToken
@@ -223,7 +224,7 @@ object DefaultKryoCustomizer {
                 val lazyAttachment = object : AbstractAttachment({
                     val attachment = attachmentStorage.openAttachment(attachmentHash)
                             ?: throw MissingAttachmentsException(listOf(attachmentHash))
-                    attachment.open().readBytes()
+                    attachment.open().readFully()
                 }) {
                     override val id = attachmentHash
                 }

node/src/integration-test/kotlin/net/corda/node/utilities/registration/NodeRegistrationTest.kt

@@ -3,6 +3,7 @@ package net.corda.node.utilities.registration
 import net.corda.core.identity.CordaX500Name
 import net.corda.core.internal.concurrent.transpose
 import net.corda.core.internal.logElapsedTime
+import net.corda.core.internal.readFully
 import net.corda.core.messaging.startFlow
 import net.corda.core.utilities.*
 import net.corda.finance.DOLLARS
@@ -130,7 +131,7 @@ class RegistrationHandler(private val rootCertAndKeyPair: CertificateAndKeyPair)
     @Produces(MediaType.TEXT_PLAIN)
     fun registration(input: InputStream): Response {
         return log.logElapsedTime("Registration") {
-            val certificationRequest = input.use { JcaPKCS10CertificationRequest(it.readBytes()) }
+            val certificationRequest = JcaPKCS10CertificationRequest(input.readFully())
             val (certPath, name) = createSignedClientCertificate(
                     certificationRequest,
                     rootCertAndKeyPair.keyPair,

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

@@ -11,9 +11,11 @@ import net.corda.core.contracts.Attachment
 import net.corda.core.contracts.ContractAttachment
 import net.corda.core.contracts.ContractClassName
 import net.corda.core.crypto.SecureHash
+import net.corda.core.crypto.sha256
 import net.corda.core.internal.AbstractAttachment
 import net.corda.core.internal.UNKNOWN_UPLOADER
 import net.corda.core.internal.VisibleForTesting
+import net.corda.core.internal.readFully
 import net.corda.core.node.services.AttachmentId
 import net.corda.core.node.services.AttachmentStorage
 import net.corda.core.node.services.vault.AttachmentQueryCriteria
@@ -27,7 +29,10 @@ import net.corda.node.utilities.NonInvalidatingWeightBasedCache
 import net.corda.nodeapi.internal.persistence.NODE_DATABASE_PREFIX
 import net.corda.nodeapi.internal.persistence.currentDBSession
 import net.corda.nodeapi.internal.withContractsInJar
-import java.io.*
+import java.io.FilterInputStream
+import java.io.IOException
+import java.io.InputStream
+import java.io.Serializable
 import java.nio.file.Paths
 import java.time.Instant
 import java.util.*
@@ -262,14 +267,6 @@ class NodeAttachmentService(
         return import(jar, uploader, filename)
     }
 
-    fun getAttachmentIdAndBytes(jar: InputStream): Pair<AttachmentId, ByteArray> {
-        val hs = HashingInputStream(Hashing.sha256(), jar)
-        val bytes = hs.readBytes()
-        checkIsAValidJAR(ByteArrayInputStream(bytes))
-        val id = SecureHash.SHA256(hs.hash().asBytes())
-        return Pair(id, bytes)
-    }
-
     override fun hasAttachment(attachmentId: AttachmentId): Boolean =
             currentDBSession().find(NodeAttachmentService.DBAttachment::class.java, attachmentId.toString()) != null
 
@@ -278,15 +275,16 @@ class NodeAttachmentService(
         return withContractsInJar(jar) { contractClassNames, inputStream ->
             require(inputStream !is JarInputStream)
 
-            // Read the file into RAM, hashing it to find the ID as we go. The attachment must fit into memory.
+            // Read the file into RAM and then calculate its hash. The attachment must fit into memory.
             // TODO: Switch to a two-phase insert so we can handle attachments larger than RAM.
             // To do this we must pipe stream into the database without knowing its hash, which we will learn only once
             // the insert/upload is complete. We can then query to see if it's a duplicate and if so, erase, and if not
             // set the hash field of the new attachment record.
 
-            val (id, bytes) = getAttachmentIdAndBytes(inputStream)
+            val bytes = inputStream.readFully()
+            val id = bytes.sha256()
             if (!hasAttachment(id)) {
-                checkIsAValidJAR(ByteArrayInputStream(bytes))
+                checkIsAValidJAR(bytes.inputStream())
                 val session = currentDBSession()
                 val attachment = NodeAttachmentService.DBAttachment(attId = id.toString(), content = bytes, uploader = uploader, filename = filename, contractClassNames = contractClassNames)
                 session.save(attachment)

node/src/test/kotlin/net/corda/node/services/persistence/NodeAttachmentStorageTest.kt

@@ -5,10 +5,7 @@ import com.google.common.jimfs.Configuration
 import com.google.common.jimfs.Jimfs
 import net.corda.core.crypto.SecureHash
 import net.corda.core.crypto.sha256
-import net.corda.core.internal.read
-import net.corda.core.internal.readAll
-import net.corda.core.internal.write
-import net.corda.core.internal.writeLines
+import net.corda.core.internal.*
 import net.corda.core.node.services.vault.AttachmentQueryCriteria
 import net.corda.core.node.services.vault.AttachmentSort
 import net.corda.core.node.services.vault.Builder
@@ -230,7 +227,7 @@ class NodeAttachmentStorageTest {
         database.transaction {
             val storage = NodeAttachmentService(MetricRegistry())
             val e = assertFailsWith<NodeAttachmentService.HashMismatchException> {
-                storage.openAttachment(id)!!.open().use { it.readBytes() }
+                storage.openAttachment(id)!!.open().readFully()
             }
             assertEquals(e.expected, id)
 

testing/node-driver/src/main/kotlin/net/corda/testing/node/internal/network/NetworkMapServer.kt

@@ -2,14 +2,14 @@ package net.corda.testing.node.internal.network
 
 import net.corda.core.crypto.SecureHash
 import net.corda.core.crypto.SignedData
+import net.corda.core.internal.readObject
+import net.corda.core.node.NetworkParameters
 import net.corda.core.node.NodeInfo
-import net.corda.core.serialization.deserialize
 import net.corda.core.serialization.serialize
 import net.corda.core.utilities.NetworkHostAndPort
 import net.corda.nodeapi.internal.SignedNodeInfo
 import net.corda.nodeapi.internal.createDevNetworkMapCa
 import net.corda.nodeapi.internal.crypto.CertificateAndKeyPair
-import net.corda.core.node.NetworkParameters
 import net.corda.nodeapi.internal.network.NetworkMap
 import net.corda.nodeapi.internal.network.ParametersUpdate
 import org.eclipse.jetty.server.Server
@@ -110,7 +110,7 @@ class NetworkMapServer(private val pollInterval: Duration,
         @Consumes(MediaType.APPLICATION_OCTET_STREAM)
         fun publishNodeInfo(input: InputStream): Response {
             return try {
-                val signedNodeInfo = input.readBytes().deserialize<SignedNodeInfo>()
+                val signedNodeInfo = input.readObject<SignedNodeInfo>()
                 signedNodeInfo.verified()
                 nodeInfoMap[signedNodeInfo.raw.hash] = signedNodeInfo
                 ok()
@@ -126,7 +126,7 @@ class NetworkMapServer(private val pollInterval: Duration,
         @Path("ack-parameters")
         @Consumes(MediaType.APPLICATION_OCTET_STREAM)
         fun ackNetworkParameters(input: InputStream): Response {
-            val signedParametersHash = input.readBytes().deserialize<SignedData<SecureHash>>()
+            val signedParametersHash = input.readObject<SignedData<SecureHash>>()
             val hash = signedParametersHash.verified()
             latestAcceptedParametersMap[signedParametersHash.sig.by] = hash
             return ok().build()

testing/test-utils/src/main/kotlin/net/corda/testing/services/MockAttachmentStorage.kt

@@ -7,13 +7,13 @@ import net.corda.core.crypto.SecureHash
 import net.corda.core.crypto.sha256
 import net.corda.core.internal.AbstractAttachment
 import net.corda.core.internal.UNKNOWN_UPLOADER
+import net.corda.core.internal.readFully
 import net.corda.core.node.services.AttachmentId
 import net.corda.core.node.services.AttachmentStorage
 import net.corda.core.node.services.vault.AttachmentQueryCriteria
 import net.corda.core.node.services.vault.AttachmentSort
 import net.corda.core.serialization.SingletonSerializeAsToken
 import net.corda.nodeapi.internal.withContractsInJar
-import java.io.ByteArrayOutputStream
 import java.io.InputStream
 import java.util.*
 import java.util.jar.JarInputStream
@@ -53,7 +53,7 @@ class MockAttachmentStorage : AttachmentStorage, SingletonSerializeAsToken() {
 
     fun importContractAttachment(contractClassNames: List<ContractClassName>, uploader: String, jar: InputStream): AttachmentId = importAttachmentInternal(jar, uploader, null, contractClassNames)
 
-    fun getAttachmentIdAndBytes(jar: InputStream): Pair<AttachmentId, ByteArray> = jar.readBytes().let { bytes -> Pair(bytes.sha256(), bytes) }
+    fun getAttachmentIdAndBytes(jar: InputStream): Pair<AttachmentId, ByteArray> = jar.readFully().let { bytes -> Pair(bytes.sha256(), bytes) }
 
     private class MockAttachment(dataLoader: () -> ByteArray, override val id: SecureHash) : AbstractAttachment(dataLoader)
 
@@ -61,7 +61,7 @@ class MockAttachmentStorage : AttachmentStorage, SingletonSerializeAsToken() {
         // JIS makes read()/readBytes() return bytes of the current file, but we want to hash the entire container here.
         require(jar !is JarInputStream)
 
-        val bytes = jar.readBytes()
+        val bytes = jar.readFully()
 
         val sha256 = bytes.sha256()
         if (sha256 !in files.keys) {