Refactored to use StreamableWriter

Bumped the msgpack library to 2.0.1 (the 2.0.0 build was fubar)

build.gradle

@@ -1,5 +1,5 @@
 buildscript {
-    ext.kotlin_version = '1.1.4-3'
+    ext.kotlin_version = '1.1.60'
     repositories {
         mavenCentral()
     }
@@ -13,12 +13,12 @@ plugins {
 }
 
 subprojects {
+    apply plugin: 'io.spring.dependency-management'
     apply plugin: 'kotlin'
     apply plugin: 'maven'
     apply plugin: 'signing'
     apply plugin: 'jacoco'
     apply plugin: 'gitflow-version'
-    apply plugin: 'io.spring.dependency-management'
     apply plugin: 'com.github.ben-manes.versions'
 
     sourceCompatibility = 1.7
@@ -138,7 +138,7 @@ subprojects {
                 entry 'slf4j-simple'
             }
 
-            dependency 'ch.dissem.msgpack:msgpack:2.0.0'
+            dependency 'ch.dissem.msgpack:msgpack:2.0.1'
             dependency 'org.bouncycastle:bcprov-jdk15on:1.57'
             dependency 'com.madgag.spongycastle:prov:1.56.0.0'
             dependency 'org.apache.commons:commons-lang3:3.6'

core/build.gradle

@@ -25,10 +25,10 @@ artifacts {
 
 dependencies {
     compile 'org.slf4j:slf4j-api'
-    compile 'ch.dissem.msgpack:msgpack:1.0.0'
+    compile 'ch.dissem.msgpack:msgpack'
-    testCompile 'junit:junit:4.12'
+    testCompile 'junit:junit'
-    testCompile 'org.hamcrest:hamcrest-library:1.3'
+    testCompile 'org.hamcrest:hamcrest-library'
-    testCompile 'com.nhaarman:mockito-kotlin:1.5.0'
+    testCompile 'com.nhaarman:mockito-kotlin'
     testCompile project(':cryptography-bc')
 }
 

core/src/main/kotlin/ch/dissem/bitmessage/BitmessageContext.kt

@@ -272,6 +272,7 @@ class BitmessageContext(
      */
     fun cleanup() {
         internals.inventory.cleanup()
+        internals.nodeRegistry.cleanup()
     }
 
     /**

core/src/main/kotlin/ch/dissem/bitmessage/entity/Addr.kt

@@ -25,19 +25,28 @@ import java.nio.ByteBuffer
  * The 'addr' command holds a list of known active Bitmessage nodes.
  */
 data class Addr constructor(val addresses: List<NetworkAddress>) : MessagePayload {
+
     override val command: MessagePayload.Command = MessagePayload.Command.ADDR
 
-    override fun write(out: OutputStream) {
+    override fun writer(): StreamableWriter = Writer(this)
-        Encode.varInt(addresses.size, out)
-        for (address in addresses) {
-            address.write(out)
-        }
-    }
 
-    override fun write(buffer: ByteBuffer) {
+    private class Writer(
-        Encode.varInt(addresses.size, buffer)
+        private val item: Addr
-        for (address in addresses) {
+    ) : StreamableWriter {
-            address.write(buffer)
+
+        override fun write(out: OutputStream) {
+            Encode.varInt(item.addresses.size, out)
+            for (address in item.addresses) {
+                address.writer().write(out)
+            }
         }
+
+        override fun write(buffer: ByteBuffer) {
+            Encode.varInt(item.addresses.size, buffer)
+            for (address in item.addresses) {
+                address.writer().write(buffer)
+            }
+        }
+
     }
 }

core/src/main/kotlin/ch/dissem/bitmessage/entity/CustomMessage.kt

@@ -33,36 +33,42 @@ open class CustomMessage(val customCommand: String, private val data: ByteArray?
 
     override val command: MessagePayload.Command = MessagePayload.Command.CUSTOM
 
-    val isError: Boolean
+    val isError = COMMAND_ERROR == customCommand
 
     fun getData(): ByteArray {
-        if (data != null) {
+        return data ?: {
-            return data
-        } else {
             val out = ByteArrayOutputStream()
-            write(out)
+            writer().write(out)
-            return out.toByteArray()
+            out.toByteArray()
-        }
+        }.invoke()
     }
 
-    override fun write(out: OutputStream) {
+    override fun writer(): StreamableWriter = Writer(this)
-        if (data != null) {
-            Encode.varString(customCommand, out)
-            out.write(data)
-        } else {
-            throw ApplicationException("Tried to write custom message without data. "
-                + "Programmer: did you forget to override #write()?")
-        }
-    }
 
-    override fun write(buffer: ByteBuffer) {
+    protected open class Writer(
-        if (data != null) {
+        private val item: CustomMessage
-            Encode.varString(customCommand, buffer)
+    ) : StreamableWriter {
-            buffer.put(data)
+
-        } else {
+        override fun write(out: OutputStream) {
-            throw ApplicationException("Tried to write custom message without data. "
+            if (item.data != null) {
-                + "Programmer: did you forget to override #write()?")
+                Encode.varString(item.customCommand, out)
+                out.write(item.data)
+            } else {
+                throw ApplicationException("Tried to write custom message without data. "
+                    + "Programmer: did you forget to override #write()?")
+            }
         }
+
+        override fun write(buffer: ByteBuffer) {
+            if (item.data != null) {
+                Encode.varString(item.customCommand, buffer)
+                buffer.put(item.data)
+            } else {
+                throw ApplicationException("Tried to write custom message without data. "
+                    + "Programmer: did you forget to override #write()?")
+            }
+        }
+
     }
 
     companion object {
@@ -75,12 +81,6 @@ open class CustomMessage(val customCommand: String, private val data: ByteArray?
         }
 
         @JvmStatic
-        fun error(message: String): CustomMessage {
+        fun error(message: String) = CustomMessage(COMMAND_ERROR, message.toByteArray(charset("UTF-8")))
-            return CustomMessage(COMMAND_ERROR, message.toByteArray(charset("UTF-8")))
-        }
-    }
-
-    init {
-        this.isError = COMMAND_ERROR == customCommand
     }
 }

core/src/main/kotlin/ch/dissem/bitmessage/entity/GetData.kt

@@ -28,21 +28,30 @@ class GetData constructor(var inventory: List<InventoryVector>) : MessagePayload
 
     override val command: MessagePayload.Command = MessagePayload.Command.GETDATA
 
-    override fun write(out: OutputStream) {
+    override fun writer(): StreamableWriter = Writer(this)
-        Encode.varInt(inventory.size, out)
-        for (iv in inventory) {
-            iv.write(out)
-        }
-    }
 
-    override fun write(buffer: ByteBuffer) {
+    private class Writer(
-        Encode.varInt(inventory.size, buffer)
+        private val item: GetData
-        for (iv in inventory) {
+    ) : StreamableWriter {
-            iv.write(buffer)
+
+        override fun write(out: OutputStream) {
+            Encode.varInt(item.inventory.size, out)
+            for (iv in item.inventory) {
+                iv.writer().write(out)
+            }
         }
+
+        override fun write(buffer: ByteBuffer) {
+            Encode.varInt(item.inventory.size, buffer)
+            for (iv in item.inventory) {
+                iv.writer().write(buffer)
+            }
+        }
+
     }
 
     companion object {
-        @JvmField val MAX_INVENTORY_SIZE = 50000
+        @JvmField
+        val MAX_INVENTORY_SIZE = 50000
     }
 }

core/src/main/kotlin/ch/dissem/bitmessage/entity/Inv.kt

@@ -28,17 +28,25 @@ class Inv constructor(val inventory: List<InventoryVector>) : MessagePayload {
 
     override val command: MessagePayload.Command = MessagePayload.Command.INV
 
-    override fun write(out: OutputStream) {
+    override fun writer(): StreamableWriter = Writer(this)
-        Encode.varInt(inventory.size, out)
-        for (iv in inventory) {
-            iv.write(out)
-        }
-    }
 
-    override fun write(buffer: ByteBuffer) {
+    private class Writer(
-        Encode.varInt(inventory.size, buffer)
+        private val item: Inv
-        for (iv in inventory) {
+    ) : StreamableWriter {
-            iv.write(buffer)
+
+        override fun write(out: OutputStream) {
+            Encode.varInt(item.inventory.size, out)
+            for (iv in item.inventory) {
+                iv.writer().write(out)
+            }
         }
+
+        override fun write(buffer: ByteBuffer) {
+            Encode.varInt(item.inventory.size, buffer)
+            for (iv in item.inventory) {
+                iv.writer().write(buffer)
+            }
+        }
+
     }
 }

core/src/main/kotlin/ch/dissem/bitmessage/entity/NetworkMessage.kt

@@ -18,7 +18,6 @@ package ch.dissem.bitmessage.entity
 
 import ch.dissem.bitmessage.utils.Encode
 import ch.dissem.bitmessage.utils.Singleton.cryptography
-import java.io.IOException
 import java.io.OutputStream
 import java.nio.ByteBuffer
 
@@ -32,87 +31,95 @@ data class NetworkMessage(
     val payload: MessagePayload
 ) : Streamable {
 
-    /**
+    override fun writer(): Writer = Writer(this)
-     * First 4 bytes of sha512(payload)
-     */
-    private fun getChecksum(bytes: ByteArray): ByteArray {
-        val d = cryptography().sha512(bytes)
-        return byteArrayOf(d[0], d[1], d[2], d[3])
-    }
 
-    override fun write(out: OutputStream) {
+    class Writer internal constructor(
-        // magic
+        private val item: NetworkMessage
-        Encode.int32(MAGIC, out)
+    ) : StreamableWriter {
 
-        // ASCII string identifying the packet content, NULL padded (non-NULL padding results in packet rejected)
+        override fun write(out: OutputStream) {
-        val command = payload.command.name.toLowerCase()
+            // magic
-        out.write(command.toByteArray(charset("ASCII")))
+            Encode.int32(MAGIC, out)
-        for (i in command.length..11) {
+
-            out.write(0x0)
+            // ASCII string identifying the packet content, NULL padded (non-NULL padding results in packet rejected)
+            val command = item.payload.command.name.toLowerCase()
+            out.write(command.toByteArray(charset("ASCII")))
+            for (i in command.length..11) {
+                out.write(0x0)
+            }
+
+            val payloadBytes = Encode.bytes(item.payload)
+
+            // Length of payload in number of bytes. Because of other restrictions, there is no reason why this length would
+            // ever be larger than 1600003 bytes. Some clients include a sanity-check to avoid processing messages which are
+            // larger than this.
+            Encode.int32(payloadBytes.size, out)
+
+            // checksum
+            out.write(getChecksum(payloadBytes))
+
+            // message payload
+            out.write(payloadBytes)
         }
 
-        val payloadBytes = Encode.bytes(payload)
+        /**
+         * A more efficient implementation of the write method, writing header data to the provided buffer and returning
+         * a new buffer containing the payload.
 
-        // Length of payload in number of bytes. Because of other restrictions, there is no reason why this length would
+         * @param headerBuffer where the header data is written to (24 bytes)
-        // ever be larger than 1600003 bytes. Some clients include a sanity-check to avoid processing messages which are
+         * *
-        // larger than this.
+         * @return a buffer containing the payload, ready to be read.
-        Encode.int32(payloadBytes.size, out)
+         */
-
+        fun writeHeaderAndGetPayloadBuffer(headerBuffer: ByteBuffer): ByteBuffer {
-        // checksum
+            return ByteBuffer.wrap(writeHeader(headerBuffer))
-        out.write(getChecksum(payloadBytes))
-
-        // message payload
-        out.write(payloadBytes)
-    }
-
-    /**
-     * A more efficient implementation of the write method, writing header data to the provided buffer and returning
-     * a new buffer containing the payload.
-
-     * @param headerBuffer where the header data is written to (24 bytes)
-     * *
-     * @return a buffer containing the payload, ready to be read.
-     */
-    fun writeHeaderAndGetPayloadBuffer(headerBuffer: ByteBuffer): ByteBuffer {
-        return ByteBuffer.wrap(writeHeader(headerBuffer))
-    }
-
-    /**
-     * For improved memory efficiency, you should use [.writeHeaderAndGetPayloadBuffer]
-     * and write the header buffer as well as the returned payload buffer into the channel.
-
-     * @param buffer where everything gets written to. Needs to be large enough for the whole message
-     * *               to be written.
-     */
-    override fun write(buffer: ByteBuffer) {
-        val payloadBytes = writeHeader(buffer)
-        buffer.put(payloadBytes)
-    }
-
-    private fun writeHeader(out: ByteBuffer): ByteArray {
-        // magic
-        Encode.int32(MAGIC, out)
-
-        // ASCII string identifying the packet content, NULL padded (non-NULL padding results in packet rejected)
-        val command = payload.command.name.toLowerCase()
-        out.put(command.toByteArray(charset("ASCII")))
-
-        for (i in command.length..11) {
-            out.put(0.toByte())
         }
 
-        val payloadBytes = Encode.bytes(payload)
+        /**
+         * For improved memory efficiency, you should use [.writeHeaderAndGetPayloadBuffer]
+         * and write the header buffer as well as the returned payload buffer into the channel.
 
-        // Length of payload in number of bytes. Because of other restrictions, there is no reason why this length would
+         * @param buffer where everything gets written to. Needs to be large enough for the whole message
-        // ever be larger than 1600003 bytes. Some clients include a sanity-check to avoid processing messages which are
+         * *               to be written.
-        // larger than this.
+         */
-        Encode.int32(payloadBytes.size, out)
+        override fun write(buffer: ByteBuffer) {
+            val payloadBytes = writeHeader(buffer)
+            buffer.put(payloadBytes)
+        }
 
-        // checksum
+        private fun writeHeader(out: ByteBuffer): ByteArray {
-        out.put(getChecksum(payloadBytes))
+            // magic
+            Encode.int32(MAGIC, out)
+
+            // ASCII string identifying the packet content, NULL padded (non-NULL padding results in packet rejected)
+            val command = item.payload.command.name.toLowerCase()
+            out.put(command.toByteArray(charset("ASCII")))
+
+            for (i in command.length..11) {
+                out.put(0.toByte())
+            }
+
+            val payloadBytes = Encode.bytes(item.payload)
+
+            // Length of payload in number of bytes. Because of other restrictions, there is no reason why this length would
+            // ever be larger than 1600003 bytes. Some clients include a sanity-check to avoid processing messages which are
+            // larger than this.
+            Encode.int32(payloadBytes.size, out)
+
+            // checksum
+            out.put(getChecksum(payloadBytes))
+
+            // message payload
+            return payloadBytes
+        }
+
+        /**
+         * First 4 bytes of sha512(payload)
+         */
+        private fun getChecksum(bytes: ByteArray): ByteArray {
+            val d = cryptography().sha512(bytes)
+            return byteArrayOf(d[0], d[1], d[2], d[3])
+        }
 
-        // message payload
-        return payloadBytes
     }
 
     companion object {

core/src/main/kotlin/ch/dissem/bitmessage/entity/ObjectMessage.kt

@@ -81,8 +81,8 @@ data class ObjectMessage(
         get() {
             try {
                 val out = ByteArrayOutputStream()
-                writeHeaderWithoutNonce(out)
+                writer.writeHeaderWithoutNonce(out)
-                payload.writeBytesToSign(out)
+                payload.writer().writeBytesToSign(out)
                 return out.toByteArray()
             } catch (e: IOException) {
                 throw ApplicationException(e)
@@ -131,30 +131,39 @@ data class ObjectMessage(
         return cryptography().isSignatureValid(bytesToSign, payload.signature ?: return false, pubkey)
     }
 
-    override fun write(out: OutputStream) {
-        out.write(nonce ?: ByteArray(8))
-        out.write(payloadBytesWithoutNonce)
-    }
-
-    override fun write(buffer: ByteBuffer) {
-        buffer.put(nonce ?: ByteArray(8))
-        buffer.put(payloadBytesWithoutNonce)
-    }
-
-    private fun writeHeaderWithoutNonce(out: OutputStream) {
-        Encode.int64(expiresTime, out)
-        Encode.int32(type, out)
-        Encode.varInt(version, out)
-        Encode.varInt(stream, out)
-    }
-
     val payloadBytesWithoutNonce: ByteArray by lazy {
         val out = ByteArrayOutputStream()
-        writeHeaderWithoutNonce(out)
+        writer.writeHeaderWithoutNonce(out)
-        payload.write(out)
+        payload.writer().write(out)
         out.toByteArray()
     }
 
+    private val writer = Writer(this)
+    override fun writer(): StreamableWriter = writer
+
+    private class Writer(
+        private val item: ObjectMessage
+    ) : StreamableWriter {
+
+        override fun write(out: OutputStream) {
+            out.write(item.nonce ?: ByteArray(8))
+            out.write(item.payloadBytesWithoutNonce)
+        }
+
+        override fun write(buffer: ByteBuffer) {
+            buffer.put(item.nonce ?: ByteArray(8))
+            buffer.put(item.payloadBytesWithoutNonce)
+        }
+
+        internal fun writeHeaderWithoutNonce(out: OutputStream) {
+            Encode.int64(item.expiresTime, out)
+            Encode.int32(item.type, out)
+            Encode.varInt(item.version, out)
+            Encode.varInt(item.stream, out)
+        }
+
+    }
+
     class Builder {
         private var nonce: ByteArray? = null
         private var expiresTime: Long = 0

core/src/main/kotlin/ch/dissem/bitmessage/entity/Plaintext.kt

@@ -260,102 +260,6 @@ class Plaintext private constructor(
         id = builder.id
     }
 
-    fun write(out: OutputStream, includeSignature: Boolean) {
-        Encode.varInt(from.version, out)
-        Encode.varInt(from.stream, out)
-        from.pubkey?.apply {
-            Encode.int32(behaviorBitfield, out)
-            out.write(signingKey, 1, 64)
-            out.write(encryptionKey, 1, 64)
-            if (from.version >= 3) {
-                Encode.varInt(nonceTrialsPerByte, out)
-                Encode.varInt(extraBytes, out)
-            }
-        } ?: {
-            Encode.int32(0, out)
-            val empty = ByteArray(64)
-            out.write(empty)
-            out.write(empty)
-            if (from.version >= 3) {
-                Encode.varInt(0, out)
-                Encode.varInt(0, out)
-            }
-        }.invoke()
-        if (type == MSG) {
-            out.write(to?.ripe ?: throw IllegalStateException("No recipient set for message"))
-        }
-        Encode.varInt(encodingCode, out)
-        Encode.varInt(message.size, out)
-        out.write(message)
-        if (type == MSG) {
-            if (to?.has(Feature.DOES_ACK) ?: false) {
-                val ack = ByteArrayOutputStream()
-                ackMessage?.write(ack)
-                Encode.varBytes(ack.toByteArray(), out)
-            } else {
-                Encode.varInt(0, out)
-            }
-        }
-        if (includeSignature) {
-            if (signature == null) {
-                Encode.varInt(0, out)
-            } else {
-                Encode.varBytes(signature!!, out)
-            }
-        }
-    }
-
-    fun write(buffer: ByteBuffer, includeSignature: Boolean) {
-        Encode.varInt(from.version, buffer)
-        Encode.varInt(from.stream, buffer)
-        if (from.pubkey == null) {
-            Encode.int32(0, buffer)
-            val empty = ByteArray(64)
-            buffer.put(empty)
-            buffer.put(empty)
-            if (from.version >= 3) {
-                Encode.varInt(0, buffer)
-                Encode.varInt(0, buffer)
-            }
-        } else {
-            Encode.int32(from.pubkey!!.behaviorBitfield, buffer)
-            buffer.put(from.pubkey!!.signingKey, 1, 64)
-            buffer.put(from.pubkey!!.encryptionKey, 1, 64)
-            if (from.version >= 3) {
-                Encode.varInt(from.pubkey!!.nonceTrialsPerByte, buffer)
-                Encode.varInt(from.pubkey!!.extraBytes, buffer)
-            }
-        }
-        if (type == MSG) {
-            buffer.put(to!!.ripe)
-        }
-        Encode.varInt(encodingCode, buffer)
-        Encode.varBytes(message, buffer)
-        if (type == MSG) {
-            if (to!!.has(Feature.DOES_ACK) && ackMessage != null) {
-                Encode.varBytes(Encode.bytes(ackMessage!!), buffer)
-            } else {
-                Encode.varInt(0, buffer)
-            }
-        }
-        if (includeSignature) {
-            val sig = signature
-            if (sig == null) {
-                Encode.varInt(0, buffer)
-            } else {
-                Encode.varBytes(sig, buffer)
-            }
-        }
-    }
-
-    override fun write(out: OutputStream) {
-        write(out, true)
-    }
-
-    override fun write(buffer: ByteBuffer) {
-        write(buffer, true)
-    }
-
     fun updateNextTry() {
         if (to != null) {
             if (nextTry == null) {
@@ -484,6 +388,7 @@ class Plaintext private constructor(
     }
 
     enum class Encoding constructor(code: Long) {
+
         IGNORE(0), TRIVIAL(1), SIMPLE(2), EXTENDED(3);
 
         var code: Long = 0
@@ -495,7 +400,8 @@ class Plaintext private constructor(
 
         companion object {
 
-            @JvmStatic fun fromCode(code: Long): Encoding? {
+            @JvmStatic
+            fun fromCode(code: Long): Encoding? {
                 for (e in values()) {
                     if (e.code == code) {
                         return e
@@ -503,12 +409,13 @@ class Plaintext private constructor(
                 }
                 return null
             }
+
         }
     }
 
     enum class Status {
+
         DRAFT,
-        // For sent messages
         PUBKEY_REQUESTED,
         DOING_PROOF_OF_WORK,
         SENT,
@@ -517,9 +424,110 @@ class Plaintext private constructor(
     }
 
     enum class Type {
+
         MSG, BROADCAST
     }
 
+    fun writer(includeSignature: Boolean): StreamableWriter = Writer(this, includeSignature)
+
+    override fun writer(): StreamableWriter = Writer(this)
+
+    private class Writer(
+        private val item: Plaintext,
+        private val includeSignature: Boolean = true
+    ) : StreamableWriter {
+
+        override fun write(out: OutputStream) {
+            Encode.varInt(item.from.version, out)
+            Encode.varInt(item.from.stream, out)
+            item.from.pubkey?.apply {
+                Encode.int32(behaviorBitfield, out)
+                out.write(signingKey, 1, 64)
+                out.write(encryptionKey, 1, 64)
+                if (item.from.version >= 3) {
+                    Encode.varInt(nonceTrialsPerByte, out)
+                    Encode.varInt(extraBytes, out)
+                }
+            } ?: {
+                Encode.int32(0, out)
+                val empty = ByteArray(64)
+                out.write(empty)
+                out.write(empty)
+                if (item.from.version >= 3) {
+                    Encode.varInt(0, out)
+                    Encode.varInt(0, out)
+                }
+            }.invoke()
+            if (item.type == MSG) {
+                out.write(item.to?.ripe ?: throw IllegalStateException("No recipient set for message"))
+            }
+            Encode.varInt(item.encodingCode, out)
+            Encode.varInt(item.message.size, out)
+            out.write(item.message)
+            if (item.type == MSG) {
+                if (item.to?.has(Feature.DOES_ACK) == true) {
+                    val ack = ByteArrayOutputStream()
+                    item.ackMessage?.writer()?.write(ack)
+                    Encode.varBytes(ack.toByteArray(), out)
+                } else {
+                    Encode.varInt(0, out)
+                }
+            }
+            if (includeSignature) {
+                val sig = item.signature
+                if (sig == null) {
+                    Encode.varInt(0, out)
+                } else {
+                    Encode.varBytes(sig, out)
+                }
+            }
+        }
+
+        override fun write(buffer: ByteBuffer) {
+            Encode.varInt(item.from.version, buffer)
+            Encode.varInt(item.from.stream, buffer)
+            if (item.from.pubkey == null) {
+                Encode.int32(0, buffer)
+                val empty = ByteArray(64)
+                buffer.put(empty)
+                buffer.put(empty)
+                if (item.from.version >= 3) {
+                    Encode.varInt(0, buffer)
+                    Encode.varInt(0, buffer)
+                }
+            } else {
+                Encode.int32(item.from.pubkey!!.behaviorBitfield, buffer)
+                buffer.put(item.from.pubkey!!.signingKey, 1, 64)
+                buffer.put(item.from.pubkey!!.encryptionKey, 1, 64)
+                if (item.from.version >= 3) {
+                    Encode.varInt(item.from.pubkey!!.nonceTrialsPerByte, buffer)
+                    Encode.varInt(item.from.pubkey!!.extraBytes, buffer)
+                }
+            }
+            if (item.type == MSG) {
+                buffer.put(item.to!!.ripe)
+            }
+            Encode.varInt(item.encodingCode, buffer)
+            Encode.varBytes(item.message, buffer)
+            if (item.type == MSG) {
+                if (item.to!!.has(Feature.DOES_ACK) && item.ackMessage != null) {
+                    Encode.varBytes(Encode.bytes(item.ackMessage!!), buffer)
+                } else {
+                    Encode.varInt(0, buffer)
+                }
+            }
+            if (includeSignature) {
+                val sig = item.signature
+                if (sig == null) {
+                    Encode.varInt(0, buffer)
+                } else {
+                    Encode.varBytes(sig, buffer)
+                }
+            }
+        }
+
+    }
+
     class Builder(internal val type: Type) {
         internal var id: Any? = null
         internal var inventoryVector: InventoryVector? = null
@@ -742,14 +750,16 @@ class Plaintext private constructor(
 
     companion object {
 
-        @JvmStatic fun read(type: Type, `in`: InputStream): Plaintext {
+        @JvmStatic
+        fun read(type: Type, `in`: InputStream): Plaintext {
             return readWithoutSignature(type, `in`)
                 .signature(Decode.varBytes(`in`))
                 .received(UnixTime.now)
                 .build()
         }
 
-        @JvmStatic fun readWithoutSignature(type: Type, `in`: InputStream): Plaintext.Builder {
+        @JvmStatic
+        fun readWithoutSignature(type: Type, `in`: InputStream): Plaintext.Builder {
             val version = Decode.varInt(`in`)
             return Builder(type)
                 .addressVersion(version)

core/src/main/kotlin/ch/dissem/bitmessage/entity/Streamable.kt

@@ -24,7 +24,27 @@ import java.nio.ByteBuffer
  * An object that can be written to an [OutputStream]
  */
 interface Streamable : Serializable {
-    fun write(out: OutputStream)
+    fun writer(): StreamableWriter
+}
 
+interface SignedStreamable : Streamable {
+    override fun writer(): SignedStreamableWriter
+}
+
+interface EncryptedStreamable : SignedStreamable {
+    override fun writer(): EncryptedStreamableWriter
+}
+
+interface StreamableWriter: Serializable {
+    fun write(out: OutputStream)
     fun write(buffer: ByteBuffer)
 }
+
+interface SignedStreamableWriter : StreamableWriter {
+    fun writeBytesToSign(out: OutputStream)
+}
+
+interface EncryptedStreamableWriter : SignedStreamableWriter {
+    fun writeUnencrypted(out: OutputStream)
+    fun writeUnencrypted(buffer: ByteBuffer)
+}

core/src/main/kotlin/ch/dissem/bitmessage/entity/VerAck.kt

@@ -26,12 +26,12 @@ class VerAck : MessagePayload {
 
     override val command: MessagePayload.Command = MessagePayload.Command.VERACK
 
-    override fun write(out: OutputStream) {
+    // 'verack' doesn't have any payload, so there is nothing to write
-        // 'verack' doesn't have any payload, so there is nothing to write
+    override fun writer(): StreamableWriter = EmptyWriter
-    }
 
-    override fun write(buffer: ByteBuffer) {
+    internal object EmptyWriter : StreamableWriter {
-        // 'verack' doesn't have any payload, so there is nothing to write
+        override fun write(out: OutputStream) = Unit
+        override fun write(buffer: ByteBuffer) = Unit
     }
 
     companion object {

core/src/main/kotlin/ch/dissem/bitmessage/entity/Version.kt

@@ -71,32 +71,36 @@ class Version constructor(
     val streams: LongArray = longArrayOf(1)
 ) : MessagePayload {
 
-    fun provides(service: Service?): Boolean {
+    fun provides(service: Service?) = service?.isEnabled(services) == true
-        return service != null && service.isEnabled(services)
-    }
 
     override val command: MessagePayload.Command = MessagePayload.Command.VERSION
 
-    override fun write(out: OutputStream) {
+    override fun writer(): StreamableWriter = Writer(this)
-        Encode.int32(version, out)
-        Encode.int64(services, out)
-        Encode.int64(timestamp, out)
-        addrRecv.write(out, true)
-        addrFrom.write(out, true)
-        Encode.int64(nonce, out)
-        Encode.varString(userAgent, out)
-        Encode.varIntList(streams, out)
-    }
 
-    override fun write(buffer: ByteBuffer) {
+    private class Writer(
-        Encode.int32(version, buffer)
+        private val item: Version
-        Encode.int64(services, buffer)
+    ) : StreamableWriter {
-        Encode.int64(timestamp, buffer)
+        override fun write(out: OutputStream) {
-        addrRecv.write(buffer, true)
+            Encode.int32(item.version, out)
-        addrFrom.write(buffer, true)
+            Encode.int64(item.services, out)
-        Encode.int64(nonce, buffer)
+            Encode.int64(item.timestamp, out)
-        Encode.varString(userAgent, buffer)
+            item.addrRecv.writer(true).write(out)
-        Encode.varIntList(streams, buffer)
+            item.addrFrom.writer(true).write(out)
+            Encode.int64(item.nonce, out)
+            Encode.varString(item.userAgent, out)
+            Encode.varIntList(item.streams, out)
+        }
+
+        override fun write(buffer: ByteBuffer) {
+            Encode.int32(item.version, buffer)
+            Encode.int64(item.services, buffer)
+            Encode.int64(item.timestamp, buffer)
+            item.addrRecv.writer(true).write(buffer)
+            item.addrFrom.writer(true).write(buffer)
+            Encode.int64(item.nonce, buffer)
+            Encode.varString(item.userAgent, buffer)
+            Encode.varIntList(item.streams, buffer)
+        }
     }
 
     class Builder {
@@ -187,9 +191,7 @@ class Version constructor(
         // TODO: NODE_SSL(2);
         NODE_NETWORK(1);
 
-        fun isEnabled(flag: Long): Boolean {
+        fun isEnabled(flag: Long) = (flag and this.flag) != 0L
-            return (flag and this.flag) != 0L
-        }
 
         companion object {
             fun getServiceFlag(vararg services: Service): Long {

core/src/main/kotlin/ch/dissem/bitmessage/entity/payload/Broadcast.kt

@@ -29,7 +29,12 @@ import java.util.*
  * Users who are subscribed to the sending address will see the message appear in their inbox.
  * Broadcasts are version 4 or 5.
  */
-abstract class Broadcast protected constructor(version: Long, override val stream: Long, protected var encrypted: CryptoBox?, override var plaintext: Plaintext?) : ObjectPayload(version), Encrypted, PlaintextHolder {
+abstract class Broadcast protected constructor(
+    version: Long,
+    override val stream: Long,
+    protected var encrypted: CryptoBox?,
+    override var plaintext: Plaintext?
+) : ObjectPayload(version), Encrypted, PlaintextHolder {
 
     override val isSigned: Boolean = true
 

core/src/main/kotlin/ch/dissem/bitmessage/entity/payload/CryptoBox.kt

@@ -17,6 +17,7 @@
 package ch.dissem.bitmessage.entity.payload
 
 import ch.dissem.bitmessage.entity.Streamable
+import ch.dissem.bitmessage.entity.StreamableWriter
 import ch.dissem.bitmessage.entity.valueobject.PrivateKey.Companion.PRIVATE_KEY_SIZE
 import ch.dissem.bitmessage.exception.DecryptionFailedException
 import ch.dissem.bitmessage.utils.*
@@ -108,44 +109,53 @@ class CryptoBox : Streamable {
 
     private fun calculateMac(key_m: ByteArray): ByteArray {
         val macData = ByteArrayOutputStream()
-        writeWithoutMAC(macData)
+        writer.writeWithoutMAC(macData)
         return cryptography().mac(key_m, macData.toByteArray())
     }
 
-    private fun writeWithoutMAC(out: OutputStream) {
+    private val writer = Writer(this)
-        out.write(initializationVector)
+    override fun writer(): StreamableWriter = writer
-        Encode.int16(curveType, out)
-        writeCoordinateComponent(out, Points.getX(R))
-        writeCoordinateComponent(out, Points.getY(R))
-        out.write(encrypted)
-    }
 
-    private fun writeCoordinateComponent(out: OutputStream, x: ByteArray) {
+    private class Writer(
-        val offset = Bytes.numberOfLeadingZeros(x)
+        private val item: CryptoBox
-        val length = x.size - offset
+    ) : StreamableWriter {
-        Encode.int16(length, out)
-        out.write(x, offset, length)
-    }
 
-    private fun writeCoordinateComponent(buffer: ByteBuffer, x: ByteArray) {
+        override fun write(out: OutputStream) {
-        val offset = Bytes.numberOfLeadingZeros(x)
+            writeWithoutMAC(out)
-        val length = x.size - offset
+            out.write(item.mac)
-        Encode.int16(length, buffer)
+        }
-        buffer.put(x, offset, length)
-    }
 
-    override fun write(out: OutputStream) {
+        internal fun writeWithoutMAC(out: OutputStream) {
-        writeWithoutMAC(out)
+            out.write(item.initializationVector)
-        out.write(mac)
+            Encode.int16(item.curveType, out)
-    }
+            writeCoordinateComponent(out, Points.getX(item.R))
+            writeCoordinateComponent(out, Points.getY(item.R))
+            out.write(item.encrypted)
+        }
+
+        override fun write(buffer: ByteBuffer) {
+            buffer.put(item.initializationVector)
+            Encode.int16(item.curveType, buffer)
+            writeCoordinateComponent(buffer, Points.getX(item.R))
+            writeCoordinateComponent(buffer, Points.getY(item.R))
+            buffer.put(item.encrypted)
+            buffer.put(item.mac)
+        }
+
+        private fun writeCoordinateComponent(out: OutputStream, x: ByteArray) {
+            val offset = Bytes.numberOfLeadingZeros(x)
+            val length = x.size - offset
+            Encode.int16(length, out)
+            out.write(x, offset, length)
+        }
+
+        private fun writeCoordinateComponent(buffer: ByteBuffer, x: ByteArray) {
+            val offset = Bytes.numberOfLeadingZeros(x)
+            val length = x.size - offset
+            Encode.int16(length, buffer)
+            buffer.put(x, offset, length)
+        }
 
-    override fun write(buffer: ByteBuffer) {
-        buffer.put(initializationVector)
-        Encode.int16(curveType, buffer)
-        writeCoordinateComponent(buffer, Points.getX(R))
-        writeCoordinateComponent(buffer, Points.getY(R))
-        buffer.put(encrypted)
-        buffer.put(mac)
     }
 
     class Builder {
@@ -187,15 +197,14 @@ class CryptoBox : Streamable {
             return this
         }
 
-        fun build(): CryptoBox {
+        fun build() = CryptoBox(this)
-            return CryptoBox(this)
-        }
     }
 
     companion object {
         private val LOG = LoggerFactory.getLogger(CryptoBox::class.java)
 
-        @JvmStatic fun read(stream: InputStream, length: Int): CryptoBox {
+        @JvmStatic
+        fun read(stream: InputStream, length: Int): CryptoBox {
             val counter = AccessCounter()
             return Builder()
                 .IV(Decode.bytes(stream, 16, counter))

core/src/main/kotlin/ch/dissem/bitmessage/entity/payload/GenericPayload.kt

@@ -16,6 +16,7 @@
 
 package ch.dissem.bitmessage.entity.payload
 
+import ch.dissem.bitmessage.entity.SignedStreamableWriter
 import ch.dissem.bitmessage.utils.Decode
 import java.io.InputStream
 import java.io.OutputStream
@@ -30,14 +31,6 @@ class GenericPayload(version: Long, override val stream: Long, val data: ByteArr
 
     override val type: ObjectType? = null
 
-    override fun write(out: OutputStream) {
-        out.write(data)
-    }
-
-    override fun write(buffer: ByteBuffer) {
-        buffer.put(data)
-    }
-
     override fun equals(other: Any?): Boolean {
         if (this === other) return true
         if (other !is GenericPayload) return false
@@ -52,9 +45,27 @@ class GenericPayload(version: Long, override val stream: Long, val data: ByteArr
         return result
     }
 
-    companion object {
+    override fun writer(): SignedStreamableWriter = Writer(this)
-        @JvmStatic fun read(version: Long, stream: Long, `is`: InputStream, length: Int): GenericPayload {
+
-            return GenericPayload(version, stream, Decode.bytes(`is`, length))
+    private class Writer(
+        private val item: GenericPayload
+    ) : SignedStreamableWriter {
+
+        override fun write(out: OutputStream) {
+            out.write(item.data)
         }
+
+        override fun write(buffer: ByteBuffer) {
+            buffer.put(item.data)
+        }
+
+        override fun writeBytesToSign(out: OutputStream) = Unit // nothing to do
+
+    }
+
+    companion object {
+        @JvmStatic
+        fun read(version: Long, stream: Long, `is`: InputStream, length: Int) =
+            GenericPayload(version, stream, Decode.bytes(`is`, length))
     }
 }

core/src/main/kotlin/ch/dissem/bitmessage/entity/payload/GetPubkey.kt

@@ -17,6 +17,7 @@
 package ch.dissem.bitmessage.entity.payload
 
 import ch.dissem.bitmessage.entity.BitmessageAddress
+import ch.dissem.bitmessage.entity.SignedStreamableWriter
 import ch.dissem.bitmessage.utils.Decode
 import java.io.InputStream
 import java.io.OutputStream
@@ -47,16 +48,27 @@ class GetPubkey : ObjectPayload {
         this.ripeTag = ripeOrTag
     }
 
-    override fun write(out: OutputStream) {
+    override fun writer(): SignedStreamableWriter = Writer(this)
-        out.write(ripeTag)
+
-    }
+    private class Writer(
+        private val item: GetPubkey
+    ) : SignedStreamableWriter {
+
+        override fun write(out: OutputStream) {
+            out.write(item.ripeTag)
+        }
+
+        override fun write(buffer: ByteBuffer) {
+            buffer.put(item.ripeTag)
+        }
+
+        override fun writeBytesToSign(out: OutputStream) = Unit // nothing to sign
 
-    override fun write(buffer: ByteBuffer) {
-        buffer.put(ripeTag)
     }
 
     companion object {
-        @JvmStatic fun read(`is`: InputStream, stream: Long, length: Int, version: Long): GetPubkey {
+        @JvmStatic
+        fun read(`is`: InputStream, stream: Long, length: Int, version: Long): GetPubkey {
             return GetPubkey(version, stream, Decode.bytes(`is`, length))
         }
     }

core/src/main/kotlin/ch/dissem/bitmessage/entity/payload/Msg.kt

@@ -16,10 +16,8 @@
 
 package ch.dissem.bitmessage.entity.payload
 
-import ch.dissem.bitmessage.entity.Encrypted
+import ch.dissem.bitmessage.entity.*
-import ch.dissem.bitmessage.entity.Plaintext
 import ch.dissem.bitmessage.entity.Plaintext.Type.MSG
-import ch.dissem.bitmessage.entity.PlaintextHolder
 import ch.dissem.bitmessage.exception.DecryptionFailedException
 import java.io.InputStream
 import java.io.OutputStream
@@ -51,10 +49,6 @@ class Msg : ObjectPayload, Encrypted, PlaintextHolder {
 
     override val isSigned: Boolean = true
 
-    override fun writeBytesToSign(out: OutputStream) {
-        plaintext?.write(out, false) ?: throw IllegalStateException("no plaintext data available")
-    }
-
     override var signature: ByteArray?
         get() = plaintext?.signature
         set(signature) {
@@ -73,14 +67,6 @@ class Msg : ObjectPayload, Encrypted, PlaintextHolder {
     override val isDecrypted: Boolean
         get() = plaintext != null
 
-    override fun write(out: OutputStream) {
-        encrypted?.write(out) ?: throw IllegalStateException("Msg must be signed and encrypted before writing it.")
-    }
-
-    override fun write(buffer: ByteBuffer) {
-        encrypted?.write(buffer) ?: throw IllegalStateException("Msg must be signed and encrypted before writing it.")
-    }
-
     override fun equals(other: Any?): Boolean {
         if (this === other) return true
         if (other !is Msg) return false
@@ -89,14 +75,35 @@ class Msg : ObjectPayload, Encrypted, PlaintextHolder {
         return stream == other.stream && (encrypted == other.encrypted || plaintext == other.plaintext)
     }
 
-    override fun hashCode(): Int {
+    override fun hashCode() = stream.toInt()
-        return stream.toInt()
+
+    override fun writer(): SignedStreamableWriter = Writer(this)
+
+    private class Writer(
+        private val item: Msg
+    ) : SignedStreamableWriter {
+
+        val encryptedDataWriter = item.encrypted?.writer()
+
+        override fun write(out: OutputStream) {
+            encryptedDataWriter?.write(out) ?: throw IllegalStateException("Msg must be signed and encrypted before writing it.")
+        }
+
+        override fun write(buffer: ByteBuffer) {
+            encryptedDataWriter?.write(buffer) ?: throw IllegalStateException("Msg must be signed and encrypted before writing it.")
+        }
+
+        override fun writeBytesToSign(out: OutputStream) {
+            item.plaintext?.writer(false)?.write(out) ?: throw IllegalStateException("no plaintext data available")
+        }
+
     }
 
     companion object {
         val ACK_LENGTH = 32
 
-        @JvmStatic fun read(`in`: InputStream, stream: Long, length: Int): Msg {
+        @JvmStatic
+        fun read(`in`: InputStream, stream: Long, length: Int): Msg {
             return Msg(stream, CryptoBox.read(`in`, length))
         }
     }

core/src/main/kotlin/ch/dissem/bitmessage/entity/payload/ObjectPayload.kt

@@ -17,13 +17,14 @@
 package ch.dissem.bitmessage.entity.payload
 
 import ch.dissem.bitmessage.entity.ObjectMessage
+import ch.dissem.bitmessage.entity.SignedStreamable
 import ch.dissem.bitmessage.entity.Streamable
 import java.io.OutputStream
 
 /**
  * The payload of an 'object' command. This is shared by the network.
  */
-abstract class ObjectPayload protected constructor(val version: Long) : Streamable {
+abstract class ObjectPayload protected constructor(val version: Long) : SignedStreamable {
 
     abstract val type: ObjectType?
 
@@ -31,10 +32,6 @@ abstract class ObjectPayload protected constructor(val version: Long) : Streamab
 
     open val isSigned: Boolean = false
 
-    open fun writeBytesToSign(out: OutputStream) {
-        // nothing to do
-    }
-
     /**
      * @return the ECDSA signature which, as of protocol v3, covers the object header starting with the time,
      * * appended with the data described in this table down to the extra_bytes. Therefore, this must

core/src/main/kotlin/ch/dissem/bitmessage/entity/payload/Pubkey.kt

@@ -18,6 +18,8 @@ package ch.dissem.bitmessage.entity.payload
 
 import ch.dissem.bitmessage.InternalContext.Companion.NETWORK_EXTRA_BYTES
 import ch.dissem.bitmessage.InternalContext.Companion.NETWORK_NONCE_TRIALS_PER_BYTE
+import ch.dissem.bitmessage.entity.EncryptedStreamableWriter
+import ch.dissem.bitmessage.entity.SignedStreamableWriter
 import ch.dissem.bitmessage.utils.Singleton.cryptography
 import java.io.OutputStream
 import java.nio.ByteBuffer
@@ -42,13 +44,7 @@ abstract class Pubkey protected constructor(version: Long) : ObjectPayload(versi
 
     open val extraBytes: Long = NETWORK_EXTRA_BYTES
 
-    open fun writeUnencrypted(out: OutputStream) {
+    abstract override fun writer(): EncryptedStreamableWriter
-        write(out)
-    }
-
-    open fun writeUnencrypted(buffer: ByteBuffer) {
-        write(buffer)
-    }
 
     /**
      * Bits 0 through 29 are yet undefined

core/src/main/kotlin/ch/dissem/bitmessage/entity/payload/V2Pubkey.kt

@@ -16,6 +16,7 @@
 
 package ch.dissem.bitmessage.entity.payload
 
+import ch.dissem.bitmessage.entity.EncryptedStreamableWriter
 import ch.dissem.bitmessage.utils.Decode
 import ch.dissem.bitmessage.utils.Encode
 import java.io.InputStream
@@ -25,21 +26,47 @@ import java.nio.ByteBuffer
 /**
  * A version 2 public key.
  */
-open class V2Pubkey constructor(version: Long, override val stream: Long, override val behaviorBitfield: Int, signingKey: ByteArray, encryptionKey: ByteArray) : Pubkey(version) {
+open class V2Pubkey constructor(
+    version: Long,
+    override val stream: Long,
+    override val behaviorBitfield: Int,
+    signingKey: ByteArray,
+    encryptionKey: ByteArray
+) : Pubkey(version) {
 
     override val signingKey: ByteArray = if (signingKey.size == 64) add0x04(signingKey) else signingKey
     override val encryptionKey: ByteArray = if (encryptionKey.size == 64) add0x04(encryptionKey) else encryptionKey
 
-    override fun write(out: OutputStream) {
+    override fun writer(): EncryptedStreamableWriter = Writer(this)
-        Encode.int32(behaviorBitfield, out)
+
-        out.write(signingKey, 1, 64)
+    protected open class Writer(
-        out.write(encryptionKey, 1, 64)
+        private val item: V2Pubkey
-    }
+    ) : EncryptedStreamableWriter {
+
+        override fun write(out: OutputStream) {
+            Encode.int32(item.behaviorBitfield, out)
+            out.write(item.signingKey, 1, 64)
+            out.write(item.encryptionKey, 1, 64)
+        }
+
+        override fun write(buffer: ByteBuffer) {
+            Encode.int32(item.behaviorBitfield, buffer)
+            buffer.put(item.signingKey, 1, 64)
+            buffer.put(item.encryptionKey, 1, 64)
+        }
+
+        override fun writeBytesToSign(out: OutputStream) {
+            // Nothing to do
+        }
+
+        override fun writeUnencrypted(out: OutputStream) {
+            write(out)
+        }
+
+        override fun writeUnencrypted(buffer: ByteBuffer) {
+            write(buffer)
+        }
 
-    override fun write(buffer: ByteBuffer) {
-        Encode.int32(behaviorBitfield, buffer)
-        buffer.put(signingKey, 1, 64)
-        buffer.put(encryptionKey, 1, 64)
     }
 
     class Builder {

core/src/main/kotlin/ch/dissem/bitmessage/entity/payload/V3Pubkey.kt

@@ -16,6 +16,9 @@
 
 package ch.dissem.bitmessage.entity.payload
 
+import ch.dissem.bitmessage.entity.EncryptedStreamableWriter
+import ch.dissem.bitmessage.entity.SignedStreamableWriter
+import ch.dissem.bitmessage.entity.StreamableWriter
 import ch.dissem.bitmessage.utils.Decode
 import ch.dissem.bitmessage.utils.Encode
 import java.io.InputStream
@@ -34,32 +37,8 @@ class V3Pubkey protected constructor(
     override var signature: ByteArray? = null
 ) : V2Pubkey(version, stream, behaviorBitfield, signingKey, encryptionKey) {
 
-    override fun write(out: OutputStream) {
-        writeBytesToSign(out)
-        Encode.varBytes(
-            signature ?: throw IllegalStateException("signature not available"),
-            out
-        )
-    }
-
-    override fun write(buffer: ByteBuffer) {
-        super.write(buffer)
-        Encode.varInt(nonceTrialsPerByte, buffer)
-        Encode.varInt(extraBytes, buffer)
-        Encode.varBytes(
-            signature ?: throw IllegalStateException("signature not available"),
-            buffer
-        )
-    }
-
     override val isSigned: Boolean = true
 
-    override fun writeBytesToSign(out: OutputStream) {
-        super.write(out)
-        Encode.varInt(nonceTrialsPerByte, out)
-        Encode.varInt(extraBytes, out)
-    }
-
     override fun equals(other: Any?): Boolean {
         if (this === other) return true
         if (other !is V3Pubkey) return false
@@ -75,6 +54,37 @@ class V3Pubkey protected constructor(
         return Objects.hash(nonceTrialsPerByte, extraBytes)
     }
 
+    override fun writer(): EncryptedStreamableWriter = Writer(this)
+
+    protected open class Writer(
+        private val item: V3Pubkey
+    ) : V2Pubkey.Writer(item) {
+
+        override fun write(out: OutputStream) {
+            writeBytesToSign(out)
+            Encode.varBytes(
+                item.signature ?: throw IllegalStateException("signature not available"),
+                out
+            )
+        }
+
+        override fun write(buffer: ByteBuffer) {
+            super.write(buffer)
+            Encode.varInt(item.nonceTrialsPerByte, buffer)
+            Encode.varInt(item.extraBytes, buffer)
+            Encode.varBytes(
+                item.signature ?: throw IllegalStateException("signature not available"),
+                buffer
+            )
+        }
+
+        override fun writeBytesToSign(out: OutputStream) {
+            super.write(out)
+            Encode.varInt(item.nonceTrialsPerByte, out)
+            Encode.varInt(item.extraBytes, out)
+        }
+    }
+
     class Builder {
         private var streamNumber: Long = 0
         private var behaviorBitfield: Int = 0

core/src/main/kotlin/ch/dissem/bitmessage/entity/payload/V4Broadcast.kt

@@ -18,7 +18,7 @@ package ch.dissem.bitmessage.entity.payload
 
 import ch.dissem.bitmessage.entity.BitmessageAddress
 import ch.dissem.bitmessage.entity.Plaintext
-
+import ch.dissem.bitmessage.entity.SignedStreamableWriter
 import java.io.InputStream
 import java.io.OutputStream
 import java.nio.ByteBuffer
@@ -38,21 +38,28 @@ open class V4Broadcast : Broadcast {
             throw IllegalArgumentException("Address version 3 or older expected, but was " + senderAddress.version)
     }
 
+    override fun writer(): SignedStreamableWriter = Writer(this)
 
-    override fun writeBytesToSign(out: OutputStream) {
+    protected open class Writer(
-        plaintext?.write(out, false) ?: throw IllegalStateException("no plaintext data available")
+        private val item: V4Broadcast
-    }
+    ) : SignedStreamableWriter {
 
-    override fun write(out: OutputStream) {
+        override fun writeBytesToSign(out: OutputStream) {
-        encrypted?.write(out) ?: throw IllegalStateException("broadcast not encrypted")
+            item.plaintext?.writer(false)?.write(out) ?: throw IllegalStateException("no plaintext data available")
-    }
+        }
 
-    override fun write(buffer: ByteBuffer) {
+        override fun write(out: OutputStream) {
-        encrypted?.write(buffer) ?: throw IllegalStateException("broadcast not encrypted")
+            item.encrypted?.writer()?.write(out) ?: throw IllegalStateException("broadcast not encrypted")
+        }
+
+        override fun write(buffer: ByteBuffer) {
+            item.encrypted?.writer()?.write(buffer) ?: throw IllegalStateException("broadcast not encrypted")
+        }
     }
 
     companion object {
-        @JvmStatic fun read(`in`: InputStream, stream: Long, length: Int): V4Broadcast {
+        @JvmStatic
+        fun read(`in`: InputStream, stream: Long, length: Int): V4Broadcast {
             return V4Broadcast(4, stream, CryptoBox.read(`in`, length), null)
         }
     }

core/src/main/kotlin/ch/dissem/bitmessage/entity/payload/V4Pubkey.kt

@@ -18,6 +18,7 @@ package ch.dissem.bitmessage.entity.payload
 
 import ch.dissem.bitmessage.entity.BitmessageAddress
 import ch.dissem.bitmessage.entity.Encrypted
+import ch.dissem.bitmessage.entity.EncryptedStreamableWriter
 import ch.dissem.bitmessage.exception.DecryptionFailedException
 import ch.dissem.bitmessage.utils.Decode
 import java.io.InputStream
@@ -63,29 +64,6 @@ class V4Pubkey : Pubkey, Encrypted {
     override val isDecrypted: Boolean
         get() = decrypted != null
 
-    override fun write(out: OutputStream) {
-        out.write(tag)
-        encrypted?.write(out) ?: throw IllegalStateException("pubkey is encrypted")
-    }
-
-    override fun write(buffer: ByteBuffer) {
-        buffer.put(tag)
-        encrypted?.write(buffer) ?: throw IllegalStateException("pubkey is encrypted")
-    }
-
-    override fun writeUnencrypted(out: OutputStream) {
-        decrypted?.write(out) ?: throw IllegalStateException("pubkey is encrypted")
-    }
-
-    override fun writeUnencrypted(buffer: ByteBuffer) {
-        decrypted?.write(buffer) ?: throw IllegalStateException("pubkey is encrypted")
-    }
-
-    override fun writeBytesToSign(out: OutputStream) {
-        out.write(tag)
-        decrypted?.writeBytesToSign(out) ?: throw IllegalStateException("pubkey is encrypted")
-    }
-
     override val signingKey: ByteArray
         get() = decrypted?.signingKey ?: throw IllegalStateException("pubkey is encrypted")
 
@@ -126,8 +104,40 @@ class V4Pubkey : Pubkey, Encrypted {
         return result
     }
 
+    override fun writer(): EncryptedStreamableWriter = Writer(this)
+
+    private class Writer(
+        val item: V4Pubkey
+    ) : EncryptedStreamableWriter {
+
+        override fun write(out: OutputStream) {
+            out.write(item.tag)
+            item.encrypted?.writer()?.write(out) ?: throw IllegalStateException("pubkey is encrypted")
+        }
+
+        override fun write(buffer: ByteBuffer) {
+            buffer.put(item.tag)
+            item.encrypted?.writer()?.write(buffer) ?: throw IllegalStateException("pubkey is encrypted")
+        }
+
+        override fun writeUnencrypted(out: OutputStream) {
+            item.decrypted?.writer()?.write(out) ?: throw IllegalStateException("pubkey is encrypted")
+        }
+
+        override fun writeUnencrypted(buffer: ByteBuffer) {
+            item.decrypted?.writer()?.write(buffer) ?: throw IllegalStateException("pubkey is encrypted")
+        }
+
+        override fun writeBytesToSign(out: OutputStream) {
+            out.write(item.tag)
+            item.decrypted?.writer()?.writeBytesToSign(out) ?: throw IllegalStateException("pubkey is encrypted")
+        }
+
+    }
+
     companion object {
-        @JvmStatic fun read(`in`: InputStream, stream: Long, length: Int, encrypted: Boolean): V4Pubkey {
+        @JvmStatic
+        fun read(`in`: InputStream, stream: Long, length: Int, encrypted: Boolean): V4Pubkey {
             if (encrypted)
                 return V4Pubkey(stream,
                     Decode.bytes(`in`, 32),

core/src/main/kotlin/ch/dissem/bitmessage/entity/payload/V5Broadcast.kt

@@ -18,6 +18,7 @@ package ch.dissem.bitmessage.entity.payload
 
 import ch.dissem.bitmessage.entity.BitmessageAddress
 import ch.dissem.bitmessage.entity.Plaintext
+import ch.dissem.bitmessage.entity.SignedStreamableWriter
 import ch.dissem.bitmessage.utils.Decode
 
 import java.io.InputStream
@@ -40,18 +41,27 @@ class V5Broadcast : V4Broadcast {
         this.tag = senderAddress.tag ?: throw IllegalStateException("version 4 address without tag")
     }
 
-    override fun writeBytesToSign(out: OutputStream) {
+    override fun writer(): SignedStreamableWriter = Writer(this)
-        out.write(tag)
+
-        super.writeBytesToSign(out)
+    private class Writer(
-    }
+        private val item: V5Broadcast
+    ) : V4Broadcast.Writer(item) {
+
+        override fun writeBytesToSign(out: OutputStream) {
+            out.write(item.tag)
+            super.writeBytesToSign(out)
+        }
+
+        override fun write(out: OutputStream) {
+            out.write(item.tag)
+            super.write(out)
+        }
 
-    override fun write(out: OutputStream) {
-        out.write(tag)
-        super.write(out)
     }
 
     companion object {
-        @JvmStatic fun read(`is`: InputStream, stream: Long, length: Int): V5Broadcast {
+        @JvmStatic
+        fun read(`is`: InputStream, stream: Long, length: Int): V5Broadcast {
             return V5Broadcast(stream, Decode.bytes(`is`, 32), CryptoBox.read(`is`, length - 32))
         }
     }

core/src/main/kotlin/ch/dissem/bitmessage/entity/valueobject/InventoryVector.kt

@@ -17,6 +17,7 @@
 package ch.dissem.bitmessage.entity.valueobject
 
 import ch.dissem.bitmessage.entity.Streamable
+import ch.dissem.bitmessage.entity.StreamableWriter
 import ch.dissem.bitmessage.utils.Strings
 import java.io.OutputStream
 import java.nio.ByteBuffer
@@ -39,20 +40,29 @@ data class InventoryVector constructor(
         return Arrays.hashCode(hash)
     }
 
-    override fun write(out: OutputStream) {
-        out.write(hash)
-    }
-
-    override fun write(buffer: ByteBuffer) {
-        buffer.put(hash)
-    }
-
     override fun toString(): String {
         return Strings.hex(hash)
     }
 
+    override fun writer(): StreamableWriter = Writer(this)
+
+    private class Writer(
+        private val item: InventoryVector
+    ) : StreamableWriter {
+
+        override fun write(out: OutputStream) {
+            out.write(item.hash)
+        }
+
+        override fun write(buffer: ByteBuffer) {
+            buffer.put(item.hash)
+        }
+
+    }
+
     companion object {
-        @JvmStatic fun fromHash(hash: ByteArray?): InventoryVector? {
+        @JvmStatic
+        fun fromHash(hash: ByteArray?): InventoryVector? {
             return InventoryVector(
                 hash ?: return null
             )

core/src/main/kotlin/ch/dissem/bitmessage/entity/valueobject/NetworkAddress.kt

@@ -17,6 +17,7 @@
 package ch.dissem.bitmessage.entity.valueobject
 
 import ch.dissem.bitmessage.entity.Streamable
+import ch.dissem.bitmessage.entity.StreamableWriter
 import ch.dissem.bitmessage.entity.Version
 import ch.dissem.bitmessage.utils.Encode
 import ch.dissem.bitmessage.utils.UnixTime
@@ -77,9 +78,7 @@ data class NetworkAddress(
 
     fun provides(service: Version.Service?): Boolean = service?.isEnabled(services) ?: false
 
-    fun toInetAddress(): InetAddress {
+    fun toInetAddress() = InetAddress.getByAddress(IPv6)
-        return InetAddress.getByAddress(IPv6)
-    }
 
     override fun equals(other: Any?): Boolean {
         if (this === other) return true
@@ -98,32 +97,40 @@ data class NetworkAddress(
         return "[" + toInetAddress() + "]:" + port
     }
 
-    override fun write(out: OutputStream) {
+    fun writer(light: Boolean): StreamableWriter = Writer(
-        write(out, false)
+        item = this,
-    }
+        light = light
+    )
 
-    fun write(out: OutputStream, light: Boolean) {
+    override fun writer(): StreamableWriter = Writer(
-        if (!light) {
+        item = this
-            Encode.int64(time, out)
+    )
-            Encode.int32(stream, out)
+
+    private class Writer(
+        private val item: NetworkAddress,
+        private val light: Boolean = false
+    ) : StreamableWriter {
+
+        override fun write(out: OutputStream) {
+            if (!light) {
+                Encode.int64(item.time, out)
+                Encode.int32(item.stream, out)
+            }
+            Encode.int64(item.services, out)
+            out.write(item.IPv6)
+            Encode.int16(item.port, out)
         }
-        Encode.int64(services, out)
-        out.write(IPv6)
-        Encode.int16(port, out)
-    }
 
-    override fun write(buffer: ByteBuffer) {
+        override fun write(buffer: ByteBuffer) {
-        write(buffer, false)
+            if (!light) {
-    }
+                Encode.int64(item.time, buffer)
-
+                Encode.int32(item.stream, buffer)
-    fun write(buffer: ByteBuffer, light: Boolean) {
+            }
-        if (!light) {
+            Encode.int64(item.services, buffer)
-            Encode.int64(time, buffer)
+            buffer.put(item.IPv6)
-            Encode.int32(stream, buffer)
+            Encode.int16(item.port, buffer)
         }
-        Encode.int64(services, buffer)
+
-        buffer.put(IPv6)
-        Encode.int16(port, buffer)
     }
 
     class Builder {
@@ -194,6 +201,7 @@ data class NetworkAddress(
     }
 
     companion object {
-        @JvmField val ANY = NetworkAddress(time = 0, stream = 0, services = 0, IPv6 = ByteArray(16), port = 0)
+        @JvmField
+        val ANY = NetworkAddress(time = 0, stream = 0, services = 0, IPv6 = ByteArray(16), port = 0)
     }
 }

core/src/main/kotlin/ch/dissem/bitmessage/entity/valueobject/PrivateKey.kt

@@ -19,6 +19,7 @@ package ch.dissem.bitmessage.entity.valueobject
 import ch.dissem.bitmessage.InternalContext
 import ch.dissem.bitmessage.entity.BitmessageAddress
 import ch.dissem.bitmessage.entity.Streamable
+import ch.dissem.bitmessage.entity.StreamableWriter
 import ch.dissem.bitmessage.entity.payload.Pubkey
 import ch.dissem.bitmessage.exception.ApplicationException
 import ch.dissem.bitmessage.factory.Factory
@@ -66,7 +67,52 @@ data class PrivateKey(
             builder.nonceTrialsPerByte, builder.extraBytes, *builder.features)
     )
 
-    private class Builder internal constructor(internal val version: Long, internal val stream: Long, internal val shorter: Boolean) {
+    override fun equals(other: Any?) = other is PrivateKey
+        && Arrays.equals(privateEncryptionKey, other.privateEncryptionKey)
+        && Arrays.equals(privateSigningKey, other.privateSigningKey)
+        && pubkey == other.pubkey
+
+    override fun hashCode() = pubkey.hashCode()
+
+    override fun writer(): StreamableWriter = Writer(this)
+
+    private class Writer(
+        private val item: PrivateKey
+    ) : StreamableWriter {
+
+        override fun write(out: OutputStream) {
+            Encode.varInt(item.pubkey.version, out)
+            Encode.varInt(item.pubkey.stream, out)
+            val baos = ByteArrayOutputStream()
+            item.pubkey.writer().writeUnencrypted(baos)
+            Encode.varInt(baos.size(), out)
+            out.write(baos.toByteArray())
+            Encode.varBytes(item.privateSigningKey, out)
+            Encode.varBytes(item.privateEncryptionKey, out)
+        }
+
+        override fun write(buffer: ByteBuffer) {
+            Encode.varInt(item.pubkey.version, buffer)
+            Encode.varInt(item.pubkey.stream, buffer)
+            try {
+                val baos = ByteArrayOutputStream()
+                item.pubkey.writer().writeUnencrypted(baos)
+                Encode.varBytes(baos.toByteArray(), buffer)
+            } catch (e: IOException) {
+                throw ApplicationException(e)
+            }
+
+            Encode.varBytes(item.privateSigningKey, buffer)
+            Encode.varBytes(item.privateEncryptionKey, buffer)
+        }
+
+    }
+
+    private class Builder internal constructor(
+        internal val version: Long,
+        internal val stream: Long,
+        internal val shorter: Boolean
+    ) {
 
         internal var seed: ByteArray? = null
         internal var nextNonce: Long = 0
@@ -129,44 +175,12 @@ data class PrivateKey(
         }
     }
 
-    override fun write(out: OutputStream) {
-        Encode.varInt(pubkey.version, out)
-        Encode.varInt(pubkey.stream, out)
-        val baos = ByteArrayOutputStream()
-        pubkey.writeUnencrypted(baos)
-        Encode.varInt(baos.size(), out)
-        out.write(baos.toByteArray())
-        Encode.varBytes(privateSigningKey, out)
-        Encode.varBytes(privateEncryptionKey, out)
-    }
-
-
-    override fun write(buffer: ByteBuffer) {
-        Encode.varInt(pubkey.version, buffer)
-        Encode.varInt(pubkey.stream, buffer)
-        try {
-            val baos = ByteArrayOutputStream()
-            pubkey.writeUnencrypted(baos)
-            Encode.varBytes(baos.toByteArray(), buffer)
-        } catch (e: IOException) {
-            throw ApplicationException(e)
-        }
-
-        Encode.varBytes(privateSigningKey, buffer)
-        Encode.varBytes(privateEncryptionKey, buffer)
-    }
-
-    override fun equals(other: Any?) = other is PrivateKey
-        && Arrays.equals(privateEncryptionKey, other.privateEncryptionKey)
-        && Arrays.equals(privateSigningKey, other.privateSigningKey)
-        && pubkey == other.pubkey
-
-    override fun hashCode() = pubkey.hashCode()
-
     companion object {
-        @JvmField val PRIVATE_KEY_SIZE = 32
+        @JvmField
+        val PRIVATE_KEY_SIZE = 32
 
-        @JvmStatic fun deterministic(passphrase: String, numberOfAddresses: Int, version: Long, stream: Long, shorter: Boolean): List<PrivateKey> {
+        @JvmStatic
+        fun deterministic(passphrase: String, numberOfAddresses: Int, version: Long, stream: Long, shorter: Boolean): List<PrivateKey> {
             val result = ArrayList<PrivateKey>(numberOfAddresses)
             val builder = Builder(version, stream, shorter).seed(passphrase)
             for (i in 0..numberOfAddresses - 1) {
@@ -176,7 +190,8 @@ data class PrivateKey(
             return result
         }
 
-        @JvmStatic fun read(`is`: InputStream): PrivateKey {
+        @JvmStatic
+        fun read(`is`: InputStream): PrivateKey {
             val version = Decode.varInt(`is`).toInt()
             val stream = Decode.varInt(`is`)
             val len = Decode.varInt(`is`).toInt()

core/src/main/kotlin/ch/dissem/bitmessage/entity/valueobject/extended/Message.kt

@@ -45,25 +45,25 @@ data class Message constructor(
 
     override fun pack(): MPMap<MPString, MPType<*>> {
         val result = MPMap<MPString, MPType<*>>()
-        result.put(mp(""), mp(TYPE))
+        result.put("".mp, TYPE.mp)
-        result.put(mp("subject"), mp(subject))
+        result.put("subject".mp, subject.mp)
-        result.put(mp("body"), mp(body))
+        result.put("body".mp, body.mp)
 
         if (!files.isEmpty()) {
             val items = MPArray<MPMap<MPString, MPType<*>>>()
-            result.put(mp("files"), items)
+            result.put("files".mp, items)
             for (file in files) {
                 val item = MPMap<MPString, MPType<*>>()
-                item.put(mp("name"), mp(file.name))
+                item.put("name".mp, file.name.mp)
-                item.put(mp("data"), mp(*file.data))
+                item.put("data".mp, file.data.mp)
-                item.put(mp("type"), mp(file.type))
+                item.put("type".mp, file.type.mp)
-                item.put(mp("disposition"), mp(file.disposition.name))
+                item.put("disposition".mp, file.disposition.name.mp)
                 items.add(item)
             }
         }
         if (!parents.isEmpty()) {
             val items = MPArray<MPBinary>()
-            result.put(mp("parents"), items)
+            result.put("parents".mp, items)
             for ((hash) in parents) {
                 items.add(mp(*hash))
             }
@@ -139,26 +139,26 @@ data class Message constructor(
         override val type: String = TYPE
 
         override fun unpack(map: MPMap<MPString, MPType<*>>): Message {
-            val subject = str(map[mp("subject")]) ?: ""
+            val subject = str(map["subject".mp]) ?: ""
-            val body = str(map[mp("body")]) ?: ""
+            val body = str(map["body".mp]) ?: ""
             val parents = LinkedList<InventoryVector>()
             val files = LinkedList<Attachment>()
-            val mpParents = map[mp("parents")] as? MPArray<*>
+            val mpParents = map["parents".mp] as? MPArray<*>
             for (parent in mpParents ?: emptyList<MPArray<MPBinary>>()) {
                 parents.add(InventoryVector.fromHash(
                     (parent as? MPBinary)?.value ?: continue
                 ) ?: continue)
             }
-            val mpFiles = map[mp("files")] as? MPArray<*>
+            val mpFiles = map["files".mp] as? MPArray<*>
             for (item in mpFiles ?: emptyList<Any>()) {
                 if (item is MPMap<*, *>) {
                     val b = Attachment.Builder()
-                    b.name(str(item[mp("name")])!!)
+                    b.name(str(item["name".mp])!!)
                     b.data(
-                        bin(item[mp("data")] ?: continue) ?: continue
+                        bin(item["data".mp] ?: continue) ?: continue
                     )
-                    b.type(str(item[mp("type")])!!)
+                    b.type(str(item["type".mp])!!)
-                    val disposition = str(item[mp("disposition")])
+                    val disposition = str(item["disposition".mp])
                     if ("inline" == disposition) {
                         b.inline()
                     } else if ("attachment" == disposition) {

core/src/main/kotlin/ch/dissem/bitmessage/entity/valueobject/extended/Vote.kt

@@ -35,9 +35,9 @@ data class Vote constructor(val msgId: InventoryVector, val vote: String) : Exte
 
     override fun pack(): MPMap<MPString, MPType<*>> {
         val result = MPMap<MPString, MPType<*>>()
-        result.put(mp(""), mp(TYPE))
+        result.put("".mp, TYPE.mp)
-        result.put(mp("msgId"), mp(*msgId.hash))
+        result.put("msgId".mp, msgId.hash.mp)
-        result.put(mp("vote"), mp(vote))
+        result.put("vote".mp, vote.mp)
         return result
     }
 
@@ -77,13 +77,14 @@ data class Vote constructor(val msgId: InventoryVector, val vote: String) : Exte
             get() = TYPE
 
         override fun unpack(map: MPMap<MPString, MPType<*>>): Vote {
-            val msgId = InventoryVector.fromHash((map[mp("msgId")] as? MPBinary)?.value) ?: throw IllegalArgumentException("data doesn't contain proper msgId")
+            val msgId = InventoryVector.fromHash((map["msgId".mp] as? MPBinary)?.value) ?: throw IllegalArgumentException("data doesn't contain proper msgId")
-            val vote = str(map[mp("vote")]) ?: throw IllegalArgumentException("no vote given")
+            val vote = str(map["vote".mp]) ?: throw IllegalArgumentException("no vote given")
             return Vote(msgId, vote)
         }
     }
 
     companion object {
-        @JvmField val TYPE = "vote"
+        @JvmField
+        val TYPE = "vote"
     }
 }

core/src/main/kotlin/ch/dissem/bitmessage/factory/ExtendedEncodingFactory.kt

@@ -54,9 +54,8 @@ object ExtendedEncodingFactory {
     fun unzip(zippedData: ByteArray): ExtendedEncoding? {
         try {
             InflaterInputStream(ByteArrayInputStream(zippedData)).use { unzipper ->
-                val reader = Reader.getInstance()
                 @Suppress("UNCHECKED_CAST")
-                val map = reader.read(unzipper) as MPMap<MPString, MPType<*>>
+                val map = Reader.read(unzipper) as MPMap<MPString, MPType<*>>
                 val messageType = map[KEY_MESSAGE_TYPE]
                 if (messageType == null) {
                     LOG.error("Missing message type")

core/src/main/kotlin/ch/dissem/bitmessage/ports/NodeRegistry.kt

@@ -31,4 +31,13 @@ interface NodeRegistry {
     fun getKnownAddresses(limit: Int, vararg streams: Long): List<NetworkAddress>
 
     fun offerAddresses(nodes: List<NetworkAddress>)
+
+    fun update(node: NetworkAddress)
+
+    fun remove(node: NetworkAddress)
+
+    /**
+     * Remove stale nodes
+     */
+    fun cleanup()
 }

core/src/main/kotlin/ch/dissem/bitmessage/utils/DebugUtils.kt

@@ -29,7 +29,7 @@ object DebugUtils {
         try {
             val f = File(System.getProperty("user.home") + "/jabit.error/" + objectMessage.inventoryVector + ".inv")
             f.createNewFile()
-            objectMessage.write(FileOutputStream(f))
+            objectMessage.writer().write(FileOutputStream(f))
         } catch (e: IOException) {
             LOG.debug(e.message, e)
         }

core/src/main/kotlin/ch/dissem/bitmessage/utils/Encode.kt

@@ -26,22 +26,27 @@ import java.nio.ByteBuffer
  * https://bitmessage.org/wiki/Protocol_specification#Common_structures
  */
 object Encode {
-    @JvmStatic fun varIntList(values: LongArray, stream: OutputStream) {
+    @JvmStatic
+    fun varIntList(values: LongArray, stream: OutputStream) {
         varInt(values.size, stream)
         for (value in values) {
             varInt(value, stream)
         }
     }
 
-    @JvmStatic fun varIntList(values: LongArray, buffer: ByteBuffer) {
+    @JvmStatic
+    fun varIntList(values: LongArray, buffer: ByteBuffer) {
         varInt(values.size, buffer)
         for (value in values) {
             varInt(value, buffer)
         }
     }
 
-    @JvmStatic fun varInt(value: Int, buffer: ByteBuffer) = varInt(value.toLong(), buffer)
+    @JvmStatic
-    @JvmStatic fun varInt(value: Long, buffer: ByteBuffer) {
+    fun varInt(value: Int, buffer: ByteBuffer) = varInt(value.toLong(), buffer)
+
+    @JvmStatic
+    fun varInt(value: Long, buffer: ByteBuffer) {
         if (value < 0) {
             // This is due to the fact that Java doesn't really support unsigned values.
             // Please be aware that this might be an error due to a smaller negative value being cast to long.
@@ -63,16 +68,24 @@ object Encode {
         }
     }
 
-    @JvmStatic fun varInt(value: Int) = varInt(value.toLong())
+    @JvmStatic
-    @JvmStatic fun varInt(value: Long): ByteArray {
+    fun varInt(value: Int) = varInt(value.toLong())
+
+    @JvmStatic
+    fun varInt(value: Long): ByteArray {
         val buffer = ByteBuffer.allocate(9)
         varInt(value, buffer)
         buffer.flip()
         return Bytes.truncate(buffer.array(), buffer.limit())
     }
 
-    @JvmStatic @JvmOverloads fun varInt(value: Int, stream: OutputStream, counter: AccessCounter? = null) = varInt(value.toLong(), stream, counter)
+    @JvmStatic
-    @JvmStatic @JvmOverloads fun varInt(value: Long, stream: OutputStream, counter: AccessCounter? = null) {
+    @JvmOverloads
+    fun varInt(value: Int, stream: OutputStream, counter: AccessCounter? = null) = varInt(value.toLong(), stream, counter)
+
+    @JvmStatic
+    @JvmOverloads
+    fun varInt(value: Long, stream: OutputStream, counter: AccessCounter? = null) {
         val buffer = ByteBuffer.allocate(9)
         varInt(value, buffer)
         buffer.flip()
@@ -80,46 +93,76 @@ object Encode {
         AccessCounter.inc(counter, buffer.limit())
     }
 
-    @JvmStatic @JvmOverloads fun int8(value: Long, stream: OutputStream, counter: AccessCounter? = null) = int8(value.toInt(), stream, counter)
+    @JvmStatic
-    @JvmStatic @JvmOverloads fun int8(value: Int, stream: OutputStream, counter: AccessCounter? = null) {
+    @JvmOverloads
+    fun int8(value: Long, stream: OutputStream, counter: AccessCounter? = null) = int8(value.toInt(), stream, counter)
+
+    @JvmStatic
+    @JvmOverloads
+    fun int8(value: Int, stream: OutputStream, counter: AccessCounter? = null) {
         stream.write(value)
         AccessCounter.inc(counter)
     }
 
-    @JvmStatic @JvmOverloads fun int16(value: Long, stream: OutputStream, counter: AccessCounter? = null) = int16(value.toShort(), stream, counter)
+    @JvmStatic
-    @JvmStatic @JvmOverloads fun int16(value: Int, stream: OutputStream, counter: AccessCounter? = null) = int16(value.toShort(), stream, counter)
+    @JvmOverloads
-    @JvmStatic @JvmOverloads fun int16(value: Short, stream: OutputStream, counter: AccessCounter? = null) {
+    fun int16(value: Long, stream: OutputStream, counter: AccessCounter? = null) = int16(value.toShort(), stream, counter)
+
+    @JvmStatic
+    @JvmOverloads
+    fun int16(value: Int, stream: OutputStream, counter: AccessCounter? = null) = int16(value.toShort(), stream, counter)
+
+    @JvmStatic
+    @JvmOverloads
+    fun int16(value: Short, stream: OutputStream, counter: AccessCounter? = null) {
         stream.write(ByteBuffer.allocate(2).putShort(value).array())
         AccessCounter.inc(counter, 2)
     }
 
-    @JvmStatic fun int16(value: Long, buffer: ByteBuffer) = int16(value.toShort(), buffer)
+    @JvmStatic
-    @JvmStatic fun int16(value: Int, buffer: ByteBuffer) = int16(value.toShort(), buffer)
+    fun int16(value: Long, buffer: ByteBuffer) = int16(value.toShort(), buffer)
-    @JvmStatic fun int16(value: Short, buffer: ByteBuffer) {
+
+    @JvmStatic
+    fun int16(value: Int, buffer: ByteBuffer) = int16(value.toShort(), buffer)
+
+    @JvmStatic
+    fun int16(value: Short, buffer: ByteBuffer) {
         buffer.putShort(value)
     }
 
-    @JvmStatic @JvmOverloads fun int32(value: Long, stream: OutputStream, counter: AccessCounter? = null) = int32(value.toInt(), stream, counter)
+    @JvmStatic
-    @JvmStatic @JvmOverloads fun int32(value: Int, stream: OutputStream, counter: AccessCounter? = null) {
+    @JvmOverloads
+    fun int32(value: Long, stream: OutputStream, counter: AccessCounter? = null) = int32(value.toInt(), stream, counter)
+
+    @JvmStatic
+    @JvmOverloads
+    fun int32(value: Int, stream: OutputStream, counter: AccessCounter? = null) {
         stream.write(ByteBuffer.allocate(4).putInt(value).array())
         AccessCounter.inc(counter, 4)
     }
 
-    @JvmStatic fun int32(value: Long, buffer: ByteBuffer) = int32(value.toInt(), buffer)
+    @JvmStatic
-    @JvmStatic fun int32(value: Int, buffer: ByteBuffer) {
+    fun int32(value: Long, buffer: ByteBuffer) = int32(value.toInt(), buffer)
+
+    @JvmStatic
+    fun int32(value: Int, buffer: ByteBuffer) {
         buffer.putInt(value)
     }
 
-    @JvmStatic @JvmOverloads fun int64(value: Long, stream: OutputStream, counter: AccessCounter? = null) {
+    @JvmStatic
+    @JvmOverloads
+    fun int64(value: Long, stream: OutputStream, counter: AccessCounter? = null) {
         stream.write(ByteBuffer.allocate(8).putLong(value).array())
         AccessCounter.inc(counter, 8)
     }
 
-    @JvmStatic fun int64(value: Long, buffer: ByteBuffer) {
+    @JvmStatic
+    fun int64(value: Long, buffer: ByteBuffer) {
         buffer.putLong(value)
     }
 
-    @JvmStatic fun varString(value: String, out: OutputStream) {
+    @JvmStatic
+    fun varString(value: String, out: OutputStream) {
         val bytes = value.toByteArray(charset("utf-8"))
         // Technically, it says the length in characters, but I think this one might be correct.
         // It doesn't really matter, as only ASCII characters are being used.
@@ -128,7 +171,8 @@ object Encode {
         out.write(bytes)
     }
 
-    @JvmStatic fun varString(value: String, buffer: ByteBuffer) {
+    @JvmStatic
+    fun varString(value: String, buffer: ByteBuffer) {
         val bytes = value.toByteArray()
         // Technically, it says the length in characters, but I think this one might be correct.
         // It doesn't really matter, as only ASCII characters are being used.
@@ -137,12 +181,14 @@ object Encode {
         buffer.put(bytes)
     }
 
-    @JvmStatic fun varBytes(data: ByteArray, out: OutputStream) {
+    @JvmStatic
+    fun varBytes(data: ByteArray, out: OutputStream) {
         varInt(data.size.toLong(), out)
         out.write(data)
     }
 
-    @JvmStatic fun varBytes(data: ByteArray, buffer: ByteBuffer) {
+    @JvmStatic
+    fun varBytes(data: ByteArray, buffer: ByteBuffer) {
         varInt(data.size.toLong(), buffer)
         buffer.put(data)
     }
@@ -152,9 +198,10 @@ object Encode {
      * @param streamable the object to be serialized
      * @return an array of bytes representing the given streamable object.
      */
-    @JvmStatic fun bytes(streamable: Streamable): ByteArray {
+    @JvmStatic
+    fun bytes(streamable: Streamable): ByteArray {
         val stream = ByteArrayOutputStream()
-        streamable.write(stream)
+        streamable.writer().write(stream)
         return stream.toByteArray()
     }
 
@@ -163,9 +210,10 @@ object Encode {
      * @param padding    the result will be padded such that its length is a multiple of *padding*
      * @return the bytes of the given [Streamable] object, 0-padded such that the final length is x*padding.
      */
-    @JvmStatic fun bytes(streamable: Streamable, padding: Int): ByteArray {
+    @JvmStatic
+    fun bytes(streamable: Streamable, padding: Int): ByteArray {
         val stream = ByteArrayOutputStream()
-        streamable.write(stream)
+        streamable.writer().write(stream)
         val offset = padding - stream.size() % padding
         val length = stream.size() + offset
         val result = ByteArray(length)

core/src/test/kotlin/ch/dissem/bitmessage/entity/SerializationTest.kt

@@ -86,7 +86,7 @@ class SerializationTest : TestBase() {
             .signature(ByteArray(0))
             .build()
         val out = ByteArrayOutputStream()
-        expected.write(out)
+        expected.writer().write(out)
         val `in` = ByteArrayInputStream(out.toByteArray())
         val actual = Plaintext.read(MSG, `in`)
 
@@ -111,7 +111,7 @@ class SerializationTest : TestBase() {
             .signature(ByteArray(0))
             .build()
         val out = ByteArrayOutputStream()
-        expected.write(out)
+        expected.writer().write(out)
         val `in` = ByteArrayInputStream(out.toByteArray())
         val actual = Plaintext.read(MSG, `in`)
 
@@ -136,7 +136,7 @@ class SerializationTest : TestBase() {
         assertNotNull(ackMessage1)
 
         val out = ByteArrayOutputStream()
-        expected.write(out)
+        expected.writer().write(out)
         val `in` = ByteArrayInputStream(out.toByteArray())
         val actual = Plaintext.read(MSG, `in`)
 
@@ -159,7 +159,7 @@ class SerializationTest : TestBase() {
         val inv = Inv(ivs)
         val before = NetworkMessage(inv)
         val out = ByteArrayOutputStream()
-        before.write(out)
+        before.writer().write(out)
 
         val after = Factory.getNetworkMessage(3, ByteArrayInputStream(out.toByteArray()))
         assertNotNull(after)
@@ -173,7 +173,7 @@ class SerializationTest : TestBase() {
         val objectMessage = Factory.getObjectMessage(version, `in`, data.size)
         val out = ByteArrayOutputStream()
         assertNotNull(objectMessage)
-        objectMessage!!.write(out)
+        objectMessage!!.writer().write(out)
         assertArrayEquals(data, out.toByteArray())
         assertEquals(expectedPayloadType.canonicalName, objectMessage.payload.javaClass.canonicalName)
     }

demo/src/main/java/ch/dissem/bitmessage/demo/Main.java

@@ -24,6 +24,7 @@ import ch.dissem.bitmessage.ports.NodeRegistry;
 import ch.dissem.bitmessage.repository.*;
 import ch.dissem.bitmessage.wif.WifExporter;
 import ch.dissem.bitmessage.wif.WifImporter;
+import org.jetbrains.annotations.NotNull;
 import org.kohsuke.args4j.CmdLineException;
 import org.kohsuke.args4j.CmdLineParser;
 import org.kohsuke.args4j.Option;
@@ -68,12 +69,28 @@ public class Main {
         if (options.localPort != null) {
             ctxBuilder.nodeRegistry(new NodeRegistry() {
                 @Override
-                public void clear() {
+                public void cleanup() {
                     // NO OP
                 }
 
                 @Override
-                public List<NetworkAddress> getKnownAddresses(int limit, long... streams) {
+                public void remove(@NotNull NetworkAddress node) {
+                    // NO OP
+                }
+
+                @Override
+                public void update(@NotNull NetworkAddress node) {
+                    // NO OP
+                }
+
+                @Override
+                public void clear() {
+                    // NO OP
+                }
+
+                @NotNull
+                @Override
+                public List<NetworkAddress> getKnownAddresses(int limit, @NotNull long... streams) {
                     return Arrays.stream(streams)
                         .mapToObj(s -> new NetworkAddress.Builder()
                             .ipv4(127, 0, 0, 1)
@@ -83,7 +100,7 @@ public class Main {
                 }
 
                 @Override
-                public void offerAddresses(List<NetworkAddress> nodes) {
+                public void offerAddresses(@NotNull List<NetworkAddress> nodes) {
                     LOG.info("Local node registry ignored offered addresses: " + nodes);
                 }
             });

demo/src/test/java/ch/dissem/bitmessage/TestNodeRegistry.java

@@ -18,6 +18,7 @@ package ch.dissem.bitmessage;
 
 import ch.dissem.bitmessage.entity.valueobject.NetworkAddress;
 import ch.dissem.bitmessage.ports.NodeRegistry;
+import org.jetbrains.annotations.NotNull;
 
 import java.util.LinkedList;
 import java.util.List;
@@ -44,13 +45,29 @@ class TestNodeRegistry implements NodeRegistry {
         // NO OP
     }
 
+    @NotNull
     @Override
-    public List<NetworkAddress> getKnownAddresses(int limit, long... streams) {
+    public List<NetworkAddress> getKnownAddresses(int limit, @NotNull long... streams) {
         return nodes;
     }
 
     @Override
-    public void offerAddresses(List<NetworkAddress> nodes) {
+    public void offerAddresses(@NotNull List<NetworkAddress> nodes) {
+        // Ignore
+    }
+
+    @Override
+    public void update(@NotNull NetworkAddress node) {
+        // Ignore
+    }
+
+    @Override
+    public void remove(@NotNull NetworkAddress node) {
+        // Ignore
+    }
+
+    @Override
+    public void cleanup() {
         // Ignore
     }
 }

exports/build.gradle

@@ -15,9 +15,9 @@ dependencies {
     compile 'org.slf4j:slf4j-api'
     compile 'com.beust:klaxon'
 
-    testCompile 'junit:junit:4.12'
+    testCompile 'junit:junit'
-    testCompile 'org.hamcrest:hamcrest-library:1.3'
+    testCompile 'org.hamcrest:hamcrest-library'
-    testCompile 'com.nhaarman:mockito-kotlin:1.5.0'
+    testCompile 'com.nhaarman:mockito-kotlin'
     testCompile project(path: ':core', configuration: 'testArtifacts')
     testCompile project(':cryptography-bc')
 }

exports/src/main/kotlin/ch/dissem/bitmessage/exports/ContactExport.kt

@@ -39,7 +39,7 @@ object ContactExport {
                     "subscribed" to it.isSubscribed,
                     "pubkey" to it.pubkey?.let {
                         val out = ByteArrayOutputStream()
-                        it.writeUnencrypted(out)
+                        it.writer().writeUnencrypted(out)
                         Base64.encodeToString(out.toByteArray())
                     },
                     "privateKey" to if (includePrivateKey) {

extensions/src/main/kotlin/ch/dissem/bitmessage/extensions/CryptoCustomMessage.kt

@@ -19,6 +19,7 @@ package ch.dissem.bitmessage.extensions
 import ch.dissem.bitmessage.entity.BitmessageAddress
 import ch.dissem.bitmessage.entity.CustomMessage
 import ch.dissem.bitmessage.entity.Streamable
+import ch.dissem.bitmessage.entity.StreamableWriter
 import ch.dissem.bitmessage.entity.payload.CryptoBox
 import ch.dissem.bitmessage.entity.payload.Pubkey
 import ch.dissem.bitmessage.exception.DecryptionFailedException
@@ -73,7 +74,7 @@ class CryptoCustomMessage<T : Streamable> : CustomMessage {
             Encode.varInt(privateKey.pubkey.extraBytes, out)
         }
 
-        data?.write(out) ?: throw IllegalStateException("no unencrypted data available")
+        data?.writer()?.write(out) ?: throw IllegalStateException("no unencrypted data available")
         Encode.varBytes(cryptography().getSignature(out.toByteArray(), privateKey), out)
         container = CryptoBox(out.toByteArray(), publicKey)
     }
@@ -109,9 +110,17 @@ class CryptoCustomMessage<T : Streamable> : CustomMessage {
         return data!!
     }
 
-    override fun write(out: OutputStream) {
+    override fun writer(): StreamableWriter = Writer(this)
-        Encode.varString(COMMAND, out)
+
-        container?.write(out) ?: throw IllegalStateException("not encrypted yet")
+    private class Writer(
+        private val item: CryptoCustomMessage<*>
+    ) : CustomMessage.Writer(item) {
+
+        override fun write(out: OutputStream) {
+            Encode.varString(COMMAND, out)
+            item.container?.writer()?.write(out) ?: throw IllegalStateException("not encrypted yet")
+        }
+
     }
 
     interface Reader<out T> {
@@ -136,9 +145,11 @@ class CryptoCustomMessage<T : Streamable> : CustomMessage {
     }
 
     companion object {
-        @JvmField val COMMAND = "ENCRYPTED"
+        @JvmField
+        val COMMAND = "ENCRYPTED"
 
-        @JvmStatic fun <T : Streamable> read(data: CustomMessage, dataReader: Reader<T>): CryptoCustomMessage<T> {
+        @JvmStatic
+        fun <T : Streamable> read(data: CustomMessage, dataReader: Reader<T>): CryptoCustomMessage<T> {
             val cryptoBox = CryptoBox.read(ByteArrayInputStream(data.getData()), data.getData().size)
             return CryptoCustomMessage(cryptoBox, dataReader)
         }

extensions/src/main/kotlin/ch/dissem/bitmessage/extensions/pow/ProofOfWorkRequest.kt

@@ -18,6 +18,7 @@ package ch.dissem.bitmessage.extensions.pow
 
 import ch.dissem.bitmessage.entity.BitmessageAddress
 import ch.dissem.bitmessage.entity.Streamable
+import ch.dissem.bitmessage.entity.StreamableWriter
 import ch.dissem.bitmessage.extensions.CryptoCustomMessage
 import ch.dissem.bitmessage.utils.Decode.bytes
 import ch.dissem.bitmessage.utils.Decode.varBytes
@@ -33,16 +34,24 @@ import java.util.*
  */
 data class ProofOfWorkRequest @JvmOverloads constructor(val sender: BitmessageAddress, val initialHash: ByteArray, val request: ProofOfWorkRequest.Request, val data: ByteArray = ByteArray(0)) : Streamable {
 
-    override fun write(out: OutputStream) {
+    override fun writer(): StreamableWriter = Writer(this)
-        out.write(initialHash)
+
-        Encode.varString(request.name, out)
+    private class Writer(
-        Encode.varBytes(data, out)
+        private val item: ProofOfWorkRequest
-    }
+    ) : StreamableWriter {
+
+        override fun write(out: OutputStream) {
+            out.write(item.initialHash)
+            Encode.varString(item.request.name, out)
+            Encode.varBytes(item.data, out)
+        }
+
+        override fun write(buffer: ByteBuffer) {
+            buffer.put(item.initialHash)
+            Encode.varString(item.request.name, buffer)
+            Encode.varBytes(item.data, buffer)
+        }
 
-    override fun write(buffer: ByteBuffer) {
-        buffer.put(initialHash)
-        Encode.varString(request.name, buffer)
-        Encode.varBytes(data, buffer)
     }
 
     class Reader(private val identity: BitmessageAddress) : CryptoCustomMessage.Reader<ProofOfWorkRequest> {

extensions/src/test/kotlin/ch/dissem/bitmessage/extensions/CryptoCustomMessageTest.kt

@@ -43,7 +43,7 @@ class CryptoCustomMessageTest : TestBase() {
         messageBefore.signAndEncrypt(sendingIdentity, cryptography().createPublicKey(sendingIdentity.publicDecryptionKey))
 
         val out = ByteArrayOutputStream()
-        messageBefore.write(out)
+        messageBefore.writer().write(out)
         val `in` = ByteArrayInputStream(out.toByteArray())
 
         val customMessage = CustomMessage.read(`in`, out.size())
@@ -71,7 +71,7 @@ class CryptoCustomMessageTest : TestBase() {
 
 
         val out = ByteArrayOutputStream()
-        messageBefore.write(out)
+        messageBefore.writer().write(out)
         val `in` = ByteArrayInputStream(out.toByteArray())
 
         val customMessage = CustomMessage.read(`in`, out.size())

gradle/wrapper/gradle-wrapper.properties

@@ -1,6 +1,6 @@
-#Mon Jul 17 06:32:41 CEST 2017
+#Tue Nov 07 17:21:36 CET 2017
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-4.0-all.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-4.2-all.zip

gradlew

@@ -33,11 +33,11 @@ DEFAULT_JVM_OPTS=""
 # Use the maximum available, or set MAX_FD != -1 to use that value.
 MAX_FD="maximum"
 
-warn ( ) {
+warn () {
     echo "$*"
 }
 
-die ( ) {
+die () {
     echo
     echo "$*"
     echo
@@ -155,7 +155,7 @@ if $cygwin ; then
 fi
 
 # Escape application args
-save ( ) {
+save () {
     for i do printf %s\\n "$i" | sed "s/'/'\\\\''/g;1s/^/'/;\$s/\$/' \\\\/" ; done
     echo " "
 }

networking/src/main/kotlin/ch/dissem/bitmessage/networking/nio/ConnectionIO.kt

@@ -61,7 +61,7 @@ class ConnectionIO(
         if (!headerOut.hasRemaining() && !sendingQueue.isEmpty()) {
             headerOut.clear()
             val payload = sendingQueue.poll()
-            payloadOut = NetworkMessage(payload).writeHeaderAndGetPayloadBuffer(headerOut)
+            payloadOut = NetworkMessage(payload).writer().writeHeaderAndGetPayloadBuffer(headerOut)
             headerOut.flip()
             lastUpdate = System.currentTimeMillis()
         }

networking/src/main/kotlin/ch/dissem/bitmessage/networking/nio/NioNetworkHandler.kt

@@ -90,7 +90,7 @@ class NioNetworkHandler : NetworkHandler, InternalContext.ContextHolder {
         SocketChannel.open(InetSocketAddress(server, port)).use { channel ->
             channel.configureBlocking(true)
             val headerBuffer = ByteBuffer.allocate(HEADER_SIZE)
-            val payloadBuffer = NetworkMessage(request).writeHeaderAndGetPayloadBuffer(headerBuffer)
+            val payloadBuffer = NetworkMessage(request).writer().writeHeaderAndGetPayloadBuffer(headerBuffer)
             headerBuffer.flip()
             while (headerBuffer.hasRemaining()) {
                 channel.write(headerBuffer)

networking/src/test/kotlin/ch/dissem/bitmessage/networking/TestNodeRegistry.kt

@@ -38,4 +38,16 @@ internal class TestNodeRegistry(vararg nodes: NetworkAddress) : NodeRegistry {
     override fun offerAddresses(nodes: List<NetworkAddress>) {
         // Ignore
     }
+
+    override fun update(node: NetworkAddress) {
+        // Ignore
+    }
+
+    override fun remove(node: NetworkAddress) {
+        // Ignore
+    }
+
+    override fun cleanup() {
+        // Ignore
+    }
 }

repositories/src/main/kotlin/ch/dissem/bitmessage/repository/JdbcAddressRepository.kt

@@ -175,7 +175,7 @@ class JdbcAddressRepository(config: JdbcConfig) : JdbcHelper(config), AddressRep
     private fun writePubkey(ps: PreparedStatement, parameterIndex: Int, data: Pubkey?) {
         if (data != null) {
             val out = ByteArrayOutputStream()
-            data.writeUnencrypted(out)
+            data.writer().writeUnencrypted(out)
             ps.setBytes(parameterIndex, out.toByteArray())
         } else {
             ps.setBytes(parameterIndex, null)

repositories/src/main/kotlin/ch/dissem/bitmessage/repository/JdbcHelper.kt

@@ -30,7 +30,7 @@ abstract class JdbcHelper protected constructor(@JvmField protected val config:
                 ps.setBytes(parameterIndex, null)
             } else {
                 val os = ByteArrayOutputStream()
-                data.write(os)
+                data.writer().write(os)
                 ps.setBytes(parameterIndex, os.toByteArray())
             }
         }

repositories/src/main/kotlin/ch/dissem/bitmessage/repository/JdbcNodeRegistry.kt

@@ -19,9 +19,8 @@ package ch.dissem.bitmessage.repository
 import ch.dissem.bitmessage.entity.valueobject.NetworkAddress
 import ch.dissem.bitmessage.ports.NodeRegistry
 import ch.dissem.bitmessage.ports.NodeRegistryHelper.loadStableNodes
+import ch.dissem.bitmessage.utils.*
 import ch.dissem.bitmessage.utils.Collections
-import ch.dissem.bitmessage.utils.SqlStrings
-import ch.dissem.bitmessage.utils.Strings
 import ch.dissem.bitmessage.utils.UnixTime.DAY
 import ch.dissem.bitmessage.utils.UnixTime.MINUTE
 import ch.dissem.bitmessage.utils.UnixTime.now
@@ -155,7 +154,18 @@ class JdbcNodeRegistry(config: JdbcConfig) : JdbcHelper(config), NodeRegistry {
                     ps.setBytes(2, node.IPv6)
                     ps.setInt(3, node.port)
                     ps.setLong(4, node.services)
-                    ps.setLong(5, node.time)
+                    ps.setLong(5,
+                        if (node.time > UnixTime.now) {
+                            // This might be an attack, let's not use those nodes with priority
+                            UnixTime.now - 7 * UnixTime.DAY
+                        } else if (node.time == 0L) {
+                            // Those just don't have a time set
+                            // let's give them slightly higher priority than the possible attack ones
+                            UnixTime.now - 6 * UnixTime.DAY
+                        } else {
+                            node.time
+                        }
+                    )
                     ps.executeUpdate()
                 }
             }
@@ -164,13 +174,20 @@ class JdbcNodeRegistry(config: JdbcConfig) : JdbcHelper(config), NodeRegistry {
         }
     }
 
-    private fun update(node: NetworkAddress) {
+    override fun update(node: NetworkAddress) {
         try {
+            val time = if (node.time > UnixTime.now) {
+                // This might be an attack, let's not use those nodes with priority
+                UnixTime.now - 7 * UnixTime.DAY
+            } else {
+                node.time
+            }
+
             config.getConnection().use { connection ->
                 connection.prepareStatement(
                     "UPDATE Node SET services=?, time=? WHERE stream=? AND address=? AND port=?").use { ps ->
                     ps.setLong(1, node.services)
-                    ps.setLong(2, node.time)
+                    ps.setLong(2, max(node.time, time))
                     ps.setLong(3, node.stream)
                     ps.setBytes(4, node.IPv6)
                     ps.setInt(5, node.port)
@@ -182,6 +199,36 @@ class JdbcNodeRegistry(config: JdbcConfig) : JdbcHelper(config), NodeRegistry {
         }
     }
 
+    override fun remove(node: NetworkAddress) {
+        try {
+            config.getConnection().use { connection ->
+                connection.prepareStatement(
+                    "DELETE FROM Node WHERE stream=? AND address=? AND port=?").use { ps ->
+                    ps.setLong(1, node.stream)
+                    ps.setBytes(2, node.IPv6)
+                    ps.setInt(3, node.port)
+                    ps.executeUpdate()
+                }
+            }
+        } catch (e: SQLException) {
+            LOG.error(e.message, e)
+        }
+    }
+
+    override fun cleanup() {
+        try {
+            config.getConnection().use { connection ->
+                connection.prepareStatement(
+                    "DELETE FROM Node WHERE time<?").use { ps ->
+                    ps.setLong(1, UnixTime.now - 8 * DAY)
+                    ps.executeUpdate()
+                }
+            }
+        } catch (e: SQLException) {
+            LOG.error(e.message, e)
+        }
+    }
+
     companion object {
         private val LOG = LoggerFactory.getLogger(JdbcNodeRegistry::class.java)
     }