Add RawSource.asFlow transformer

- Introduced converter from `RawSource` to Kotlin Flows.
- Added `StreamingDecoder` interface and its implementation `DelimitingByteStreamDecoder` for processing byte streams with a specified delimiter.
- Added tests for decoder, and flow behavior cases.

Author: kpavlov

core/build.gradle.kts

@@ -48,9 +48,10 @@ kotlin {
     sourceSets {
         commonMain.dependencies {
             api(project(":kotlinx-io-bytestring"))
+            api(libs.kotlinx.coroutines.core)
         }
-        appleTest.dependencies {
-            implementation(libs.kotlinx.coroutines.core)
+        commonTest.dependencies {
+            implementation(libs.kotlinx.coroutines.test)
         }
     }
 }

core/common/src/coroutines/DelimitingByteStreamDecoder.kt

@@ -0,0 +1,68 @@
+/*
+ * Copyright 2017-2025 JetBrains s.r.o. and respective authors and developers.
+ * Use of this source code is governed by the Apache 2.0 license that can be found in the LICENCE file.
+ */
+package kotlinx.io.coroutines
+
+import kotlinx.io.Buffer
+import kotlinx.io.readByteArray
+
+/**
+ * A streaming decoder that reads a continuous stream of bytes and separates it into discrete
+ * chunks based on a specified delimiter. The default delimiter is the newline character (`'\n'`).
+ *
+ * This class buffers incoming byte arrays and emits individual byte arrays once a delimiter is
+ * encountered. Any remaining bytes in the buffer are emitted when [onClose] is called.
+ *
+ * ## Example
+ *
+ * ```kotlin
+ * val decoder = DelimitingByteStreamDecoder()
+ * val source: RawSource = // ...
+ * source.asFlow(decoder).collect { line ->
+ *     println("Received: ${line.decodeToString()}")
+ * }
+ * ```
+ *
+ * ## Thread Safety
+ *
+ * This class is **not thread-safe**. Each instance maintains internal mutable state and must
+ * not be shared across multiple flows or concurrent coroutines.
+ *
+ * ## Lifecycle
+ *
+ * After [onClose] is called, this decoder **cannot be reused**. The internal buffer is closed
+ * and the decoder should be discarded.
+ *
+ * @property delimiter The byte value used as a delimiter to separate the stream into chunks.
+ *                      Defaults to the newline character (`'\n'`).
+ */
+public class DelimitingByteStreamDecoder(
+    public val delimiter: Byte = '\n'.code.toByte(),
+) : StreamingDecoder<ByteArray> {
+
+    private val buffer = Buffer()
+
+    override suspend fun decode(bytes: ByteArray, byteConsumer: suspend (ByteArray) -> Unit) {
+        var startIndex = 0
+        for (i in bytes.indices) {
+            if (bytes[i] == delimiter) {
+                buffer.write(bytes, startIndex, i)
+                // flush and clear buffer
+                byteConsumer.invoke(buffer.readByteArray())
+                startIndex = i + 1
+            }
+        }
+        // Buffer any remaining bytes after the last delimiter
+        if (startIndex < bytes.size) {
+            buffer.write(bytes, startIndex, bytes.size)
+        }
+    }
+
+    override suspend fun onClose(byteConsumer: suspend (ByteArray) -> Unit) {
+        if (buffer.size > 0) {
+            byteConsumer.invoke(buffer.readByteArray())
+        }
+        buffer.close()
+    }
+}

core/common/src/coroutines/SourceFlow.kt

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2017-2025 JetBrains s.r.o. and respective authors and developers.
+ * Use of this source code is governed by the Apache 2.0 license that can be found in the LICENCE file.
+ */
+package kotlinx.io.coroutines
+
+import kotlinx.coroutines.NonCancellable
+import kotlinx.coroutines.flow.Flow
+import kotlinx.coroutines.flow.channelFlow
+import kotlinx.coroutines.isActive
+import kotlinx.coroutines.withContext
+import kotlinx.coroutines.yield
+import kotlinx.io.Buffer
+import kotlinx.io.IOException
+import kotlinx.io.RawSource
+import kotlinx.io.readByteArray
+
+public const val READ_BUFFER_SIZE: Long = 8196
+
+/**
+ * Converts this [RawSource] into a Kotlin [Flow], emitting decoded data using the provided [StreamingDecoder].
+ *
+ * This function reads data from the source in chunks, decodes it using the provided decoder, and emits
+ * the decoded elements downstream. The returned flow is cold and will start reading from the source
+ * when collected.
+ *
+ * ## Lifecycle and Resource Management
+ *
+ * - The source is automatically closed when the flow completes, fails, or is cancelled
+ * - The decoder's [StreamingDecoder.onClose] is always called for cleanup, even on cancellation
+ * - On normal completion or [IOException], any remaining buffered data in the decoder is emitted
+ * - On cancellation, the decoder is cleaned up but remaining data is discarded
+ *
+ * ## Backpressure
+ *
+ * The flow respects structured concurrency and backpressure. Reading from the source is suspended
+ * when the downstream collector cannot keep up.
+ *
+ * @param T The type of elements emitted by the Flow after decoding.
+ * @param decoder The [StreamingDecoder] used to decode data read from this source.
+ * @param readBufferSize The size of the buffer used for reading from the source. Defaults to [READ_BUFFER_SIZE].
+ * @return A cold [Flow] that emits decoded elements of type [T].
+ * @throws IOException if an I/O error occurs while reading from the source.
+ */
+public fun <T> RawSource.asFlow(
+    decoder: StreamingDecoder<T>,
+    readBufferSize: Long = READ_BUFFER_SIZE
+): Flow<T> =
+    channelFlow {
+        val source = this@asFlow
+        val buffer = Buffer()
+        var decoderClosed = false
+        try {
+            source.use { source ->
+                while (isActive) {
+                    val bytesRead = source.readAtMostTo(buffer, readBufferSize)
+                    if (bytesRead == -1L) {
+                        break
+                    }
+
+                    if (bytesRead > 0L) {
+                        val bytes = buffer.readByteArray()
+                        buffer.clear()
+                        decoder.decode(bytes) {
+                            send(it)
+                        }
+                    }
+
+                    yield() // Giving other coroutines a chance to run
+                }
+            }
+            // Normal completion: emit any remaining buffered data
+            decoder.onClose { send(it) }
+            decoderClosed = true
+        } catch (exception: IOException) {
+            // IO error: try to emit remaining data, then close with error
+            runCatching { decoder.onClose { send(it) } }.onSuccess { decoderClosed = true }
+            throw exception
+        } finally {
+            // Ensure decoder cleanup even on cancellation or other exceptions
+            if (!decoderClosed) {
+                withContext(NonCancellable) {
+                    runCatching { decoder.onClose { /* discard data, cleanup only */ } }
+                }
+            }
+            buffer.clear()
+        }
+    }

core/common/src/coroutines/StreamingDecoder.kt

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2017-2025 JetBrains s.r.o. and respective authors and developers.
+ * Use of this source code is governed by the Apache 2.0 license that can be found in the LICENCE file.
+ */
+package kotlinx.io.coroutines
+
+/**
+ * A generic interface for decoding a stream of bytes into discrete elements of type [T].
+ *
+ * Implementations of this interface are responsible for processing input byte arrays, decoding
+ * them into meaningful elements, and delivering them to the provided `byteConsumer` function in
+ * sequential order. This allows for efficient handling of streaming data and enables
+ * processing without requiring the entire stream to be loaded into memory.
+ *
+ * ## Lifecycle
+ *
+ * The decoder processes a stream through repeated calls to [decode], followed by a final call
+ * to [onClose] when the stream ends. After [onClose] is called, the decoder should not be reused.
+ *
+ * ## Thread Safety
+ *
+ * Implementations are not required to be thread-safe. Each decoder instance should be used with
+ * a single stream and should not be shared across concurrent coroutines.
+ *
+ * @param T The type of elements produced by the decoder.
+ */
+public interface StreamingDecoder<T> {
+    /**
+     * Decodes a chunk of bytes from the input stream.
+     *
+     * This method may be called multiple times as data arrives. Implementations should buffer
+     * incomplete elements internally and emit complete elements via [byteConsumer].
+     *
+     * @param bytes The input byte array to decode.
+     * @param byteConsumer A suspend function that receives decoded elements.
+     */
+    public suspend fun decode(bytes: ByteArray, byteConsumer: suspend (T) -> Unit)
+
+    /**
+     * Called when the input stream ends, allowing the decoder to emit any remaining buffered data
+     * and perform cleanup.
+     *
+     * After this method is called, the decoder should not be used again.
+     *
+     * @param byteConsumer A suspend function that receives any final decoded elements.
+     */
+    public suspend fun onClose(byteConsumer: suspend (T) -> Unit)
+}
+