feat: add java.io framing streams and a coroutines Flow API

Zero-dep java.io adapters; Flow via compileOnly coroutines (no runtime dep).

Author: ajsb85

README.md

@@ -34,6 +34,14 @@ as a serial/UART, USB, or BLE link.
   `unframe`, and the incremental `CobsStreamDecoder` (reassembles packets across
   arbitrary chunk boundaries, with a `maxFrameLength` guard). Each takes a
   `sentinel` byte (default `0x00`) so frames can be delimited by any chosen byte.
+- **`java.io` stream adapters** — `CobsFramedOutputStream.writeFrame` /
+  `CobsFramedInputStream.readFrame` (plus `frames()` as a `Sequence`) wrap any
+  `OutputStream` / `InputStream` to write and read self-delimiting frames. They
+  use only `java.io`, so they add no dependency.
+- **Coroutines `Flow` hook** — `Flow<ByteArray>.cobsFrames()` reassembles a flow
+  of raw chunks into a flow of decoded packets. `kotlinx-coroutines` is a
+  `compileOnly` dependency, so the published artifact stays free of any runtime
+  dependency; the extension is available to consumers who already use coroutines.
 - **Zero dependencies**, pure Kotlin, no Android framework APIs in the logic.
   `minSdk 21`, `compileSdk 35`.
 
@@ -122,6 +130,39 @@ CobsFraming.unframe(framed, sentinel = s)                                   // [
 val rxAA = CobsStreamDecoder(maxFrameLength = 4096, sentinel = s)
 ```
 
+### `java.io` stream adapters
+
+Wrap any `OutputStream` / `InputStream` to write and read self-delimiting frames
+(dependency-free — `java.io` only). `readFrame()` returns `null` at end of stream;
+`frames()` exposes the same reads as a `Sequence`.
+
+```kotlin
+import dev.firechip.cobs.CobsFramedOutputStream
+import dev.firechip.cobs.CobsFramedInputStream
+
+CobsFramedOutputStream(socket.outputStream).use { out ->
+    out.writeFrame(byteArrayOf(0x11, 0x00, 0x22)) // encoded frame + 0x00 delimiter
+}
+
+val input = CobsFramedInputStream(socket.inputStream) // reduced / sentinel optional
+for (packet in input.frames()) handlePacket(packet)
+```
+
+### Coroutines `Flow`
+
+`Flow<ByteArray>.cobsFrames()` reassembles a flow of raw chunks (however
+misaligned) into a flow of decoded packets. `kotlinx-coroutines` is a
+`compileOnly` dependency, so it adds nothing to the published artifact; add it to
+your own build to use this extension.
+
+```kotlin
+import dev.firechip.cobs.cobsFrames
+
+serialBytes // Flow<ByteArray> of raw reads
+    .cobsFrames() // reduced / skipEmpty / sentinel optional
+    .collect { packet -> handlePacket(packet) }
+```
+
 Invalid encoded input throws `CobsDecodeException`.
 
 ## Build

cobs/build.gradle.kts

@@ -58,7 +58,14 @@ kotlin {
 }
 
 dependencies {
+    // Coroutines back the Flow.cobsFrames extension only. It is compileOnly so the
+    // published AAR/POM gains NO runtime dependency: consumers who use the Flow API
+    // already have kotlinx-coroutines on their classpath, and those who don't pay
+    // nothing. The test source set needs it (and coroutines-test) at runtime.
+    compileOnly("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.9.0")
     testImplementation("junit:junit:4.13.2")
+    testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.9.0")
+    testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-test:1.9.0")
 }
 
 publishing {

cobs/src/main/kotlin/dev/firechip/cobs/Flow.kt

@@ -0,0 +1,47 @@
+// Copyright (c) 2026 Alexander Salas Bastidas <ajsb85@firechip.dev>
+// SPDX-License-Identifier: MIT
+
+package dev.firechip.cobs
+
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.flow
+
+/**
+ * Reassembles a [Flow] of raw byte chunks — as delivered by a serial/UART, USB,
+ * or BLE source whose reads do not align with frame boundaries — into a [Flow]
+ * of decoded packets.
+ *
+ * Each collection drives a fresh [CobsStreamDecoder]: upstream chunks are fed to
+ * it in order and every packet it completes is emitted downstream, so a frame
+ * split across several chunks (or several frames in one chunk) is handled
+ * transparently. The returned flow is cold and re-collectable.
+ *
+ * This mirrors [CobsFramedInputStream] for coroutine-based sources. Coroutines
+ * is a `compileOnly` dependency of this library, so this extension is available
+ * only to consumers who already depend on `kotlinx-coroutines-core`.
+ *
+ * @param reduced decode with COBS/R (reduced) instead of basic COBS.
+ * @param skipEmpty drop empty frames instead of emitting empty arrays.
+ * @param sentinel the delimiter byte (and the byte each frame avoids); defaults
+ *   to the `0x00` [COBS_DELIMITER] and must match the sender's sentinel.
+ * @throws CobsDecodeException if a completed frame is not valid encoded data.
+ */
+public fun Flow<ByteArray>.cobsFrames(
+    reduced: Boolean = false,
+    skipEmpty: Boolean = true,
+    sentinel: Byte = COBS_DELIMITER,
+): Flow<ByteArray> {
+    val chunks = this
+    return flow {
+        val decoder = CobsStreamDecoder(
+            reduced = reduced,
+            skipEmpty = skipEmpty,
+            sentinel = sentinel,
+        )
+        chunks.collect { chunk ->
+            for (frame in decoder.feed(chunk)) {
+                emit(frame)
+            }
+        }
+    }
+}

cobs/src/main/kotlin/dev/firechip/cobs/Streams.kt

@@ -0,0 +1,133 @@
+// Copyright (c) 2026 Alexander Salas Bastidas <ajsb85@firechip.dev>
+// SPDX-License-Identifier: MIT
+
+package dev.firechip.cobs
+
+import java.io.ByteArrayOutputStream
+import java.io.FilterInputStream
+import java.io.FilterOutputStream
+import java.io.IOException
+import java.io.InputStream
+import java.io.OutputStream
+
+/**
+ * An [OutputStream] wrapper that writes whole packets as self-delimiting COBS
+ * frames, for pushing packets onto a byte stream such as a serial/UART link.
+ *
+ * Each [writeFrame] call COBS-encodes a packet (COBS/R when [reduced] is true,
+ * otherwise basic COBS) and writes the encoded bytes followed by the [sentinel]
+ * delimiter. Because the encoding never emits the sentinel, the delimiter
+ * unambiguously ends the frame. This is the streaming counterpart of
+ * [CobsFraming.frame].
+ *
+ * The class extends [FilterOutputStream], so [flush] and [close] delegate to the
+ * wrapped [out] stream; closing this stream closes the underlying one.
+ *
+ * @param out the underlying stream that framed bytes are written to.
+ * @param reduced use COBS/R (reduced) encoding instead of basic COBS.
+ * @param sentinel the delimiter byte (and the byte the encoding avoids); defaults
+ *   to the `0x00` [COBS_DELIMITER] and must match the reader's sentinel.
+ */
+public class CobsFramedOutputStream @JvmOverloads constructor(
+    out: OutputStream,
+    private val reduced: Boolean = false,
+    private val sentinel: Byte = COBS_DELIMITER,
+) : FilterOutputStream(out) {
+
+    /**
+     * Encodes [packet] and writes the resulting frame — the encoded bytes plus a
+     * trailing [sentinel] delimiter — to the underlying stream.
+     *
+     * The bytes are written straight to the wrapped stream; call [flush] to push
+     * them through a buffered stream. An empty [packet] still produces a
+     * non-empty frame (a single length byte plus the delimiter).
+     *
+     * @throws IOException if the underlying stream fails to accept the bytes.
+     */
+    @Throws(IOException::class)
+    public fun writeFrame(packet: ByteArray) {
+        val encoded = if (reduced) {
+            Cobsr.encodeWithSentinel(packet, sentinel)
+        } else {
+            Cobs.encodeWithSentinel(packet, sentinel)
+        }
+        out.write(encoded)
+        out.write(sentinel.toInt())
+    }
+}
+
+/**
+ * An [InputStream] wrapper that reads self-delimiting COBS frames back into
+ * packets, the reading counterpart of [CobsFramedOutputStream].
+ *
+ * [readFrame] consumes bytes up to the next [sentinel] delimiter (or end of
+ * stream), decodes them (COBS/R when [reduced] is true, otherwise basic COBS)
+ * and returns the recovered packet, or `null` once the stream is exhausted.
+ * Empty frames — produced by a leading or consecutive delimiter — are skipped
+ * when [skipEmpty] is true. [frames] exposes the same reads as a [Sequence].
+ *
+ * Bytes are pulled one at a time from the source, so wrap an unbuffered source
+ * (a socket or file) in a [java.io.BufferedInputStream] for throughput. The
+ * class extends [FilterInputStream], so [close] delegates to the wrapped
+ * [source].
+ *
+ * @param source the underlying stream framed bytes are read from.
+ * @param reduced decode with COBS/R (reduced) instead of basic COBS.
+ * @param skipEmpty drop empty frames instead of returning empty arrays.
+ * @param sentinel the delimiter byte (and the byte each frame avoids); defaults
+ *   to the `0x00` [COBS_DELIMITER] and must match the writer's sentinel.
+ */
+public class CobsFramedInputStream @JvmOverloads constructor(
+    source: InputStream,
+    private val reduced: Boolean = false,
+    private val skipEmpty: Boolean = true,
+    private val sentinel: Byte = COBS_DELIMITER,
+) : FilterInputStream(source) {
+
+    /**
+     * Reads and decodes the next frame, returning the recovered packet, or `null`
+     * at end of stream.
+     *
+     * Bytes are read until the [sentinel] delimiter or end of stream; the bytes
+     * before it form the frame. A frame that ends at end of stream without a
+     * delimiter (a truncated tail) is still decoded and returned. Empty frames
+     * are skipped while [skipEmpty] is true, otherwise each yields an empty array.
+     *
+     * @throws IOException if the underlying stream fails.
+     * @throws CobsDecodeException if a complete frame is not valid encoded data.
+     */
+    @Throws(IOException::class)
+    public fun readFrame(): ByteArray? {
+        while (true) {
+            val first = `in`.read()
+            if (first == -1) return null
+            if (first.toByte() == sentinel) {
+                // Empty frame from a leading or consecutive delimiter.
+                if (skipEmpty) continue
+                return ByteArray(0)
+            }
+            val buffer = ByteArrayOutputStream()
+            buffer.write(first)
+            while (true) {
+                val b = `in`.read()
+                if (b == -1 || b.toByte() == sentinel) break
+                buffer.write(b)
+            }
+            return decodeFrame(buffer.toByteArray())
+        }
+    }
+
+    /**
+     * Returns a [Sequence] that yields each frame from [readFrame] until the
+     * stream is exhausted. Iterating the sequence drives the same underlying
+     * reads, so it is single-pass.
+     */
+    public fun frames(): Sequence<ByteArray> = generateSequence { readFrame() }
+
+    private fun decodeFrame(frame: ByteArray): ByteArray =
+        if (reduced) {
+            Cobsr.decodeWithSentinel(frame, sentinel)
+        } else {
+            Cobs.decodeWithSentinel(frame, sentinel)
+        }
+}

cobs/src/test/kotlin/dev/firechip/cobs/FlowTest.kt

@@ -0,0 +1,65 @@
+// Copyright (c) 2026 Alexander Salas Bastidas <ajsb85@firechip.dev>
+// SPDX-License-Identifier: MIT
+
+package dev.firechip.cobs
+
+import kotlinx.coroutines.flow.asFlow
+import kotlinx.coroutines.flow.flowOf
+import kotlinx.coroutines.flow.toList
+import kotlinx.coroutines.runBlocking
+import kotlinx.coroutines.test.runTest
+import org.junit.Assert.assertArrayEquals
+import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
+import org.junit.Test
+
+/** Reassembles flows of raw chunks into decoded frames via [cobsFrames]. */
+class FlowTest {
+
+    @Test
+    fun reassemblesMisalignedChunks() = runTest {
+        val wire = cat(
+            CobsFraming.frame(bytes(0x01, 0x02)),
+            CobsFraming.frame(bytes(0x00, 0xFF)),
+            CobsFraming.frame(bytes(0x2A)),
+        )
+        // Cut the wire at boundaries that fall in the middle of frames.
+        val cuts = intArrayOf(0, 1, 6, 7, wire.size)
+        val chunks = (0 until cuts.size - 1).map { wire.copyOfRange(cuts[it], cuts[it + 1]) }
+
+        val frames = flowOf(*chunks.toTypedArray()).cobsFrames().toList()
+        assertEquals(3, frames.size)
+        assertArrayEquals(bytes(0x01, 0x02), frames[0])
+        assertArrayEquals(bytes(0x00, 0xFF), frames[1])
+        assertArrayEquals(bytes(0x2A), frames[2])
+    }
+
+    @Test
+    fun reducedAndSentinelByteAtATime() = runTest {
+        val s = 0xAA.toByte()
+        val packets = listOf(ascii("12345"), bytes(0x11, 0x00, 0x22), bytes(0x2A))
+        val wire = cat(
+            *Array(packets.size) {
+                CobsFraming.frame(packets[it], reduced = true, sentinel = s)
+            },
+        )
+        // One byte per chunk is the most misaligned drip possible.
+        val chunks = wire.map { byteArrayOf(it) }
+
+        val frames = chunks.asFlow().cobsFrames(reduced = true, sentinel = s).toList()
+        assertEquals(packets.size, frames.size)
+        for (k in packets.indices) assertArrayEquals(packets[k], frames[k])
+    }
+
+    @Test
+    fun skipEmptyFalseKeepsEmptyFrames() = runBlocking {
+        val wire = cat(bytes(0x00), CobsFraming.frame(bytes(0x42)), bytes(0x00, 0x00))
+
+        val frames = flowOf(wire).cobsFrames(skipEmpty = false).toList()
+        assertEquals(4, frames.size)
+        assertTrue(frames[0].isEmpty())
+        assertArrayEquals(bytes(0x42), frames[1])
+        assertTrue(frames[2].isEmpty())
+        assertTrue(frames[3].isEmpty())
+    }
+}