heheHEADmaster

10 files changed, 374 insertions, 157 deletions

diff --git a/.gitignore b/.gitignore
index 567609b..d6b3d95 100644
--- a/.gitignore
+++ b/.gitignore
@@ -1 +1,4 @@
 build/
+.idea/
+.gradle/
+
diff --git a/build.gradle.kts b/build.gradle.kts
index 1cd365e..326bf4c 100644
--- a/build.gradle.kts
+++ b/build.gradle.kts
@@ -1,10 +1,11 @@
 plugins {
-    kotlin("jvm") version "1.5.31"
+    kotlin("jvm") version "1.7.10"
+    id("com.bnorm.power.kotlin-power-assert") version "0.12.0"
     application
 }
 
 group = "moe.nea89"
-version = "0.0.0"
+version = "0.0.1"
 
 repositories {
     mavenCentral()
@@ -16,5 +17,18 @@ application {
 
 dependencies {
     implementation(kotlin("stdlib"))
-    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.5.2")
+    implementation(platform("org.jetbrains.kotlinx:kotlinx-coroutines-bom:1.6.4"))
+    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core-jvm")
+    testImplementation(platform("io.kotest:kotest-bom:5.4.2"))
+    testImplementation("io.kotest:kotest-runner-junit5")
+    testImplementation("io.kotest:kotest-property")
+    testImplementation("io.kotest:kotest-framework-datatest")
+}
+
+kotlin {
+
+}
+
+tasks.withType<Test>() {
+    useJUnitPlatform()
 }
diff --git a/portforward.sh b/portforward.sh
new file mode 100755
index 0000000..2009574
--- /dev/null
+++ b/portforward.sh
@@ -0,0 +1 @@
+ssh -t -R 25:localhost:2500 nea89.moe "firewall-cmd --zone=public --add-port=25/tcp;/bin/bash"
diff --git a/samplemail.txt b/samplemail.txt
new file mode 100644
index 0000000..b75fce7
--- /dev/null
+++ b/samplemail.txt
@@ -0,0 +1,53 @@
+Received: by mail-lf1-x136.google.com with SMTP id k10so15448381lfm.4
+        for <test@nea.moe>; Thu, 08 Sep 2022 15:02:33 -0700 (PDT)
+DKIM-Signature: v=1; a=rsa-sha256; c=relaxed/relaxed;
+        d=gmail.com; s=20210112;
+        h=to:subject:message-id:date:from:mime-version:from:to:cc:subject
+         :date;
+        bh=OXwwfyS8bH5Cy8p4ETq5qLrpI/TRP0h4jRF5ayaHUlk=;
+        b=iXXYMc9MBQGhrGfRXJogalR4gUzRZhvKm6K48h4RJaHbOkgQONdwsSN/bk1GvPKev/
+         xwItDFSFAvq4mX6cpKKMGxp0DcgzCGJJojuiB5JvLTT2fqOQG/f4sWI66L+onTQS9tEu
+         HcbD7rW43SjyNh7NJGzxm2C8Lkh4ysHt/BMpltepJT/X1sysmf/dFVwk1VQO8Wx/7io2
+         mXIR64oINiSMUuBusCEKEyY2d8p/RBq1GmgCR3vfzd9u/LU/OCxqairGsgJgceyZ3VDG
+         qkyCJKUp4X2K2itKuFlOJgWi/rKKagDu18kQa7WFS9nZlh0nGu1Dse9XGHHb0PB2KIfI
+         Yw6A==
+MIME-Version: 1.0
+From: =?UTF-8?Q?Roman_Gr=C3=A4f?= <roman.graef@gmail.com>
+Date: Fri, 9 Sep 2022 00:02:21 +0200
+Message-ID: <CAAVxOtZbzPdkSQD2Qdo0XbXmGQc9-DmiFCvHnTXg6tR_ee5_oQ@mail.gmail.com>
+Subject: a
+To: test@nea.moe
+Content-Type: multipart/alternative; boundary="00000000000016ae1c05e8319674"
+
+--00000000000016ae1c05e8319674
+Content-Type: text/plain; charset="UTF-8"
+Content-Transfer-Encoding: quoted-printable
+
+a
+--=20
+*****************************
+Roman                    Gr=C3=A4f
+Mainzer Landstra=C3=9Fe        372
+60326       Frankfurt am Main
+Mob.            0157 35735001
+mailto: roman.graef@gmail.com
+
+--00000000000016ae1c05e8319674
+Content-Type: text/html; charset="UTF-8"
+Content-Transfer-Encoding: quoted-printable
+
+<div dir=3D"ltr"><div><br clear=3D"all"></div>a<br><div>-- <br><div dir=3D"=
+ltr" class=3D"gmail_signature" data-smartmail=3D"gmail_signature"><div dir=
+=3D"ltr"><div><div dir=3D"ltr"><div><div dir=3D"ltr"><div><div dir=3D"ltr">=
+<div><div><div dir=3D"ltr"><div><div dir=3D"ltr"><div><div dir=3D"ltr"><div=
+><span style=3D"font-family:monospace,monospace">**************************=
+***</span><br><span style=3D"font-family:monospace,monospace">Roman=C2=A0=
+=C2=A0=C2=A0=C2=A0=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0=C2=A0 =C2=A0=C2=
+=A0 Gr=C3=A4f<br>Mainzer Landstra=C3=9Fe=C2=A0=C2=A0=C2=A0=C2=A0=C2=A0=C2=
+=A0=C2=A0 372<br>60326 =C2=A0 =C2=A0 =C2=A0 Frankfurt am Main<br>Mob.=C2=A0=
+ =C2=A0 =C2=A0=C2=A0 =C2=A0 =C2=A0=C2=A0 0157 35735001<br>mailto: <a href=
+=3D"mailto:roman.graef@gmail.com" target=3D"_blank">roman.graef@gmail.com</=
+a></span></div></div></div></div></div></div></div></div></div></div></div>=
+</div></div></div></div></div></div></div>
+
+--00000000000016ae1c05e8319674--
diff --git a/src/main/kotlin/MailServer.kt b/src/main/kotlin/MailServer.kt
new file mode 100644
index 0000000..94a53c9
--- /dev/null
+++ b/src/main/kotlin/MailServer.kt
@@ -0,0 +1,30 @@
+import kotlinx.coroutines.*
+import java.net.ServerSocket
+import java.net.Socket
+import kotlin.coroutines.EmptyCoroutineContext
+
+class MailServer(
+    val localhostName: String,
+    val scope: CoroutineScope = CoroutineScope(EmptyCoroutineContext)
+) {
+
+    fun createAndLaunchHandlerFor(socket: Socket): Job {
+        val protocol = SMTPReceiveProtocol(localhostName, socket.inetAddress)
+        return protocol.executeAsync(scope + CoroutineName("connection handler from ${socket.inetAddress}"), Protocol.IO.FromSocket(socket))
+    }
+
+    suspend fun createServer(port: Int) {
+        listenToServerSocket(ServerSocket(port))
+    }
+
+    suspend fun listenToServerSocket(serverSocket: ServerSocket) {
+        withContext(Dispatchers.Unconfined) {
+            while (true) {
+                val newIncomingConnection =
+                    withContext(Dispatchers.IO) { serverSocket.accept() }
+                createAndLaunchHandlerFor(newIncomingConnection)
+            }
+        }
+    }
+
+}
diff --git a/src/main/kotlin/Main.kt b/src/main/kotlin/Main.kt
index 577c3fb..7f19f90 100644
--- a/src/main/kotlin/Main.kt
+++ b/src/main/kotlin/Main.kt
@@ -1,31 +1,21 @@
-import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.Job
 import kotlinx.coroutines.runBlocking
-import java.net.ServerSocket
+import kotlin.system.exitProcess
 
 object Main {
 
     @JvmStatic
     fun main(args: Array<String>) {
         if (args.size != 1) {
-            System.err.println("Use ./javamailteste run/install")
+            System.err.println("Use ./javamailteste run")
+            exitProcess(1)
         }
         when (args[0]) {
-            "run" -> runServer(2500)
+            "run" -> runServer(args.getOrElse(1) { "2500" }.toInt())
         }
     }
 
-    fun runServer(port: Int) = runBlocking(Dispatchers.Default) {
-        val ss = ServerSocket(port)
-        val jobs = mutableListOf<Job>()
-        println("Starting SMTP socket on port $port")
-        while (true) {
-            val scope = CoroutineScope(Dispatchers.Default)
-            val socket = with(Dispatchers.IO) { ss.accept() }
-            val prot = SMTPReceiveProtocol("nea89.moe", socket.inetAddress)
-            jobs.add(prot.executeAsync(scope, Protocol.IO.FromSocket(socket)))
-            println("jobs: $jobs")
-        }
+    fun runServer(port: Int) {
+        val mailServer = MailServer("nea89.moe")
+        runBlocking { mailServer.createServer(port) }
     }
 }
diff --git a/src/main/kotlin/Protocol.kt b/src/main/kotlin/Protocol.kt
new file mode 100644
index 0000000..019e172
--- /dev/null
+++ b/src/main/kotlin/Protocol.kt
@@ -0,0 +1,157 @@
+import kotlinx.coroutines.*
+import java.io.InputStream
+import java.io.OutputStream
+import java.net.Socket
+
+class Invalidatable {
+    var isInvalid: Boolean = false
+    fun checkValid() {
+        if (isInvalid)
+            throw IllegalStateException("Accessed invalid object")
+    }
+
+    fun invalidate() {
+        isInvalid = true
+    }
+}
+
+abstract class Protocol {
+    interface IO {
+        fun isOpen(): Boolean
+        suspend fun pushBack(data: ByteArray)
+        suspend fun readBytes(into: ByteArray): Int
+        suspend fun send(bytes: ByteArray)
+        suspend fun close()
+
+        class FromSocket(val socket: Socket) : FromStreams(socket.getInputStream(), socket.getOutputStream()) {
+            override suspend fun close() {
+                super.close()
+                withContext(Dispatchers.IO) {
+                    socket.close()
+                }
+            }
+        }
+
+        open class FromStreams(val inputStream: InputStream, val outputStream: OutputStream) : IO {
+            private val i = Invalidatable()
+            override fun isOpen(): Boolean =
+                !i.isInvalid
+
+
+            val readBuffer = mutableListOf<ByteArray>()
+            override suspend fun pushBack(data: ByteArray) {
+                i.checkValid()
+                if (data.isEmpty()) return
+                readBuffer.add(0, data)
+            }
+
+            override suspend fun send(bytes: ByteArray) {
+                i.checkValid()
+                withContext(Dispatchers.IO) {
+                    outputStream.write(bytes)
+                    outputStream.flush()
+                }
+            }
+
+            override suspend fun close() {
+                i.checkValid()
+                i.invalidate()
+                withContext(Dispatchers.IO) {
+                    inputStream.close()
+                    outputStream.close()
+                }
+            }
+
+            override suspend fun readBytes(into: ByteArray): Int {
+                i.checkValid()
+                val rb = readBuffer.removeFirstOrNull()
+                if (rb != null) {
+                    val w = minOf(rb.size, into.size)
+                    rb.copyInto(into, 0, 0, w)
+                    return w
+                }
+                return withContext(Dispatchers.IO) {
+                    inputStream.read(into)
+                }
+            }
+        }
+    }
+
+    protected abstract suspend fun IO.execute()
+
+    fun executeAsync(scope: CoroutineScope, io: Protocol.IO): Job {
+        return scope.launch {
+            io.execute()
+        }
+    }
+}
+
+suspend fun Protocol.IO.readAll(): ByteArray {
+    var ret = ByteArray(0)
+    val buffer = ByteArray(4096)
+    while (true) {
+        val read = readBytes(buffer)
+        if (read == -1) {
+            return ret
+        }
+        val oldSize = ret.size
+        ret = ret.copyOf(oldSize + read)
+        buffer.copyInto(ret, oldSize, endIndex = read)
+    }
+}
+
+suspend fun Protocol.IO.send(string: String) = send(string.encodeToByteArray())
+suspend fun Protocol.IO.readLine(): String = readUntil(CRLF).decodeToString()
+suspend fun Protocol.IO.readUntil(search: ByteArray, errorOnEOF: Boolean = true): ByteArray {
+    var ret = ByteArray(0)
+    val buffer = ByteArray(4096)
+    while (true) {
+        val read = readBytes(buffer)
+        if (read == -1) {
+            if (errorOnEOF) {
+                throw IllegalStateException("End of Protocol.IO")
+            } else {
+                return ret
+            }
+        }
+        val oldSize = ret.size
+        ret = ret.copyOf(oldSize + read)
+        buffer.copyInto(ret, oldSize, endIndex = read)
+        val firstFoundIndex = ret.findSubarray(search, startIndex = (oldSize - search.size - 1).coerceAtLeast(0))
+        if (firstFoundIndex != null) {
+            pushBack(ret.copyOfRange(firstFoundIndex + search.size, ret.size))
+            return ret.copyOf(firstFoundIndex)
+        }
+    }
+}
+
+val CRLF = "\r\n".encodeToByteArray()
+
+fun ByteArray.findSubarray(subarray: ByteArray, startIndex: Int = 0): Int? {
+    if (subarray.size > size - startIndex) return null
+    for (i in startIndex..size - subarray.size) {
+        var isEqual = true
+        for (j in subarray.indices) {
+            if (this[i + j] != subarray[j]) {
+                isEqual = false
+                break
+            }
+        }
+        if (isEqual) {
+            return i
+        }
+    }
+    return null
+}
+
+suspend fun Protocol.IO.pushBack(string: String) = pushBack(string.encodeToByteArray())
+suspend fun Protocol.IO.lookahead(string: String): Boolean = lookahead(string.encodeToByteArray())
+suspend fun Protocol.IO.lookahead(bytes: ByteArray): Boolean {
+    val buffer = ByteArray(bytes.size)
+    val read = readBytes(buffer)
+    if (read != bytes.size || !buffer.contentEquals(bytes)) {
+        pushBack(buffer.copyOf(read))
+        return false
+    }
+    return true
+}
diff --git a/src/main/kotlin/RFC822Parser.kt b/src/main/kotlin/RFC822Parser.kt
new file mode 100644
index 0000000..4a6ce74
--- /dev/null
+++ b/src/main/kotlin/RFC822Parser.kt
@@ -0,0 +1,58 @@
+import kotlinx.coroutines.MainScope
+import java.io.ByteArrayOutputStream
+import java.io.File
+
+suspend fun main() {
+    val io = Protocol.IO.FromStreams(File("samplemail.txt").inputStream(), ByteArrayOutputStream())
+    val rfc = RFC822Parser()
+    rfc.executeAsync(MainScope(), io).join()
+    println(rfc.headers)
+    println("Content: ${rfc.content.decodeToString()}")
+}
+
+class RFC822Parser() : Protocol() {
+
+    class Header(val field: String, val value: ByteArray)
+
+    private val _headers = mutableListOf<Header>()
+    val headers: List<Header> get() = _headers
+    lateinit var content: ByteArray
+        private set
+
+    override suspend fun IO.execute() {
+        while (parseField()) Unit
+        content = readAll()
+    }
+
+    private suspend fun IO.parseField(): Boolean {
+        val read = readUntil(CRLF)
+        if (read.contentEquals(CRLF)) {
+            return false
+        }
+        val indexOfColon = read.indexOf(':'.code.toByte())
+        if (indexOfColon == -1) {
+            throw IllegalStateException("Expected : in MIME header")
+        }
+        val headerField = read.sliceArray(0 until indexOfColon).decodeToString().trim()
+        var data = read.sliceArray(indexOfColon + 1 until read.size)
+        while (true) {
+            val nextLine = readUntil(CRLF)
+            if (nextLine.isNotEmpty() && isWhitespaceCharacter(nextLine[0])) {
+                val oldSize = data.size
+                data = data.copyOf(oldSize + nextLine.size)
+                nextLine.copyInto(data, oldSize)
+            } else {
+                pushBack(CRLF)
+                pushBack(nextLine)
+                break
+            }
+        }
+        _headers.add(Header(headerField, data))
+        return true
+    }
+
+    fun isWhitespaceCharacter(char: Byte): Boolean {
+        val char = char.toInt().toChar()
+        return char == ' ' || char == '\t'
+    }
+}
diff --git a/src/main/kotlin/SMTPProtocol.kt b/src/main/kotlin/SMTPProtocol.kt
index 6f01bb2..e5da601 100644
--- a/src/main/kotlin/SMTPProtocol.kt
+++ b/src/main/kotlin/SMTPProtocol.kt
@@ -1,133 +1,8 @@
-import kotlinx.coroutines.CoroutineScope
-import kotlinx.coroutines.Dispatchers
-import kotlinx.coroutines.Job
-import kotlinx.coroutines.launch
-import java.io.InputStream
-import java.io.OutputStream
 import java.net.InetAddress
-import java.net.Socket
 import kotlin.contracts.ExperimentalContracts
 import kotlin.contracts.InvocationKind
 import kotlin.contracts.contract
 
-class Invalidatable {
-    var isInvalid: Boolean = false
-    fun checkValid() {
-        if (isInvalid)
-            throw IllegalStateException("Accessed invalid object")
-    }
-
-    fun invalidate() {
-        isInvalid = true
-    }
-}
-
-abstract class Protocol {
-    interface IO {
-        fun isOpen(): Boolean
-        suspend fun pushBack(data: ByteArray)
-        suspend fun readBytes(into: ByteArray): Int
-        suspend fun send(bytes: ByteArray)
-        suspend fun close()
-        class FromSocket(val socket: Socket) : FromStreams(socket.getInputStream(), socket.getOutputStream()) {
-            override suspend fun close() {
-                super.close()
-                with(Dispatchers.IO) {
-                    socket.close()
-                }
-            }
-        }
-
-        open class FromStreams(val inputStream: InputStream, val outputStream: OutputStream) : IO {
-            private val i = Invalidatable()
-            override fun isOpen(): Boolean =
-                !i.isInvalid
-
-
-            val readBuffer = mutableListOf<ByteArray>()
-            override suspend fun pushBack(data: ByteArray) {
-                i.checkValid()
-                if (data.isEmpty()) return
-                readBuffer.add(0, data)
-            }
-
-            override suspend fun send(bytes: ByteArray) {
-                i.checkValid()
-                with(Dispatchers.IO) {
-                    outputStream.write(bytes)
-                    outputStream.flush()
-                }
-            }
-
-            override suspend fun close() {
-                i.checkValid()
-                i.invalidate()
-                with(Dispatchers.IO) {
-                    inputStream.close()
-                    outputStream.close()
-                }
-            }
-
-            override suspend fun readBytes(into: ByteArray): Int {
-                i.checkValid()
-                val rb = readBuffer.removeFirstOrNull()
-                if (rb != null) {
-                    val w = minOf(rb.size, into.size)
-                    rb.copyInto(into, 0, 0, w)
-                    return w
-                }
-                return with(Dispatchers.IO) {
-                    inputStream.read(into)
-                }
-            }
-        }
-    }
-
-    protected abstract suspend fun IO.execute()
-
-    fun executeAsync(scope: CoroutineScope, io: Protocol.IO): Job {
-        return scope.launch {
-            io.execute()
-        }
-    }
-}
-
-suspend fun Protocol.IO.send(string: String) = send(string.encodeToByteArray())
-suspend fun Protocol.IO.readLine(): String {
-    val y = mutableListOf<String>()
-    while (true) {
-        val buffer = ByteArray(4096)
-        val read = readBytes(buffer)
-        val i = buffer.findCRLF()
-        if (i in 0 until read) {
-            y.add(buffer.copyOfRange(0, i).decodeToString())
-            pushBack(buffer.copyOfRange(i + 2, read))
-            break
-        } else {
-            y.add(buffer.copyOfRange(0, read).decodeToString())
-        }
-    }
-    return y.joinToString("")
-}
-
-private fun ByteArray.findCRLF(): Int {
-    return this.asSequence().zipWithNext().withIndex().firstOrNull { (idx, v) ->
-        (v.first == '\r'.code.toByte()) and (v.second == '\n'.code.toByte())
-    }?.index ?: -1
-}
-
-suspend fun Protocol.IO.pushBack(string: String) = pushBack(string.encodeToByteArray())
-suspend fun Protocol.IO.lookahead(string: String): Boolean = lookahead(string.encodeToByteArray())
-suspend fun Protocol.IO.lookahead(bytes: ByteArray): Boolean {
-    val buffer = ByteArray(bytes.size)
-    val read = readBytes(buffer)
-    if (read != bytes.size || !buffer.contentEquals(bytes)) {
-        pushBack(buffer.copyOf(read))
-        return false
-    }
-    return true
-}
-
 
 @OptIn(ExperimentalContracts::class)
 class SMTPReceiveProtocol(val localHost: String, val inetAddress: InetAddress) : Protocol() {
@@ -136,20 +11,20 @@ class SMTPReceiveProtocol(val localHost: String, val inetAddress: InetAddress) :
         var matched = false
 
 
-        suspend inline fun command(vararg name: String, block: IO.(String) -> Unit) {
+        suspend inline fun command(vararg name: String, block: suspend IO.(String) -> Unit) {
             contract { callsInPlace(block, InvocationKind.AT_MOST_ONCE) }
             for (n in name) commandOnce(n, block)
         }
 
-        suspend inline fun commandOnce(name: String, block: IO.(String) -> Unit) {
+        suspend inline fun commandOnce(name: String, block: suspend IO.(String) -> Unit) {
             contract { callsInPlace(block, InvocationKind.AT_MOST_ONCE) }
             if (matched) return
-            if (!line.startsWith(name)) return
+            if (!line.startsWith(name, ignoreCase = true)) return
             matched = true
             block(line.substring(name.length).trimStart())
         }
 
-        suspend inline fun otherwise(block: IO.(String) -> Unit) {
+        suspend inline fun otherwise(block: suspend IO.(String) -> Unit) {
             contract { callsInPlace(block, InvocationKind.AT_MOST_ONCE) }
             if (matched) return
             matched = true
@@ -157,7 +32,7 @@ class SMTPReceiveProtocol(val localHost: String, val inetAddress: InetAddress) :
         }
     }
 
-    suspend inline fun IO.commands(line: String, block: Commands.() -> Unit) {
+    suspend inline fun IO.commands(line: String, block: suspend Commands.() -> Unit) {
         contract { callsInPlace(block, InvocationKind.EXACTLY_ONCE) }
         Commands(line, this).block()
     }
@@ -202,13 +77,8 @@ class SMTPReceiveProtocol(val localHost: String, val inetAddress: InetAddress) :
                 }
                 command("DATA") {
                     send("354 Enter mail, end with \".\" on a line by itself\r\n")
-                    var text = ""
-                    while (true) {
-                        val tmp = readLine()
-                        if (tmp == ".") break
-                        text += tmp + "\n"
-                    }
-                    messages.add(Mail(trans.sender!!, trans.recipients.toList(), text))
+                    var text = readUntil("\r\n.\r\n".encodeToByteArray())
+                    messages.add(Mail(trans.sender!!, trans.recipients.toList(), text.decodeToString()))
                     trans.reset()
                     send("250 Message accepted for delivery\r\n")
                 }
diff --git a/src/test/kotlin/moe/nea89/mail/util/ProtocolSpec.kt b/src/test/kotlin/moe/nea89/mail/util/ProtocolSpec.kt
new file mode 100644
index 0000000..9a4bcbf
--- /dev/null
+++ b/src/test/kotlin/moe/nea89/mail/util/ProtocolSpec.kt
@@ -0,0 +1,41 @@
+package moe.nea89.mail.util
+
+import Protocol
+import findSubarray
+import io.kotest.core.spec.style.FreeSpec
+import io.kotest.datatest.withData
+import readUntil
+import java.io.ByteArrayInputStream
+import java.io.ByteArrayOutputStream
+
+
+data class ByteArrayString(val hay: String, val needle: String, val position: Int?, val offset: Int = 0)
+class ProtocolSpec : FreeSpec({
+    "ByteArray.findSubarray" - {
+        "should work correctly" - {
+            withData(
+                listOf(
+                    ByteArrayString("abc", "a", 0),
+                    ByteArrayString("abca", "a", 0),
+                    ByteArrayString("abca", "a", 3, 2),
+                    ByteArrayString("abca", "a", 3, 3),
+                    ByteArrayString("bc", "a", null),
+                    ByteArrayString("acbcab", "ab", 4),
+                    ByteArrayString("abcbcab", "ab", 5, 1),
+                )
+            ) { (hay, needle, position, offset) ->
+                val hay = hay.encodeToByteArray()
+                val needle = needle.encodeToByteArray()
+                assert(hay.findSubarray(needle, offset) == position)
+            }
+        }
+    }
+    "Protocol.IO" - {
+        "readUntil" {
+            val data = ("a".repeat(9) + "01" + "b".repeat(10) + "02").encodeToByteArray()
+            val protIO = Protocol.IO.FromStreams(ByteArrayInputStream(data), ByteArrayOutputStream())
+            assert(protIO.readUntil("01".encodeToByteArray()).decodeToString() == "a".repeat(9))
+            assert(protIO.readUntil("02".encodeToByteArray()).decodeToString() == "b".repeat(10))
+        }
+    }
+})