Repository: avro-kotlin/avro4k
Pull request #314 (2025-07-01 06:54:02)

build(deps): bump jacksonVersion from 2.18.2 to 2.19.1
Bumps `jacksonVersion` from 2.18.2 to 2.19.1.
Updates `com.fasterxml.jackson.module:jackson-module-kotlin` from 2.18.2 to 2.19.1
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/FasterXML/jackson-module-kotlin/commit/439c45e41c24f343d18b5d91379d41b73300382c"><code>439c45e</code></a> [maven-release-plugin] prepare release jackson-module-kotlin-2.19.1</li>
<li><a href="https://github.com/FasterXML/jackson-module-kotlin/commit/cfa10de9bec90f762d486a1444ec42f3aec47e8c"><code>cfa10de</code></a> Prep for 2.19.1</li>
<li><a href="https://github.com/FasterXML/jackson-module-kotlin/commit/ae9769c3e8df0a3f1e2e1dc507a39ade739f63f3"><code>ae9769c</code></a> Merge pull request <a href="https://redirect.github.com/FasterXML/jackson-module-kotlin/issues/1006">#1006</a> from k163377/fix-dir</li>
<li><a href="https://github.com/FasterXML/jackson-module-kotlin/commit/31cb6045cf161eb944a67b9972ec16342753daee"><code>31cb604</code></a> Fix to correctly hierarchize directories</li>
<li><a href="https://github.com/FasterXML/jackson-module-kotlin/commit/2a637fc242bef46c4e6d2d8d6e038ee1a7c3d341"><code>2a637fc</code></a> Merge pull request <a href="https://redirect.github.com/FasterXML/jackson-module-kotlin/issues/1005">#1005</a> from k163377/add-value-class-ser-tests</li>
<li><a href="https://github.com/FasterXML/jackson-module-kotlin/commit/5a29ae5775495d4a051e56aba3098d921f57dffa"><code>5a29ae5</code></a> Add serialization test cases for value classes</li>
<li><a href="https://github.com/FasterXML/jackson-module-kotlin/commit/cf0bdc8b0cf42c1799b94cfb537626c9de8745a7"><code>cf0bdc8</code></a> Fix redundant package names</li>
<li><a href="https://github.com/FasterXML/jackson-module-kotlin/commit/7cf0f24a5e6b519c1716d1de3452bccf8f052f2a"><code>7cf0f24</code></a> Merge pull request <a href="https://redirect.github.com/FasterXML/jackson-module-kotlin/issues/1003">#1003</a> from k163377/add-comments</li>
<li><a href="https://github.com/FasterXML/jackson-module-kotlin/commit/381125287a985de594d7017d6328f2db8da6affb"><code>3811252</code></a> Added comments regarding Duration conversion</li>
<li><a href="https://github.com/FasterXML/jackson-module-kotlin/commit/e59ebd274011c1afa32b8d172f7a68b00936a67e"><code>e59ebd2</code></a> Manual pom.xml merge</li>
<li>Additional commits viewable in <a href="https://github.com/FasterXML/jackson-module-kotlin/compare/jackson-module-kotlin-2.18.2...jackson-module-kotlin-2.19.1">compare view</a></li>
</ul>
</details>
<br />

Updates `com.fasterxml.jackson.datatype:jackson-datatype-jsr310` from 2.18.2 to 2.19.1

Updates `com.fasterxml.jackson.dataformat:jackson-dataformat-avro` from 2.18.2 to 2.19.1
<details>
<summary>Commits</summary>
<ul>
<li><a href="https://github.com/FasterXML/jackson-dataformats-binary/commit/ea2fe85f69f96ab6a1dd560579e5e59ecc110c0a"><code>ea2fe85</code></a> [maven-release-plugin] prepare release jackson-dataformats-binary-2.19.1</li>
<li><a href="https://github.com/FasterXML/jackson-dataformats-binary/commit/8fb1b5bfd7f01bbd4e6497e892ea3dde65d0a6c1"><code>8fb1b5b</code></a> Prep for 2.19.1</li>
<li><a href="https://github.com/FasterXML/jackson-dataformats-binary/commit/eb550ffa6f7353f316eda21e9fc198467ca8d482"><code>eb550ff</code></a> Improve <a href="https://redirect.github.com/FasterXML/jackson-dataformats-binary/issues/599">#599</a> test</li>
<li><a href="https://github.com/FasterXML/jackson-dataformats-binary/commit/f7ede19a8b277ac61eb7e471aac63f4dc36ae41e"><code>f7ede19</code></a> Add failing (&quot;tofix&quot;) test for <a href="https://redirect.github.com/FasterXML/jackson-dataformats-binary/issues/599">#599</a></li>
<li><a href="https://github.com/FasterXML/jackson-dataformats-binary/commit/a2384dddad92f78f75d2e7eee44c2aceed682f9d"><code>a2384dd</code></a> Manual pom.xml merge</li>
<li><a href="https://github.com/FasterXML/jackson-dataformats-binary/commit/b85f4988e93618775e4b78c35f7dc21c797c8827"><code>b85f498</code></a> Merge branch '2.18' into 2.19</li>
<li><a href="https://github.com/FasterXML/jackson-dataformats-binary/commit/198031ff9617b2f8b70d7885b06c51dff587ef33"><code>198031f</code></a> Update snapshot access to Central Portal</li>
<li><a href="https://github.com/FasterXML/jackson-dataformats-binary/commit/0d7f007645a1a4278658b5f432e406540cb6c78b"><code>0d7f007</code></a> [AVRO] <a href="https://redirect.github.com/FasterXML/jackson-dataformats-binary/issues/589">#589</a>: Fix schema not including base class for records with subclasses ...</li>
<li><a href="https://github.com/FasterXML/jackson-dataformats-binary/commit/0922af78b5e9c06694e310c5539417e7ce2d839b"><code>0922af7</code></a> Merge branch '2.18' into 2.19</li>
<li><a href="https://github.com/FasterXML/jackson-dataformats-binary/commit/36e2eb4c6dbe8a2c59e6a2f6050219e14c5ef8f6"><code>36e2eb4</code></a> Fix CI wrt branch restrictions</li>
<li>Additional commits viewable in <a href="https://github.com/FasterXML/jackson-dataformats-binary/compare/jackson-dataformats-binary-2.18.2...jackson-dataformats-binary-2.19.1">compare view</a></li>
</ul>
</details>
<br />


Dependabot will resolve any conflicts with this PR as long as you don't alter it yourself. You can also trigger a rebase manually by commenting `@dependabot rebase`.

[//]: # (dependabot-automerge-start)
[//]: # (dependabot-automerge-end)

---

<details>
<summary>Dependabot commands and options</summary>
<br />

You can trigger Dependabot actions by commenting on this PR:
- `@dependabot rebase` will rebase this PR
- `@dependabot recreate` will recreate this PR, overwriting any edits that have been made to it
- `@dependabot merge` will merge this PR after your CI passes on it
- `@dependabot squash and merge` will squash and merge this PR after your CI passes on it
- `@dependabot cancel merge` will cancel a previously requested merge and block automerging
- `@dependabot reopen` will reopen this PR if it is closed
- `@dependabot close` will close this PR and stop Dependabot recreating it. You can achieve the same result by closing it manually
- `@dependabot show <dependency name> ignore conditions` will show all of the ignore conditions of the specified dependency
- `@dependabot ignore this major version` will close this PR and stop Dependabot creating any more for this major version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this minor version` will close this PR and stop Dependabot creating any more for this minor version (unless you reopen the PR or upgrade to it yourself)
- `@dependabot ignore this dependency` will close this PR and stop Dependabot creating any more for this dependency (unless you reopen the PR or upgrade to it yourself)


</details>


Patch:
diff --git a/README.md b/README.md
index 46d9821..13dd325 100644
--- a/README.md
+++ b/README.md
@@ -25,7 +25,9 @@ Here are the main features:
 
 # Quick start
 
-## Basic encoding
+## Basic "pure" avro encoding
+
+In this example, we will see how to encode and decode Avro objects in their "pure" format, which means that the schema is not prefixed to the data.
 
 <details open>
 <summary>Example:</summary>
@@ -56,6 +58,132 @@ fun main() {
 
 </details>
 
+### JVM stream encoding
+
+Avro4k is also able to encode and decode objects to and from any kotlinx-io `Sink` and `Source`, which means any supported bridge to kotlinx-io is supported by avro4k.
+
+As an example, for JVM streams, you can use `InputStream.asSource().buffered()` and `OutputStream.asSink().buffered()`.
+
+> [!NOTE]
+> Note the required `buffered()` call to allow avro4k accessing the `read` and `write` methods.
+
+> [!WARNING]
+> Do not use `ByteArrayInputStream` and `ByteArrayOutputStream`, prefer using `kotlinx.io.Buffer` instead as it is more efficient and allows better performance.
+
+<details>
+<summary>Example:</summary>
+
+```kotlin
+package myapp
+
+import com.github.avrokotlin.avro4k.*
+import kotlinx.serialization.*
+import java.io.*
+import kotlinx.io.*
+
+@Serializable
+data class Project(val name: String, val language: String)
+
+fun main() {
+    // Serializing objects
+    val data = Project("kotlinx.serialization", "Kotlin")
+    val outputStream: OutputStream = TODO("Your output stream here, e.g. FileOutputStream, etc.")
+    val sink = output.asSink().buffered()
+    Avro.encodeToSink(data, sink)
+    sink.close() // Will flush the data to the output stream at the same time
+
+    // Deserializing objects
+    val inputStream: InputStream = TODO("Your input stream here, e.g. FileInputStream, etc.")
+    val source = inputStream.asSource().buffered()
+    // Be sure to read all the content from the soure, or you may lose data from the upper stream, as buffering means that the source may read more bytes to fill the buffer.
+    val obj = Avro.decodeFromSource<Project>(source)
+    println(obj) // Project(name=kotlinx.serialization, language=Kotlin)
+}
+```
+
+</details>
+
+### Encode many objects in the same stream
+
+You can use kotlinx-io's `Buffer` to encode many objects in the same stream, and then extract the whole bytes from the buffer.
+
+> [!WARNING]
+> As kotlinx-io is using an internal buffer pool, each `Buffer` instance you create must be fully consumed to allow the internal buffer segments to be returned to the pool and reused.
+> More detailsin the [kotlinx-io documentation](https://github.com/Kotlin/kotlinx-io).
+
+<details>
+<summary>Example:</summary>
+
+```kotlin
+package myapp
+
+import com.github.avrokotlin.avro4k.*
+import kotlinx.serialization.*
+import kotlinx.io.*
+
+@Serializable
+data class Project(val name: String, val language: String)
+
+fun main() {
+    // Serializing objects
+    val buffer = Buffer()
+    Avro.encodeToSink(data1, buffer)
+    Avro.encodeToSink(data2, buffer)
+    Avro.encodeToSink(data3, buffer)
+    // You need to consume the whole buffer by passing the buffer to the wanted framework or read the byte-array, or you may have kotlinx-io internal buffer leaks.
+    val bytes = buffer.readByteArray()
+
+    // Deserializing objects
+    val source: Source = TODO("a Buffer, or any other source from kotlinx-io")
+    val data1 = Avro.decodeFromSource<Project>(source)
+    val data2 = Avro.decodeFromSource<Project>(source)
+    val data3 = Avro.decodeFromSource<Project>(source)
+}
+```
+
+</details>
+
+### Custom encoding with a Buffer
+
+The following example shows how to encode in a custom format using a `Buffer` to write the data. The example is going to encode like this: `<timestamp><schema><binary-length><binary-data>`.
+
+You will still encode the data in pure avro binary format, while you will be able to add some metadata before the data (or after), compress the data, and much more.
+
+<details>
+<summary>Example:</summary>
+
+```kotlin
+package myapp
+
+import com.github.avrokotlin.avro4k.*
+import kotlinx.serialization.*
+import kotlinx.io.*
+
+@Serializable
+data class Project(val name: String, val language: String)
+
+fun main() {
+    // Serializing objects
+    val buffer = Buffer()
+    buffer.writeInt(theTimestamp)
+    buffer.writeString(theSchema)
+    
+    val dataBuffer = Buffer()
+    Avro.encodeToSink(data, dataBuffer)
+    // You need to consume the whole buffer by passing the buffer to the wanted framework or read the byte-array, or you may have kotlinx-io internal buffer leaks.
+    val bytes = buffer.readByteArray()
+    buffer.writeInt()
+
+    // Deserializing objects
+    val source: Source = TODO("a Buffer, or any other source from kotlinx-io")
+    val data1 = Avro.decodeFromSource<Project>(source)
+    val data2 = Avro.decodeFromSource<Project>(source)
+    val data3 = Avro.decodeFromSource<Project>(source)
+}
+```
+
+</details>
+
 ## Single object
 
 Avro4k provides a way to encode and decode single objects with `AvroSingleObject` class. This encoding will prefix the binary data with the schema fingerprint to
@@ -826,7 +954,7 @@ You can force a field (or the value class' property) to have its inferred schema
 
 Compatible types visible in the [types matrix](#types-matrix), indicated by the "Other compatible writer types" column. The **writer schema compatibility is still respected**, so if the field has been written as an int, a stringified int will be deserialized as an int without the need of parsing it. It is the same for the rerverse: If an int has been written as a string, it will be deserialized as an int by parsing the string content.
 
-> [!INFO]
+> [!NOTE]
 > Note that the type must be compatible with the `string` type, otherwise it will be ignored.
 > Your custom serializer generated schema must handle this annotation, or it will be ignored.
 
@@ -868,6 +996,7 @@ So to mark a field as optional and facilitate avro contract evolution regarding
 
 - Kotlin 1.7.20 up to 1.8.10 cannot properly compile @SerialInfo-Annotations on enums (see https://github.com/Kotlin/kotlinx.serialization/issues/2121).
   This is fixed with kotlin 1.8.20. So if you are planning to use any of avro4k's annotations on enum types, please make sure that you are using kotlin >= 1.8.20.
+- Avro 1.12.0 introduced corruption for list of double in generic encoding, where doubles are casted to floats. Will be fixed for the next version when it will be released 
 
 # Migrating from v1 to v2
 Heads up to the [migration guide](Migrating-from-v1.md) to update your code from avro4k v1 to v2.
diff --git a/api/avro4k-core.api b/api/avro4k-core.api
index bab51a1..b3c4e7d 100644
--- a/api/avro4k-core.api
+++ b/api/avro4k-core.api
@@ -140,6 +140,11 @@ public final class com/github/avrokotlin/avro4k/AvroJVMExtensionsKt {
 	public static final fun encodeToStream (Lcom/github/avrokotlin/avro4k/Avro;Lorg/apache/avro/Schema;Lkotlinx/serialization/SerializationStrategy;Ljava/lang/Object;Ljava/io/OutputStream;)V
 }
 
+public final class com/github/avrokotlin/avro4k/AvroKotlinxIoExtensionsKt {
+	public static final fun decodeFromSource (Lcom/github/avrokotlin/avro4k/Avro;Lorg/apache/avro/Schema;Lkotlinx/serialization/DeserializationStrategy;Lkotlinx/io/Source;)Ljava/lang/Object;
+	public static final fun encodeToSink (Lcom/github/avrokotlin/avro4k/Avro;Lorg/apache/avro/Schema;Lkotlinx/serialization/SerializationStrategy;Ljava/lang/Object;Lkotlinx/io/Sink;)V
+}
+
 public final class com/github/avrokotlin/avro4k/AvroKt {
 	public static final fun Avro (Lcom/github/avrokotlin/avro4k/Avro;Lkotlin/jvm/functions/Function1;)Lcom/github/avrokotlin/avro4k/Avro;
 	public static synthetic fun Avro$default (Lcom/github/avrokotlin/avro4k/Avro;Lkotlin/jvm/functions/Function1;ILjava/lang/Object;)Lcom/github/avrokotlin/avro4k/Avro;
@@ -212,8 +217,10 @@ public final class com/github/avrokotlin/avro4k/AvroSingleObject : kotlinx/seria
 	public fun <init> (Lkotlin/jvm/functions/Function1;Lcom/github/avrokotlin/avro4k/Avro;)V
 	public synthetic fun <init> (Lkotlin/jvm/functions/Function1;Lcom/github/avrokotlin/avro4k/Avro;ILkotlin/jvm/internal/DefaultConstructorMarker;)V
 	public fun decodeFromByteArray (Lkotlinx/serialization/DeserializationStrategy;[B)Ljava/lang/Object;
+	public final fun decodeFromSource (Lkotlinx/serialization/DeserializationStrategy;Lkotlinx/io/Source;)Ljava/lang/Object;
 	public final fun decodeFromStream (Lkotlinx/serialization/DeserializationStrategy;Ljava/io/InputStream;)Ljava/lang/Object;
 	public fun encodeToByteArray (Lkotlinx/serialization/SerializationStrategy;Ljava/lang/Object;)[B
+	public final fun encodeToSink (Lorg/apache/avro/Schema;Lkotlinx/serialization/SerializationStrategy;Ljava/lang/Object;Lkotlinx/io/Sink;)V
 	public final fun encodeToStream (Lorg/apache/avro/Schema;Lkotlinx/serialization/SerializationStrategy;Ljava/lang/Object;Ljava/io/OutputStream;)V
 	public final fun getAvro ()Lcom/github/avrokotlin/avro4k/Avro;
 	public fun getSerializersModule ()Lkotlinx/serialization/modules/SerializersModule;
diff --git a/benchmark/README.md b/benchmark/README.md
index ecf850a..e2976c8 100644
--- a/benchmark/README.md
+++ b/benchmark/README.md
@@ -26,34 +26,34 @@ Computer: Macbook air M2
 
 ```
 Benchmark                                                       Mode   Cnt     Score       Error   Units
-ApacheAvroReflectBenchmark.read            thrpt    5  28819.641 ± 1048.236  ops/s
-Avro4kBenchmark.read                       thrpt    5  27259.164 ±  418.829  ops/s
-Avro4kGenericWithApacheAvroBenchmark.read  thrpt    5  16242.380 ±  123.109  ops/s
-
-Avro4kBenchmark.write                       thrpt    5  55760.544 ±  394.596  ops/s
-ApacheAvroReflectBenchmark.write            thrpt    5  48015.770 ± 4021.317  ops/s
-JacksonAvroBenchmark.write                  thrpt    5  34701.945 ±  719.865  ops/s
-Avro4kGenericWithApacheAvroBenchmark.write  thrpt    5  23859.446 ± 3207.023  ops/s
-
-Avro4kSimpleBenchmark.read                       thrpt    5  224766.920 ± 21004.000  ops/s
-ApacheAvroReflectSimpleBenchmark.read            thrpt    5  214986.578 ± 23138.102  ops/s
-Avro4kGenericWithApacheAvroSimpleBenchmark.read  thrpt    5  146503.504 ±  2613.945  ops/s
-JacksonAvroSimpleBenchmark.read                  thrpt    5   66071.428 ±  1571.426  ops/s
-
-Avro4kSimpleBenchmark.write                       thrpt    5  448433.701 ±  2086.119  ops/s
-ApacheAvroReflectSimpleBenchmark.write            thrpt    5  245640.080 ±  3835.344  ops/s
-Avro4kGenericWithApacheAvroSimpleBenchmark.write  thrpt    5  158308.148 ± 12314.688  ops/s
-JacksonAvroSimpleBenchmark.write                  thrpt    5  125374.072 ±  5240.116  ops/s
-
-JacksonAvroListsBenchmark.read                  thrpt    5  6.329 ± 0.631  ops/s
-Avro4kListsBenchmark.read                       thrpt    5  5.074 ± 1.624  ops/s
-ApacheAvroReflectListsBenchmark.read            thrpt    5  4.048 ± 0.366  ops/s
-Avro4kGenericWithApacheAvroListsBenchmark.read  thrpt    5  3.709 ± 1.694  ops/s
-
-ApacheAvroReflectListsBenchmark.write            thrpt    5  7.332 ± 0.532  ops/s
-Avro4kListsBenchmark.write                       thrpt    5  6.074 ± 0.087  ops/s
-JacksonAvroListsBenchmark.write                  thrpt    5  3.590 ± 1.973  ops/s
-Avro4kGenericWithApacheAvroListsBenchmark.write  thrpt    5  3.406 ± 0.900  ops/s
+c.g.a.b.complex.ApacheAvroReflectBenchmark.read                  thrpt    5   29295.192 ±  319.312  ops/s
+c.g.a.b.complex.Avro4kBenchmark.read                             thrpt    5   25766.052 ±  368.664  ops/s
+c.g.a.b.complex.Avro4kGenericWithApacheAvroBenchmark.read        thrpt    5   16209.174 ±  203.406  ops/s
+
+c.g.a.b.complex.ApacheAvroReflectBenchmark.write                 thrpt    5   46823.034 ± 2011.928  ops/s
+c.g.a.b.complex.Avro4kBenchmark.write                            thrpt    5   44890.274 ± 1080.358  ops/s
+c.g.a.b.complex.JacksonAvroBenchmark.write                       thrpt    5   34798.967 ± 2787.370  ops/s
+c.g.a.b.complex.Avro4kGenericWithApacheAvroBenchmark.write       thrpt    5   24803.165 ± 1288.472  ops/s
+
+c.g.a.b.simple.Avro4kSimpleBenchmark.read                        thrpt    5  213794.225 ± 9560.253  ops/s
+c.g.a.b.simple.ApacheAvroReflectSimpleBenchmark.read             thrpt    5  212685.281 ± 2408.201  ops/s
+c.g.a.b.simple.Avro4kGenericWithApacheAvroSimpleBenchmark.read   thrpt    5  149715.561 ± 2054.725  ops/s
+c.g.a.b.simple.JacksonAvroSimpleBenchmark.read                   thrpt    5   62663.528 ± 2370.041  ops/s
+
+c.g.a.b.simple.Avro4kSimpleBenchmark.write                       thrpt    5  328514.289 ± 1119.183  ops/s
+c.g.a.b.simple.ApacheAvroReflectSimpleBenchmark.write            thrpt    5  253784.541 ± 8960.212  ops/s
+c.g.a.b.simple.Avro4kGenericWithApacheAvroSimpleBenchmark.write  thrpt    5  169881.518 ± 2879.094  ops/s
+c.g.a.b.simple.JacksonAvroSimpleBenchmark.write                  thrpt    5  128048.496 ± 6859.076  ops/s
+
+c.g.a.b.lists.JacksonAvroListsBenchmark.read                     thrpt    5       7.865 ±    0.488  ops/s
+c.g.a.b.lists.Avro4kListsBenchmark.read                          thrpt    5       6.493 ±    0.211  ops/s
+c.g.a.b.lists.Avro4kGenericWithApacheAvroListsBenchmark.read     thrpt    5       5.416 ±    0.170  ops/s
+c.g.a.b.lists.ApacheAvroReflectListsBenchmark.read               thrpt    5       4.628 ±    1.104  ops/s
+
+c.g.a.b.lists.ApacheAvroReflectListsBenchmark.write              thrpt    5      10.237 ±    0.334  ops/s
+c.g.a.b.lists.Avro4kListsBenchmark.write                         thrpt    5       8.104 ±    0.132  ops/s
+c.g.a.b.lists.JacksonAvroListsBenchmark.write                    thrpt    5       5.881 ±    0.924  ops/s
+c.g.a.b.lists.Avro4kGenericWithApacheAvroListsBenchmark.write    thrpt    5       4.721 ±    0.249  ops/s
 
 ```
 
diff --git a/benchmark/build.gradle.kts b/benchmark/build.gradle.kts
index 524eea2..599bd77 100644
--- a/benchmark/build.gradle.kts
+++ b/benchmark/build.gradle.kts
@@ -62,7 +62,7 @@ dependencies {
     implementation("org.apache.commons:commons-lang3:3.17.0")
     implementation("org.jetbrains.kotlinx:kotlinx-benchmark-runtime:0.4.13")
 
-    val jacksonVersion = "2.18.2"
+    val jacksonVersion = "2.19.1"
     implementation("com.fasterxml.jackson.module:jackson-module-kotlin:$jacksonVersion")
     implementation("com.fasterxml.jackson.datatype:jackson-datatype-jsr310:$jacksonVersion")
     implementation("com.fasterxml.jackson.dataformat:jackson-dataformat-avro:$jacksonVersion")
diff --git a/benchmark/src/main/kotlin/com/github/avrokotlin/benchmark/ManualProfiling.kt b/benchmark/src/main/kotlin/com/github/avrokotlin/benchmark/ManualProfiling.kt
index 6997d8d..c18a4bf 100644
--- a/benchmark/src/main/kotlin/com/github/avrokotlin/benchmark/ManualProfiling.kt
+++ b/benchmark/src/main/kotlin/com/github/avrokotlin/benchmark/ManualProfiling.kt
@@ -10,7 +10,6 @@ internal object ManualProfilingWrite {
             for (i in 0 until 1_000_000) {
                 if (i % 1_000 == 0) println("Iteration $i")
                 write()
-                read()
             }
         }
     }
diff --git a/benchmark/src/main/kotlin/com/github/avrokotlin/benchmark/complex/Avro4kBenchmark.kt b/benchmark/src/main/kotlin/com/github/avrokotlin/benchmark/complex/Avro4kBenchmark.kt
index 41be322..7969e0c 100644
--- a/benchmark/src/main/kotlin/com/github/avrokotlin/benchmark/complex/Avro4kBenchmark.kt
+++ b/benchmark/src/main/kotlin/com/github/avrokotlin/benchmark/complex/Avro4kBenchmark.kt
@@ -3,9 +3,11 @@ package com.github.avrokotlin.benchmark.complex
 import com.github.avrokotlin.avro4k.Avro
 import com.github.avrokotlin.avro4k.decodeFromByteArray
 import com.github.avrokotlin.avro4k.encodeToByteArray
-import com.github.avrokotlin.avro4k.encodeToStream
+import com.github.avrokotlin.avro4k.encodeToSink
 import com.github.avrokotlin.benchmark.internal.Clients
 import kotlinx.benchmark.Benchmark
+import kotlinx.io.asSink
+import kotlinx.io.buffered
 import kotlinx.serialization.ExperimentalSerializationApi
 import java.io.OutputStream
 
@@ -27,6 +29,6 @@ internal class Avro4kBenchmark : SerializationBenchmark() {
     @OptIn(ExperimentalSerializationApi::class)
     @Benchmark
     fun write() {
-        Avro.encodeToStream(schema, clients, OutputStream.nullOutputStream())
+        Avro.encodeToSink(schema, clients, OutputStream.nullOutputStream().asSink().buffered())
     }
 }
\ No newline at end of file
diff --git a/benchmark/src/main/kotlin/com/github/avrokotlin/benchmark/lists/Avro4kListsBenchmark.kt b/benchmark/src/main/kotlin/com/github/avrokotlin/benchmark/lists/Avro4kListsBenchmark.kt
index b03eb67..a37e481 100644
--- a/benchmark/src/main/kotlin/com/github/avrokotlin/benchmark/lists/Avro4kListsBenchmark.kt
+++ b/benchmark/src/main/kotlin/com/github/avrokotlin/benchmark/lists/Avro4kListsBenchmark.kt
@@ -3,9 +3,11 @@ package com.github.avrokotlin.benchmark.lists
 import com.github.avrokotlin.avro4k.Avro
 import com.github.avrokotlin.avro4k.decodeFromByteArray
 import com.github.avrokotlin.avro4k.encodeToByteArray
-import com.github.avrokotlin.avro4k.encodeToStream
+import com.github.avrokotlin.avro4k.encodeToSink
 import com.github.avrokotlin.benchmark.internal.ListWrapperDatasClass
 import kotlinx.benchmark.Benchmark
+import kotlinx.io.asSink
+import kotlinx.io.buffered
 import kotlinx.serialization.ExperimentalSerializationApi
 import java.io.OutputStream
 
@@ -27,6 +29,6 @@ internal class Avro4kListsBenchmark : SerializationListsBenchmark() {
     @OptIn(ExperimentalSerializationApi::class)
     @Benchmark
     fun write() {
-        Avro.encodeToStream(schema, lists, OutputStream.nullOutputStream())
+        Avro.encodeToSink(schema, lists, OutputStream.nullOutputStream().asSink().buffered())
     }
 }
\ No newline at end of file
diff --git a/benchmark/src/main/kotlin/com/github/avrokotlin/benchmark/simple/Avro4kSimpleBenchmark.kt b/benchmark/src/main/kotlin/com/github/avrokotlin/benchmark/simple/Avro4kSimpleBenchmark.kt
index 580a94b..398e46b 100644
--- a/benchmark/src/main/kotlin/com/github/avrokotlin/benchmark/simple/Avro4kSimpleBenchmark.kt
+++ b/benchmark/src/main/kotlin/com/github/avrokotlin/benchmark/simple/Avro4kSimpleBenchmark.kt
@@ -3,9 +3,11 @@ package com.github.avrokotlin.benchmark.simple
 import com.github.avrokotlin.avro4k.Avro
 import com.github.avrokotlin.avro4k.decodeFromByteArray
 import com.github.avrokotlin.avro4k.encodeToByteArray
-import com.github.avrokotlin.avro4k.encodeToStream
+import com.github.avrokotlin.avro4k.encodeToSink
 import com.github.avrokotlin.benchmark.internal.SimpleDatasClass
 import kotlinx.benchmark.Benchmark
+import kotlinx.io.asSink
+import kotlinx.io.buffered
 import kotlinx.serialization.ExperimentalSerializationApi
 import java.io.OutputStream
 
@@ -27,6 +29,6 @@ internal class Avro4kSimpleBenchmark : SerializationSimpleBenchmark() {
     @OptIn(ExperimentalSerializationApi::class)
     @Benchmark
     fun write() {
-        Avro.encodeToStream(schema, clients, OutputStream.nullOutputStream())
+        Avro.encodeToSink(schema, clients, OutputStream.nullOutputStream().asSink().buffered())
     }
 }
\ No newline at end of file
diff --git a/build.gradle.kts b/build.gradle.kts
index 56cd15f..309128b 100644
--- a/build.gradle.kts
+++ b/build.gradle.kts
@@ -20,6 +20,7 @@ version = Ci.publishVersion
 dependencies {
     api(libs.apache.avro)
     api(libs.kotlinx.serialization.core)
+    api(libs.kotlinx.io)
     implementation(libs.kotlinx.serialization.json)
     implementation(libs.okio)
     testImplementation(libs.kotest.junit5)
diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml
index a4a1e3c..5e2cadc 100644
--- a/gradle/libs.versions.toml
+++ b/gradle/libs.versions.toml
@@ -1,7 +1,7 @@
 [versions]
 kotlin = "2.1.21"
-jvm = "21"
 kotlinxSerialization = "1.8.1"
+kotlinx-io = "0.8.0"
 kotestVersion = "5.9.1"
 okio = "3.9.0"
 apache-avro = "1.11.4"
@@ -12,6 +12,7 @@ apache-avro = { group = "org.apache.avro", name = "avro", version.ref = "apache-
 okio = { group = "com.squareup.okio", name = "okio", version.ref = "okio" }
 kotlinx-serialization-core = { group = "org.jetbrains.kotlinx", name = "kotlinx-serialization-core", version.ref = "kotlinxSerialization" }
 kotlinx-serialization-json = { group = "org.jetbrains.kotlinx", name = "kotlinx-serialization-json", version.ref = "kotlinxSerialization" }
+kotlinx-io = { group = "org.jetbrains.kotlinx", name = "kotlinx-io-core", version.ref = "kotlinx-io" }
 kotest-core = { group = "io.kotest", name = "kotest-assertions-core", version.ref = "kotestVersion" }
 kotest-json = { group = "io.kotest", name = "kotest-assertions-json", version.ref = "kotestVersion" }
 kotest-junit5 = { group = "io.kotest", name = "kotest-runner-junit5", version.ref = "kotestVersion" }
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/Avro.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/Avro.kt
index bc06787..2957cf8 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/Avro.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/Avro.kt
@@ -1,12 +1,14 @@
 package com.github.avrokotlin.avro4k
 
+import com.github.avrokotlin.avro4k.internal.Buffer
 import com.github.avrokotlin.avro4k.internal.EnumResolver
 import com.github.avrokotlin.avro4k.internal.PolymorphicResolver
 import com.github.avrokotlin.avro4k.internal.RecordResolver
-import com.github.avrokotlin.avro4k.internal.decodeWithApacheDecoder
 import com.github.avrokotlin.avro4k.internal.schema.ValueVisitor
 import com.github.avrokotlin.avro4k.serializer.JavaStdLibSerializersModule
 import com.github.avrokotlin.avro4k.serializer.JavaTimeSerializersModule
+import kotlinx.io.Buffer
+import kotlinx.io.readByteArray
 import kotlinx.serialization.BinaryFormat
 import kotlinx.serialization.DeserializationStrategy
 import kotlinx.serialization.ExperimentalSerializationApi
@@ -19,13 +21,17 @@ import kotlinx.serialization.modules.SerializersModule
 import kotlinx.serialization.modules.overwriteWith
 import kotlinx.serialization.modules.plus
 import kotlinx.serialization.serializer
-import okio.Buffer
 import org.apache.avro.Schema
-import org.apache.avro.io.DecoderFactory
 import java.util.WeakHashMap
 
 /**
- * The goal of this class is to serialize and deserialize in avro binary format, not in GenericRecords.
+ * The goal of this class is to serialize and deserialize in avro binary format.
+ * That said, following the avro specification, each logical type is translated to a specific avro type and serialized as a specific byte sequence (strings in UTF8 bytes, int as zig-zag varint, etc).
+ *
+ * The main advantage of this library is that it allows you to serialize your data classes (and any other serializable type) to avro binary format, with the kotlin-style native serialization API provided by `kotlinx.serialization`.
+ * This framework maintained by jetbrains provides a powerful framework and gradle plugin to generate serializers at compile time, just with a few [kotlinx.serialization.Serializable] anotations.
+ *
+ * @sample com.github.avrokotlin.avro4k.internal.Samples.customizeAvroInstance
  */
 public sealed class Avro(
     public val configuration: AvroConfiguration,
@@ -43,6 +49,47 @@ public sealed class Avro(
             JavaTimeSerializersModule
     )
 
+    /**
+     * Infers the Avro schema for the given [descriptor].
+     *
+     * You may not need to call this method if you use the other encode/decode extension methods, as they will automatically generate the schema for you based on the serializer's descriptor.
+     *
+     * ## Types mapping
+     * - Classes and objects are translated as Avro records
+     * - Sealed classes are translated as UNION types
+     * - abstract classes must have their subclasses registered in the [Avro.serializersModule] to be properly translated
+     * - Enums are translated as Avro enums, taking into account the [AvroEnumDefault] annotation to set the default value
+     * - value classes are inlined as their underlying type
+     * - Primitive types such as [Boolean], [Int], [Long], [Float], [Double], and [String] are translated as their respective Avro types.
+     * - [Byte] and [Short] are translated as [Int] in Avro.
+     * - [UByte], [UShort], [UInt], and [ULong] are translated as their signed counterparts ([Int] and [Long]).
+     * - [ByteArray] is translated as BYTES type.
+     * - For the FIXED type, you can wrap a [ByteArray] in a value class, and annotate the underlying field with [AvroFixed]
+     * - [Map] is translated as a MAP type, where the keys are always converted from/to [String] (only for the primitive types)
+     * - Any other serializable [Iterable] is translated as an ARRAY type
+     * - Nullable types are translated as UNION types, with the null type being the first type in the union
+     * - [AvroDoc] annotation will be used to generate the Avro doc for a field or a named type (record, enum, fixed)
+     * - [AvroAlias] annotation will be used to set one or more aliases for a field or a named type (record, enum, fixed)
+     * - Annotate fields with [AvroStringable] to force the field type to be a string in the schema
+     * - Annotate a field with [AvroDefault] to set a default value for the field in the schema. The default value must be compatible with the field type.
+     * - Annotate a [java.math.BigDecimal] field with [AvroDecimal] to specify the scale and precision of the decimal avro logical type
+     * - Annotate a field or a type with [AvroProp] to add a custom property to the generate type's schema
+     *
+     * ## Cache details
+     * This method will cache the schema for the given descriptor, so subsequent calls with the same descriptor will return the cached schema.
+     * The cache is weakly referenced, meaning that it will not prevent the descriptor from being garbage collected.
+     * Also, the cache is not bounded, so calling this method with many different descriptors may lead to increased memory usage.
+     *
+     * ## Compatibility
+     * Do not forget to annotate evolving types and field names with [AvroAlias] to ensure compatibility with existing schemas and the used writer schemas.
+     *
+     * @param descriptor the descriptor for which to generate the schema.
+     * @return the Avro schema for the given descriptor.
+     *
+     * @see AvroConfiguration
+     * @see AvroBuilder.serializersModule
+     * @see com.github.avrokotlin.avro4k.serializer.AvroSerializer
+     */
     public fun schema(descriptor: SerialDescriptor): Schema {
         return schemaCache.getOrPut(descriptor) {
             lateinit var output: Schema
@@ -51,6 +98,29 @@ public sealed class Avro(
         }
     }
 
+    /**
+     * Encodes the given [value] to a [ByteArray] using the provided [writerSchema] and [serializer].
+     *
+     * You may prefer the other extension methods that infer the [serializer] and [writerSchema] based on a generic type [T].
+     *
+     * The given [serializer] must be "at least" compatible with the [writerSchema], meaning that any class field that is not present in the [writerSchema] will be ignored during encoding.
+     * Avro4k also match the [AvroAlias] of the classes and fields, and matches the [writerSchema]'s aliases to maximize the chances of compatibility with existing schemas.
+     * The schema can be generated using the [schema] method, or may be different, so that avro4k will adapt the value to the schema.
+     *
+     * ## Supported conversions
+     * - int types [Byte], [Short], [Int], [Long], their unsigned counterparts, and [Char] can be encoded INT or LONG
+     * - [Long] can be encoded to INT, but will fail if the value doesn't fit in the int range (same for unsigned types)
+     * - [Float] can be encoded to DOUBLE
+     * - [Double] can be encoded to FLOAT but will fail if the value doesn't fit in the float range
+     * - [String] can be encoded to any numerical type if the string is parsable
+     * - [String] can be encoded as UTF8 to BYTES and FIXED types, the later requiring the fixed size to match the actual string length
+     * - [String] can be encoded as ENUM
+     * - [Char] can be encoded as STRING
+     * - Classes and objects are encoded as RECORD, taking into account both [AvroAlias] and the [writerSchema]'s aliases to maximize compatibility
+     *
+     * @see AvroConfiguration
+     * @see AvroBuilder.serializersModule
+     */
     public fun <T> encodeToByteArray(
         writerSchema: Schema,
         serializer: SerializationStrategy<T>,
@@ -61,19 +131,60 @@ public sealed class Avro(
         return buffer.readByteArray()
     }
 
+    /**
+     * Decode the value from the given [bytes] to a value of type [T] assuming the data is respecting the given [writerSchema].
+     * If during the decoding, the written data does not match the [writerSchema], you may end up to unexpected results or an exception.
+     *
+     * You may prefer the other extension methods that infer the [deserializer] and [writerSchema] based on a generic type [T].
+     *
+     * For any custom encoding or conversion of type [T] or its inner type, you may need to register a custom [kotlinx.serialization.KSerializer] in [Avro.serializersModule].
+     *
+     * @param writerSchema the schema to use for decoding the value.
+     * @param deserializer the deserialization strategy to use for decoding the value. You may prefer the other extension methods without the deserializer parameter for convenience.
+     * @param bytes the [ByteArray] to read the encoded value from.
+     *
+     * @throws SerializationException if not all bytes were consumed during deserialization, or if the data does not match the [writerSchema].
+     *
+     * @see AvroConfiguration
+     * @see AvroBuilder.serializersModule
+     */
     public fun <T> decodeFromByteArray(
         writerSchema: Schema,
         deserializer: DeserializationStrategy<T>,
         bytes: ByteArray,
     ): T {
-        val binaryDecoder = DecoderFactory.get().binaryDecoder(bytes, null)
-        val result = decodeWithApacheDecoder(writerSchema, deserializer, binaryDecoder)
-        if (!binaryDecoder.isEnd) {
+        val source = Buffer(bytes)
+        val result = decodeFromSource(writerSchema, deserializer, source)
+        if (!source.exhausted()) {
             throw SerializationException("Not all bytes were consumed during deserialization")
         }
         return result
     }
 
+    /**
+     * Decode the value from the given [bytes] to a value of type [T] assuming the data is respecting the given [writerSchema].
+     * If during the decoding, the written data does not match the [writerSchema], you may end up to unexpected results or an exception.
+     *
+     * You may prefer the other extension methods that infer the [serializer] based on a generic type [T].
+     *
+     * For any custom encoding or conversion of type [T] or its inner type, you may need to register a custom [kotlinx.serialization.KSerializer] in [Avro.serializersModule].
+     *
+     * ## Supported conversions
+     * - LONG and INT types can be decoded into [Byte], [Short], [Int], [Long], their unsigned counterparts, and [Char] (if the value fits in the type range)
+     * - FLOAT and DOUBLE types can be decoded into [Float] and [Double] (if the value fits in the type range)
+     * - STRING can be decoded into any numerical type, as long as the string is parsable
+     * - STRING can be decoded to [Char], as long as the string is a single character
+     * - RECORD can be decoded to classes and objects, taking into account both [AvroAlias] and the [writerSchema]'s aliases to maximize compatibility
+     * - ENUM can be decoded to [String] or [Enum] types, as long as the enum value is present in the schema
+     *
+     * @param deserializer the deserialization strategy to use for decoding the value. You may prefer the other extension methods without the deserializer parameter for convenience.
+     * @param bytes the [ByteArray] to read the encoded value from.
+     * @return the decoded value of type [T].
+     *
+     * @see AvroConfiguration
+     * @see Avro.schema
+     * @see AvroBuilder.serializersModule
+     */
     override fun <T> decodeFromByteArray(
         deserializer: DeserializationStrategy<T>,
         bytes: ByteArray,
@@ -81,6 +192,12 @@ public sealed class Avro(
         return decodeFromByteArray(schema(deserializer.descriptor), deserializer, bytes)
     }
 
+    /**
+     * Encodes the given [value] to a [ByteArray] using the provided [serializer].
+     * The schema is inferred from the [serializer]'s descriptor, and the value is encoded using that schema.
+     *
+     * You may prefer the other extension methods that infer the [serializer] based on a generic type [T].
+     */
     override fun <T> encodeToByteArray(
         serializer: SerializationStrategy<T>,
         value: T,
@@ -99,17 +216,36 @@ public fun Avro(
 }
 
 public class AvroBuilder internal constructor(avro: Avro) {
+    /**
+     * @see AvroConfiguration.fieldNamingStrategy
+     */
     @ExperimentalSerializationApi
     public var fieldNamingStrategy: FieldNamingStrategy = avro.configuration.fieldNamingStrategy
 
+    /**
+     * @see AvroConfiguration.implicitNulls
+     */
     @ExperimentalSerializationApi
     public var implicitNulls: Boolean = avro.configuration.implicitNulls
 
+    /**
+     * @see AvroConfiguration.implicitEmptyCollections
+     */
     @ExperimentalSerializationApi
     public var implicitEmptyCollections: Boolean = avro.configuration.implicitEmptyCollections
 
+    /**
+     * @see AvroConfiguration.validateSerialization
+     */
     @ExperimentalSerializationApi
     public var validateSerialization: Boolean = avro.configuration.validateSerialization
+
+    /**
+     * The serializers module to use for encoding, decoding, and inferring schemas from kotlin types.
+     *
+     * @see KSerializer
+     * @see com.github.avrokotlin.avro4k.serializer.AvroSerializer
+     */
     public var serializersModule: SerializersModule = EmptySerializersModule()
 
     internal fun build(): AvroConfiguration =
@@ -124,15 +260,30 @@ public class AvroBuilder internal constructor(avro: Avro) {
 private class AvroImpl(configuration: AvroConfiguration, serializersModule: SerializersModule) :
     Avro(configuration, serializersModule)
 
+/**
+ * Infers the Avro schema for the given type [T].
+ *
+ * @see Avro.schema
+ */
 public inline fun <reified T> Avro.schema(): Schema {
     val serializer = serializersModule.serializer<T>()
     return schema(serializer.descriptor)
 }
 
+/**
+ * Infers the Avro schema for the given [serializer].
+ *
+ * @see Avro.schema
+ */
 public fun <T> Avro.schema(serializer: KSerializer<T>): Schema {
     return schema(serializer.descriptor)
 }
 
+/**
+ * Encodes the given [value] to a [ByteArray] using the provided [writerSchema].
+ *
+ * @see Avro.encodeToByteArray
+ */
 public inline fun <reified T> Avro.encodeToByteArray(
     writerSchema: Schema,
     value: T,
@@ -141,6 +292,14 @@ public inline fun <reified T> Avro.encodeToByteArray(
     return encodeToByteArray(writerSchema, serializer, value)
 }
 
+/**
+ * Decode the value from the given [bytes] to a value of type [T] assuming the data is respecting the [writerSchema].
+ * If during the decoding, the written data does not match [T]'s schema, you may end up to unexpected results or an exception.
+ *
+ * For any custom encoding or conversion of type [T] or its inner type, you may need to register a custom [kotlinx.serialization.KSerializer] in [Avro.serializersModule].
+ *
+ * @see Avro.decodeFromByteArray
+ */
 public inline fun <reified T> Avro.decodeFromByteArray(
     writerSchema: Schema,
     bytes: ByteArray,
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/AvroJVMExtensions.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/AvroJVMExtensions.kt
index 3a28f21..a8e57f5 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/AvroJVMExtensions.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/AvroJVMExtensions.kt
@@ -1,17 +1,16 @@
 package com.github.avrokotlin.avro4k
 
-import com.github.avrokotlin.avro4k.internal.decodeWithApacheDecoder
-import com.github.avrokotlin.avro4k.internal.encodeWithApacheEncoder
+import kotlinx.io.asSink
+import kotlinx.io.asSource
+import kotlinx.io.buffered
 import kotlinx.serialization.DeserializationStrategy
 import kotlinx.serialization.ExperimentalSerializationApi
 import kotlinx.serialization.SerializationStrategy
-import kotlinx.serialization.serializer
 import org.apache.avro.Schema
-import org.apache.avro.io.DecoderFactory
-import org.apache.avro.io.EncoderFactory
 import java.io.InputStream
 import java.io.OutputStream
 
+@Deprecated("Use encodeToSink instead", ReplaceWith("encodeToSink(writerSchema, serializer, value, outputStream.asSink().buffered())"))
 @ExperimentalSerializationApi
 public fun <T> Avro.encodeToStream(
     writerSchema: Schema,
@@ -19,50 +18,49 @@ public fun <T> Avro.encodeToStream(
     value: T,
     outputStream: OutputStream,
 ) {
-    val avroEncoder = EncoderFactory.get().binaryEncoder(outputStream, null)
-    encodeWithApacheEncoder(writerSchema, serializer, value, avroEncoder)
-    avroEncoder.flush()
+    encodeToSink(writerSchema, serializer, value, outputStream.asSink().buffered())
 }
 
+@Deprecated("Use encodeToSink instead", replaceWith = ReplaceWith("encodeToSink(value, outputStream.asSink().buffered())"))
 @ExperimentalSerializationApi
 public inline fun <reified T> Avro.encodeToStream(
     value: T,
     outputStream: OutputStream,
 ) {
-    val serializer = serializersModule.serializer<T>()
-    encodeToStream(schema(serializer), serializer, value, outputStream)
+    encodeToSink(value, outputStream.asSink().buffered())
 }
 
+@Deprecated("Use encodeToSink instead", replaceWith = ReplaceWith("encodeToSink(writerSchema, value, outputStream.asSink().buffered())"))
 @ExperimentalSerializationApi
 public inline fun <reified T> Avro.encodeToStream(
     writerSchema: Schema,
     value: T,
     outputStream: OutputStream,
 ) {
-    val serializer = serializersModule.serializer<T>()
-    encodeToStream(writerSchema, serializer, value, outputStream)
+    encodeToSink(writerSchema, value, outputStream.asSink().buffered())
 }
 
+@Deprecated("Use decodeFromSource instead", replaceWith = ReplaceWith("decodeFromSource(deserializer, inputStream.asSource().buffered())"))
 @ExperimentalSerializationApi
 public fun <T> Avro.decodeFromStream(
     writerSchema: Schema,
     deserializer: DeserializationStrategy<T>,
     inputStream: InputStream,
 ): T {
-    return decodeWithApacheDecoder(writerSchema, deserializer, DecoderFactory.get().binaryDecoder(inputStream, null))
+    return decodeFromSource(writerSchema, deserializer, inputStream.asSource().buffered())
 }
 
+@Deprecated("Use decodeFromSource instead", replaceWith = ReplaceWith("decodeFromSource(inputStream.asSource().buffered())"))
 @ExperimentalSerializationApi
 public inline fun <reified T> Avro.decodeFromStream(inputStream: InputStream): T {
-    val serializer = serializersModule.serializer<T>()
-    return decodeFromStream(schema(serializer.descriptor), serializer, inputStream)
+    return decodeFromSource(inputStream.asSource().buffered())
 }
 
+@Deprecated("Use decodeFromSource instead", replaceWith = ReplaceWith("decodeFromSource(writerSchema, inputStream.asSource().buffered())"))
 @ExperimentalSerializationApi
 public inline fun <reified T> Avro.decodeFromStream(
     writerSchema: Schema,
     inputStream: InputStream,
 ): T {
-    val serializer = serializersModule.serializer<T>()
-    return decodeFromStream(writerSchema, serializer, inputStream)
+    return decodeFromSource(writerSchema, inputStream.asSource().buffered())
 }
\ No newline at end of file
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/AvroKotlinxIoExtensions.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/AvroKotlinxIoExtensions.kt
new file mode 100644
index 0000000..766b1c5
--- /dev/null
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/AvroKotlinxIoExtensions.kt
@@ -0,0 +1,159 @@
+package com.github.avrokotlin.avro4k
+
+import com.github.avrokotlin.avro4k.internal.decodeWithApacheDecoder
+import com.github.avrokotlin.avro4k.internal.decoder.direct.KotlinxIoDecoder
+import com.github.avrokotlin.avro4k.internal.encodeWithApacheEncoder
+import com.github.avrokotlin.avro4k.internal.encoder.direct.KotlinxIoEncoder
+import kotlinx.io.Sink
+import kotlinx.io.Source
+import kotlinx.serialization.DeserializationStrategy
+import kotlinx.serialization.SerializationStrategy
+import kotlinx.serialization.serializer
+import org.apache.avro.Schema
+
+/**
+ * Encode the given [value] as binary avro into the given [sink], respecting the [writerSchema].
+ *
+ * For any custom encoding or conversion of the [value] or its inner type, you may need to register a custom [kotlinx.serialization.KSerializer] in [Avro.serializersModule].
+ *
+ * Note that as the [sink] is buffered, you may need to call [Sink.flush] to ensure all data is written out.
+ * This method does not close the [sink], so you need to handle it to avoid resource leaks.
+ *
+ * @param writerSchema the schema to use for encoding the value which must be compatible with the given [serializer].
+ * @param serializer the serialization strategy to use for encoding the value. You may prefer the other extension methods without the serializer parameter for convenience.
+ * @param value the value to encode.
+ * @param sink the sink to write the encoded value to.
+ *
+ * @see Avro.encodeToByteArray
+ * @see kotlinx.io.asSink
+ * @see com.github.avrokotlin.avro4k.serializer.AvroSerializer
+ */
+public fun <T> Avro.encodeToSink(
+    writerSchema: Schema,
+    serializer: SerializationStrategy<T>,
+    value: T,
+    sink: Sink,
+) {
+    encodeWithApacheEncoder(writerSchema, serializer, value, KotlinxIoEncoder(sink))
+}
+
+/**
+ * Encode the given [value] as binary avro into the given [sink], using the [value]'s schema as the writer schema.
+ *
+ * For any custom schema, encoding or conversion of the [value] or its inner type, you may need to register a custom [kotlinx.serialization.KSerializer] in [Avro.serializersModule].
+ *
+ * Note that as the [sink] is buffered, you may need to call [Sink.flush] to ensure all data is written out.
+ * This method does not close the [sink], so you need to handle it to avoid resource leaks.
+ *
+ * @param value the value to encode.
+ * @param sink the sink to write the encoded value to.
+ *
+ * @see Avro.encodeToByteArray
+ * @see Avro.schema
+ * @see kotlinx.io.asSink
+ * @see kotlinx.serialization.modules.SerializersModule.serializer
+ * @see com.github.avrokotlin.avro4k.serializer.AvroSerializer
+ */
+public inline fun <reified T> Avro.encodeToSink(
+    value: T,
+    sink: Sink,
+) {
+    val serializer = serializersModule.serializer<T>()
+    encodeToSink(schema(serializer), serializer, value, sink)
+}
+
+/**
+ * Encode the given [value] as binary avro into the given [sink], respecting the [writerSchema].
+ *
+ * For any custom encoding or conversion of the [value] or its inner type, you may need to register a custom [kotlinx.serialization.KSerializer] in [Avro.serializersModule].
+ *
+ * Note that as the [sink] is buffered, you may need to call [Sink.flush] to ensure all data is written out.
+ * This method does not close the [sink], so you need to handle it to avoid resource leaks.
+ *
+ * @param writerSchema the schema to use for encoding the value which must be compatible with the [value].
+ * @param value the value to encode.
+ * @param sink the sink to write the encoded value to.
+ *
+ * @see Avro.encodeToByteArray
+ * @see kotlinx.io.asSink
+ * @see kotlinx.serialization.modules.SerializersModule.serializer
+ * @see com.github.avrokotlin.avro4k.serializer.AvroSerializer
+ */
+public inline fun <reified T> Avro.encodeToSink(
+    writerSchema: Schema,
+    value: T,
+    sink: Sink,
+) {
+    val serializer = serializersModule.serializer<T>()
+    encodeToSink(writerSchema, serializer, value, sink)
+}
+
+/**
+ * Decode the value from a [source] assuming the data is respecting the given [writerSchema].
+ * If during the decoding, the written data does not match the [writerSchema], you may end up to unexpected results or an exception.
+ *
+ * For any custom encoding or conversion of type [T] or its inner type, you may need to register a custom [kotlinx.serialization.KSerializer] in [Avro.serializersModule].
+ *
+ * @param writerSchema the schema to use for decoding the value.
+ * @param deserializer the deserialization strategy to use for decoding the value. You may prefer the other extension methods without the deserializer parameter for convenience.
+ * @param source the source to read the encoded value from.
+ * @return the decoded value of type [T].
+ *
+ * @see Avro.decodeFromByteArray
+ * @see kotlinx.io.asSink
+ * @see com.github.avrokotlin.avro4k.serializer.AvroSerializer
+ */
+public fun <T> Avro.decodeFromSource(
+    writerSchema: Schema,
+    deserializer: DeserializationStrategy<T>,
+    source: Source,
+): T {
+    return decodeWithApacheDecoder(
+        writerSchema,
+        deserializer,
+        KotlinxIoDecoder(source)
+    )
+}
+
+/**
+ * Decode the value from a [source] assuming the data is respecting the schema inferred from [T] using [Avro.schema].
+ * If during the decoding, the written data does not match [T]'s schema, you may end up to unexpected results or an exception.
+ *
+ * For any custom encoding or conversion of type [T] or its inner type, you may need to register a custom [kotlinx.serialization.KSerializer] in [Avro.serializersModule].
+ *
+ * @param source the source to read the encoded value from.
+ * @return the decoded value of type [T].
+ *
+ * @see Avro.decodeFromByteArray
+ * @see Avro.schema
+ * @see kotlinx.serialization.modules.SerializersModule.serializer
+ * @see kotlinx.io.asSink
+ * @see com.github.avrokotlin.avro4k.serializer.AvroSerializer
+ */
+public inline fun <reified T> Avro.decodeFromSource(source: Source): T {
+    val serializer = serializersModule.serializer<T>()
+    return decodeFromSource(schema(serializer.descriptor), serializer, source)
+}
+
+/**
+ * Decode the value from a [source] assuming the data is respecting the given [writerSchema].
+ * If during the decoding, the written data does not match the [writerSchema], you may end up to unexpected results or an exception.
+ *
+ * For any custom encoding or conversion of type [T] or its inner type, you may need to register a custom [kotlinx.serialization.KSerializer] in [Avro.serializersModule].
+ *
+ * @param writerSchema the schema to use for decoding the value.
+ * @param source the source to read the encoded value from.
+ * @return the decoded value of type [T].
+ *
+ * @see Avro.decodeFromByteArray
+ * @see kotlinx.serialization.modules.SerializersModule.serializer
+ * @see kotlinx.io.asSink
+ * @see com.github.avrokotlin.avro4k.serializer.AvroSerializer
+ */
+public inline fun <reified T> Avro.decodeFromSource(
+    writerSchema: Schema,
+    source: Source,
+): T {
+    val serializer = serializersModule.serializer<T>()
+    return decodeFromSource(writerSchema, serializer, source)
+}
\ No newline at end of file
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/AvroOkioExtensions.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/AvroOkioExtensions.kt
index d07c25f..e94a777 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/AvroOkioExtensions.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/AvroOkioExtensions.kt
@@ -12,6 +12,7 @@ import org.apache.avro.Schema
 import org.apache.avro.io.DecoderFactory
 import org.apache.avro.io.EncoderFactory
 
+@Deprecated("Use kotlinx.io's encodeToSink instead")
 @ExperimentalSerializationApi
 public fun <T> Avro.encodeToSink(
     writerSchema: Schema,
@@ -24,6 +25,7 @@ public fun <T> Avro.encodeToSink(
     binaryEncoder.flush()
 }
 
+@Deprecated("Use kotlinx.io's encodeToSink instead")
 @ExperimentalSerializationApi
 public inline fun <reified T> Avro.encodeToSink(
     value: T,
@@ -33,6 +35,7 @@ public inline fun <reified T> Avro.encodeToSink(
     encodeToSink(schema(serializer), serializer, value, sink)
 }
 
+@Deprecated("Use kotlinx.io's encodeToSink instead")
 @ExperimentalSerializationApi
 public inline fun <reified T> Avro.encodeToSink(
     writerSchema: Schema,
@@ -43,6 +46,7 @@ public inline fun <reified T> Avro.encodeToSink(
     encodeToSink(writerSchema, serializer, value, sink)
 }
 
+@Deprecated("Use kotlinx.io's decodeFromSource instead")
 @ExperimentalSerializationApi
 public fun <T> Avro.decodeFromSource(
     writerSchema: Schema,
@@ -56,12 +60,14 @@ public fun <T> Avro.decodeFromSource(
     )
 }
 
+@Deprecated("Use kotlinx.io's decodeFromSource instead")
 @ExperimentalSerializationApi
 public inline fun <reified T> Avro.decodeFromSource(source: BufferedSource): T {
     val serializer = serializersModule.serializer<T>()
     return decodeFromSource(schema(serializer.descriptor), serializer, source)
 }
 
+@Deprecated("Use kotlinx.io's decodeFromSource instead")
 @ExperimentalSerializationApi
 public inline fun <reified T> Avro.decodeFromSource(
     writerSchema: Schema,
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/AvroSingleObject.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/AvroSingleObject.kt
index 3ca113f..a6caa85 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/AvroSingleObject.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/AvroSingleObject.kt
@@ -1,5 +1,14 @@
 package com.github.avrokotlin.avro4k
 
+import com.github.avrokotlin.avro4k.internal.Buffer
+import kotlinx.io.Buffer
+import kotlinx.io.Sink
+import kotlinx.io.Source
+import kotlinx.io.asSink
+import kotlinx.io.asSource
+import kotlinx.io.buffered
+import kotlinx.io.readByteArray
+import kotlinx.io.readLongLe
 import kotlinx.serialization.BinaryFormat
 import kotlinx.serialization.DeserializationStrategy
 import kotlinx.serialization.ExperimentalSerializationApi
@@ -9,11 +18,8 @@ import kotlinx.serialization.modules.SerializersModule
 import kotlinx.serialization.serializer
 import org.apache.avro.Schema
 import org.apache.avro.SchemaNormalization
-import java.io.ByteArrayOutputStream
 import java.io.InputStream
 import java.io.OutputStream
-import java.nio.ByteBuffer
-import java.nio.ByteOrder
 
 /**
  * Single Avro objects are encoded as follows:
@@ -36,59 +42,73 @@ public class AvroSingleObject(
 
     private fun Schema.crc64avro(): ByteArray = SchemaNormalization.parsingFingerprint("CRC-64-AVRO", this)
 
+    @Deprecated("Use encodeToSink instead", ReplaceWith("encodeToSink(writerSchema, serializer, value, outputStream.asSink().buffered())"))
     public fun <T> encodeToStream(
         writerSchema: Schema,
         serializer: SerializationStrategy<T>,
         value: T,
         outputStream: OutputStream,
+    ): Unit = encodeToSink(writerSchema, serializer, value, outputStream.asSink().buffered())
+
+    public fun <T> encodeToSink(
+        writerSchema: Schema,
+        serializer: SerializationStrategy<T>,
+        value: T,
+        sink: Sink,
     ) {
-        outputStream.write(MAGIC_BYTE)
-        outputStream.write(FORMAT_VERSION)
-        outputStream.write(writerSchema.crc64avro())
-        avro.encodeToStream(writerSchema, serializer, value, outputStream)
+        sink.writeByte(MAGIC_BYTE)
+        sink.writeByte(FORMAT_VERSION)
+        sink.write(writerSchema.crc64avro())
+        avro.encodeToSink(writerSchema, serializer, value, sink)
     }
 
+    @Deprecated("Use decodeFromSource instead", ReplaceWith("decodeFromSource(deserializer, inputStream.asSource().buffered())"))
     public fun <T> decodeFromStream(
         deserializer: DeserializationStrategy<T>,
         inputStream: InputStream,
+    ): T = decodeFromSource(deserializer, inputStream.asSource().buffered())
+
+    public fun <T> decodeFromSource(
+        deserializer: DeserializationStrategy<T>,
+        source: Source,
     ): T {
-        check(inputStream.read() == MAGIC_BYTE) { "Not a valid single-object avro format, bad magic byte" }
-        check(inputStream.read() == FORMAT_VERSION) { "Not a valid single-object avro format, bad version byte" }
-        val fingerprint = ByteBuffer.wrap(ByteArray(8).apply { inputStream.read(this) }).order(ByteOrder.LITTLE_ENDIAN).getLong()
+        check(source.readByte() == MAGIC_BYTE) { "Not a valid single-object avro format, bad magic byte" }
+        check(source.readByte() == FORMAT_VERSION) { "Not a valid single-object avro format, bad version byte" }
+        val fingerprint = source.readLongLe()
         val writerSchema =
             schemaRegistry(fingerprint) ?: throw SerializationException("schema not found for the given object's schema fingerprint 0x${fingerprint.toString(16)}")
 
-        return avro.decodeFromStream(writerSchema, deserializer, inputStream)
+        return avro.decodeFromSource(writerSchema, deserializer, source)
     }
 
     public override fun <T> decodeFromByteArray(
         deserializer: DeserializationStrategy<T>,
         bytes: ByteArray,
     ): T {
-        return bytes.inputStream().use {
-            decodeFromStream(deserializer, it)
-        }
+        return decodeFromSource(deserializer, Buffer(bytes))
     }
 
     override fun <T> encodeToByteArray(
         serializer: SerializationStrategy<T>,
         value: T,
     ): ByteArray {
-        return encodeToByteArray(avro.schema(serializer.descriptor), serializer, value)
+        val writerSchema = avro.schema(serializer.descriptor)
+        return encodeToByteArray(writerSchema, serializer, value)
     }
 }
 
-private const val MAGIC_BYTE: Int = 0xC3
-private const val FORMAT_VERSION: Int = 1
+private const val MAGIC_BYTE: Byte = 0xC3.toByte()
+private const val FORMAT_VERSION: Byte = 1
 
 public fun <T> AvroSingleObject.encodeToByteArray(
     writerSchema: Schema,
     serializer: SerializationStrategy<T>,
     value: T,
-): ByteArray =
-    ByteArrayOutputStream().apply {
-        encodeToStream(writerSchema, serializer, value, this)
-    }.toByteArray()
+): ByteArray {
+    val sink = Buffer()
+    encodeToSink(writerSchema, serializer, value, sink)
+    return sink.readByteArray()
+}
 
 public inline fun <reified T> AvroSingleObject.encodeToByteArray(
     writerSchema: Schema,
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/internal/Samples.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/Samples.kt
new file mode 100644
index 0000000..962c001
--- /dev/null
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/Samples.kt
@@ -0,0 +1,56 @@
+package com.github.avrokotlin.avro4k.internal
+
+import com.github.avrokotlin.avro4k.Avro
+import com.github.avrokotlin.avro4k.FieldNamingStrategy
+import kotlinx.serialization.decodeFromByteArray
+import kotlinx.serialization.encodeToByteArray
+import kotlinx.serialization.encoding.Decoder
+import kotlinx.serialization.encoding.Encoder
+import kotlinx.serialization.modules.SerializersModule
+import kotlinx.serialization.modules.polymorphic
+
+@Suppress("KotlinUnreachableCode")
+internal object Samples {
+    fun customizeAvroInstance() {
+        val yourAvroCustomizedInstance =
+            Avro {
+                fieldNamingStrategy = FieldNamingStrategy.Builtins.SnakeCase
+                implicitNulls = false
+                implicitEmptyCollections = false
+                validateSerialization = true
+                serializersModule =
+                    SerializersModule {
+                        // Register your custom serializers here
+                        contextual(YourType::class, YourTypeSerializer())
+                        // Set the possible implementations of an abstract class or an interface
+                        polymorphic(YourParentType::class) {
+                            subclass(YourSubType::class, YourSubTypeSerializer())
+                        }
+                    }
+            }
+        yourAvroCustomizedInstance.encodeToByteArray(YourType())
+        yourAvroCustomizedInstance.decodeFromByteArray<YourParentType>(TODO("byteArray"))
+    }
+
+    class YourType
+
+    class YourTypeSerializer : kotlinx.serialization.KSerializer<YourType> {
+        override val descriptor get() = TODO("Not yet implemented")
+
+        override fun serialize(encoder: Encoder, value: YourType) = TODO("Not yet implemented")
+
+        override fun deserialize(decoder: Decoder) = TODO("Not yet implemented")
+    }
+
+    interface YourParentType
+
+    object YourSubType : YourParentType
+
+    class YourSubTypeSerializer : kotlinx.serialization.KSerializer<YourSubType> {
+        override val descriptor get() = TODO("Not yet implemented")
+
+        override fun serialize(encoder: Encoder, value: YourSubType) = TODO("Not yet implemented")
+
+        override fun deserialize(decoder: Decoder) = TODO("Not yet implemented")
+    }
+}
\ No newline at end of file
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/internal/decoder/direct/AbstractAvroDirectDecoder.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/decoder/direct/AbstractAvroDirectDecoder.kt
index 5e508ba..9243032 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/internal/decoder/direct/AbstractAvroDirectDecoder.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/decoder/direct/AbstractAvroDirectDecoder.kt
@@ -56,6 +56,7 @@ internal abstract class AbstractAvroDirectDecoder(
                             avro,
                             binaryDecoder
                         )
+
                     else -> throw unsupportedWriterTypeError(Schema.Type.ARRAY)
                 }
 
@@ -69,6 +70,7 @@ internal abstract class AbstractAvroDirectDecoder(
                             avro,
                             binaryDecoder
                         )
+
                     else -> throw unsupportedWriterTypeError(Schema.Type.MAP)
                 }
 
@@ -264,7 +266,12 @@ private fun org.apache.avro.io.Decoder.readFixedBytes(size: Int): ByteArray {
 }
 
 private fun org.apache.avro.io.Decoder.readBytes(): ByteArray {
-    return readBytes(null).array()
+    val buffer = readBytes(null)
+    if (buffer.hasArray()) {
+        return buffer.array()
+    }
+    return ByteArray(buffer.remaining())
+        .apply { buffer.get(this) }
 }
 
 private class PolymorphicDecoder(
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/internal/decoder/direct/KotlinxIoDecoder.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/decoder/direct/KotlinxIoDecoder.kt
new file mode 100644
index 0000000..1f9929f
--- /dev/null
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/decoder/direct/KotlinxIoDecoder.kt
@@ -0,0 +1,291 @@
+package com.github.avrokotlin.avro4k.internal.decoder.direct
+
+import kotlinx.io.InternalIoApi
+import kotlinx.io.Source
+import kotlinx.io.UnsafeIoApi
+import kotlinx.io.readAtMostTo
+import kotlinx.io.readByteArray
+import kotlinx.io.readDoubleLe
+import kotlinx.io.readFloatLe
+import kotlinx.io.readString
+import kotlinx.io.readTo
+import kotlinx.io.unsafe.UnsafeBufferOperations
+import kotlinx.io.unsafe.withData
+import kotlinx.serialization.SerializationException
+import org.apache.avro.SystemLimitException
+import org.apache.avro.io.Decoder
+import org.apache.avro.util.Utf8
+import java.nio.ByteBuffer
+
+internal class KotlinxIoDecoder(
+    private val source: Source,
+) : Decoder() {
+    override fun readNull() {
+    }
+
+    override fun readBoolean(): Boolean {
+        return source.readByte() == 1.toByte()
+    }
+
+    override fun readInt(): Int {
+        return source.readVarInt()
+    }
+
+    override fun readLong(): Long {
+        return source.readVarLong()
+    }
+
+    override fun readFloat(): Float {
+        return source.readFloatLe()
+    }
+
+    override fun readDouble(): Double {
+        return source.readDoubleLe()
+    }
+
+    override fun readString(old: Utf8?): Utf8 {
+        return Utf8(source.readByteArray(SystemLimitException.checkMaxStringLength(readLong())))
+    }
+
+    override fun readString(): String {
+        return source.readString(readLong())
+    }
+
+    override fun skipString() {
+        source.skip(readLong())
+    }
+
+    @OptIn(InternalIoApi::class, UnsafeIoApi::class)
+    override fun readBytes(old: ByteBuffer?): ByteBuffer {
+        val length = SystemLimitException.checkMaxBytesLength(readLong())
+        source.require(length.toLong())
+        if (old != null && length <= old.capacity()) {
+            old.clear().limit(length)
+            source.readAtMostTo(old)
+            return old
+        } else {
+            UnsafeBufferOperations.forEachSegment(source.buffer) { ctx, segment ->
+                if (length >= segment.size) {
+                    // the bytes can be split across segments.
+                    val buffer = ByteBuffer.allocate(length)
+                    source.readAtMostTo(buffer)
+                    return buffer
+                }
+                ctx.withData(segment) { bytes, offset, _ ->
+                    // IMPORTANT: don't wrap the buffer when length == segment.size
+                    //  as the segment will be recycled, so the backing array content can change
+                    source.skip(length.toLong())
+                    return ByteBuffer.wrap(bytes, offset, length).asReadOnlyBuffer()
+                }
+            }
+        }
+        error("Unreachable")
+    }
+
+    override fun skipBytes() {
+        source.skip(readLong())
+    }
+
+    override fun readFixed(bytes: ByteArray) {
+        source.readTo(bytes)
+    }
+
+    override fun readFixed(
+        bytes: ByteArray,
+        start: Int,
+        length: Int,
+    ) {
+        source.readTo(bytes, startIndex = start, endIndex = start + length)
+    }
+
+    override fun skipFixed(length: Int) {
+        source.skip(length.toLong())
+    }
+
+    override fun readEnum(): Int {
+        return readInt()
+    }
+
+    override fun readArrayStart(): Long {
+        return doReadItemCount()
+    }
+
+    override fun readMapStart(): Long {
+        return doReadItemCount()
+    }
+
+    override fun arrayNext(): Long {
+        return doReadItemCount()
+    }
+
+    override fun mapNext(): Long {
+        return doReadItemCount()
+    }
+
+    override fun skipArray(): Long {
+        return doSkipItems()
+    }
+
+    override fun skipMap(): Long {
+        return doSkipItems()
+    }
+
+    private fun doReadItemCount(): Long {
+        var result = readLong()
+        if (result < 0L) {
+            readLong()
+            result = -result
+        }
+        return result
+    }
+
+    private fun doSkipItems(): Long {
+        var result = readLong()
+        while (result < 0L) {
+            source.skip(readLong())
+            result = readLong()
+        }
+        return result
+    }
+
+    override fun readIndex(): Int {
+        return readInt()
+    }
+}
+
+private const val MAX_VARINT_INT_BYTES = 5
+
+@OptIn(InternalIoApi::class, UnsafeIoApi::class)
+private fun Source.readVarInt(): Int {
+    // This ensures having at least one byte available to read
+    // There are 3 outcomes:
+    // 1. There are already all the bytes of the varint available in the buffer: fast mode
+    // 2. There are no bytes available in the buffer, so it triggers a read of a full segment, retrieving much more than 1 byte if more available, finally reading the varint in fast mode
+    // 3. There is only one byte available in the buffer, while the varint has 2 bytes across segments, so the read will be done in slow mode
+    require(1.toLong())
+
+    // We do not really iterate, we just want to peek the first segment
+    UnsafeBufferOperations.forEachSegment(buffer) { ctx, segment ->
+        // TODO: start in fast mode, finish in slow mode if the varint is split across segments
+        if (segment.size < MAX_VARINT_INT_BYTES) {
+            // Slow mode: the int may be split across segments
+            return decodeVarInt { readByte() }
+        }
+        ctx.withData(segment) { buf, offset, _ ->
+            // Fast mode: we have all the bytes in the current segment
+            var len = 0
+            val result = decodeVarInt { buf[offset + len++] }
+            skip(len.toLong())
+            return result
+        }
+    }
+    error("Unreachable")
+}
+
+private const val MAX_VARINT_LONG_BYTES = 10
+
+@OptIn(InternalIoApi::class, UnsafeIoApi::class)
+private fun Source.readVarLong(): Long {
+    request(1.toLong())
+    // We do not really iterate, we just want to peek the first segment
+    UnsafeBufferOperations.forEachSegment(buffer) { ctx, segment ->
+        if (segment.size < MAX_VARINT_LONG_BYTES) {
+            // the long can be split across segments
+            return decodeVarLong { readByte() }
+        }
+        ctx.withData(segment) { bytes, offset, _ ->
+            var len = 0
+            val result = decodeVarLong { bytes[offset + len++] }
+            skip(len.toLong())
+            return result
+        }
+    }
+    error("Unreachable")
+}
+
+private inline fun decodeVarInt(readByte: () -> Byte): Int {
+    var b = readByte().toInt()
+    var varint = b and 0x7F
+    if (b < 0) {
+        b = readByte().toInt()
+        varint += ((b and 0x7F) shl 7)
+        if (b < 0) {
+            b = readByte().toInt()
+            varint += ((b and 0x7F) shl 14)
+            if (b < 0) {
+                b = readByte().toInt()
+                varint += ((b and 0x7F) shl 21)
+                if (b < 0) {
+                    b = readByte().toInt()
+                    if (b < 0) {
+                        throw SerializationException("Invalid varint encoding: more bytes than expected")
+                    }
+                    varint += (b shl 28)
+                }
+            }
+        }
+    }
+    return decodeZigZag(varint)
+}
+
+private inline fun decodeVarLong(readByte: () -> Byte): Long {
+    var b: Int = readByte().toInt()
+    var i = b and 0x7F
+    if (b < 0) {
+        b = readByte().toInt()
+        i += ((b and 0x7F) shl 7)
+        if (b < 0) {
+            b = readByte().toInt()
+            i += ((b and 0x7F) shl 14)
+            if (b < 0) {
+                b = readByte().toInt()
+                i += ((b and 0x7F) shl 21)
+                if (b < 0) {
+                    return decodeVarLong2(i.toLong(), readByte)
+                }
+            }
+        }
+    }
+    return decodeZigZag(i).toLong()
+}
+
+private inline fun decodeVarLong2(lo: Long, readByte: () -> Byte): Long {
+    // then next 28 bits (altogether 8 bytes)
+    var lo = lo
+    var b: Int = readByte().toInt()
+    var i = b and 0x7F
+    if (b < 0) {
+        i = i and 0x7F
+        b = readByte().toInt()
+        i += ((b and 0x7F) shl 7)
+        if (b < 0) {
+            b = readByte().toInt()
+            i += ((b and 0x7F) shl 14)
+            if (b < 0) {
+                b = readByte().toInt()
+                i += ((b and 0x7F) shl 21)
+                if (b < 0) {
+                    // Ok 56-bits gone... still going strong!
+                    lo = lo or ((i.toLong()) shl 28)
+                    b = readByte().toInt()
+                    i = b and 0x7F
+                    if (b < 0) {
+                        b = readByte().toInt()
+                        if (b < 0) {
+                            throw SerializationException("Invalid varlong encoding: more bytes than expected")
+                        }
+                        i = i or (b shl 7)
+                    }
+                    lo = lo or ((i.toLong()) shl 56)
+                    return decodeZigZag(lo)
+                }
+            }
+        }
+    }
+    lo = lo or ((i.toLong()) shl 28)
+    return decodeZigZag(lo)
+}
+
+private fun decodeZigZag(varint: Int): Int = (varint ushr 1) xor (-(varint and 1))
+
+private fun decodeZigZag(varint: Long): Long = (varint ushr 1) xor (-(varint and 1))
\ No newline at end of file
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/internal/decoder/direct/RecordDirectDecoder.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/decoder/direct/RecordDirectDecoder.kt
index e6c6d70..47ac9bf 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/internal/decoder/direct/RecordDirectDecoder.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/decoder/direct/RecordDirectDecoder.kt
@@ -18,7 +18,7 @@ internal class RecordDirectDecoder(
     private val writerRecordSchema: Schema,
     descriptor: SerialDescriptor,
     avro: Avro,
-    binaryDecoder: org.apache.avro.io.Decoder,
+    binaryDecoder: Decoder,
 ) : AbstractAvroDirectDecoder(avro, binaryDecoder) {
     // from descriptor element index to schema field. The missing fields are at the end to decode the default values
     private val classDescriptor = avro.recordResolver.resolveFields(writerRecordSchema, descriptor)
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/internal/encoder/AbstractAvroEncoder.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/encoder/AbstractAvroEncoder.kt
index ebd0ddb..65d43da 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/internal/encoder/AbstractAvroEncoder.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/encoder/AbstractAvroEncoder.kt
@@ -63,6 +63,8 @@ internal abstract class AbstractAvroEncoder : AbstractEncoder(), AvroEncoder {
 
     abstract fun encodeStringUnchecked(value: Utf8)
 
+    abstract fun encodeStringUnchecked(value: String)
+
     abstract fun encodeBytesUnchecked(value: ByteArray)
 
     abstract fun encodeFixedUnchecked(value: ByteArray)
@@ -225,7 +227,7 @@ internal abstract class AbstractAvroEncoder : AbstractEncoder(), AvroEncoder {
                     throw fullNameOrAliasMismatchError(enumDescriptor.nonNullSerialName, enumDescriptor.aliases)
                 }
 
-            Schema.Type.STRING -> encodeStringUnchecked(Utf8(enumName))
+            Schema.Type.STRING -> encodeStringUnchecked(enumName)
             else -> throw unsupportedWriterTypeError(Schema.Type.ENUM, Schema.Type.STRING)
         }
     }
@@ -237,7 +239,7 @@ internal abstract class AbstractAvroEncoder : AbstractEncoder(), AvroEncoder {
         }
         when (currentWriterSchema.type) {
             Schema.Type.BOOLEAN -> encodeBooleanUnchecked(value)
-            Schema.Type.STRING -> encodeStringUnchecked(Utf8(value.toString()))
+            Schema.Type.STRING -> encodeStringUnchecked(value.toString())
             else -> throw unsupportedWriterTypeError(Schema.Type.BOOLEAN, Schema.Type.STRING)
         }
     }
@@ -258,7 +260,7 @@ internal abstract class AbstractAvroEncoder : AbstractEncoder(), AvroEncoder {
         when (currentWriterSchema.type) {
             Schema.Type.INT -> encodeIntUnchecked(value)
             Schema.Type.LONG -> encodeLongUnchecked(value.toLong())
-            Schema.Type.STRING -> encodeStringUnchecked(Utf8(value.toString()))
+            Schema.Type.STRING -> encodeStringUnchecked(value.toString())
             else -> throw unsupportedWriterTypeError(Schema.Type.INT, Schema.Type.LONG, Schema.Type.STRING)
         }
     }
@@ -271,7 +273,7 @@ internal abstract class AbstractAvroEncoder : AbstractEncoder(), AvroEncoder {
         when (currentWriterSchema.type) {
             Schema.Type.INT -> encodeIntUnchecked(value.toIntExact())
             Schema.Type.LONG -> encodeLongUnchecked(value)
-            Schema.Type.STRING -> encodeStringUnchecked(Utf8(value.toString()))
+            Schema.Type.STRING -> encodeStringUnchecked(value.toString())
             else -> throw unsupportedWriterTypeError(Schema.Type.INT, Schema.Type.LONG, Schema.Type.STRING)
         }
     }
@@ -284,7 +286,7 @@ internal abstract class AbstractAvroEncoder : AbstractEncoder(), AvroEncoder {
         when (currentWriterSchema.type) {
             Schema.Type.FLOAT -> encodeFloatUnchecked(value)
             Schema.Type.DOUBLE -> encodeDoubleUnchecked(value.toDouble())
-            Schema.Type.STRING -> encodeStringUnchecked(Utf8(value.toString()))
+            Schema.Type.STRING -> encodeStringUnchecked(value.toString())
             else -> throw unsupportedWriterTypeError(Schema.Type.FLOAT, Schema.Type.DOUBLE, Schema.Type.STRING)
         }
     }
@@ -297,7 +299,7 @@ internal abstract class AbstractAvroEncoder : AbstractEncoder(), AvroEncoder {
         when (currentWriterSchema.type) {
             Schema.Type.FLOAT -> encodeFloatUnchecked(value.toFloatExact())
             Schema.Type.DOUBLE -> encodeDoubleUnchecked(value)
-            Schema.Type.STRING -> encodeStringUnchecked(Utf8(value.toString()))
+            Schema.Type.STRING -> encodeStringUnchecked(value.toString())
             else -> throw unsupportedWriterTypeError(Schema.Type.DOUBLE, Schema.Type.FLOAT, Schema.Type.STRING)
         }
     }
@@ -309,7 +311,7 @@ internal abstract class AbstractAvroEncoder : AbstractEncoder(), AvroEncoder {
         }
         when (currentWriterSchema.type) {
             Schema.Type.INT -> encodeIntUnchecked(value.code)
-            Schema.Type.STRING -> encodeStringUnchecked(Utf8(value.toString()))
+            Schema.Type.STRING -> encodeStringUnchecked(value.toString())
             else -> throw unsupportedWriterTypeError(Schema.Type.INT, Schema.Type.STRING)
         }
     }
@@ -344,7 +346,7 @@ internal abstract class AbstractAvroEncoder : AbstractEncoder(), AvroEncoder {
             Schema.Type.LONG -> encodeLongUnchecked(value.toLong())
             Schema.Type.FLOAT -> encodeFloatUnchecked(value.toFloat())
             Schema.Type.DOUBLE -> encodeDoubleUnchecked(value.toDouble())
-            Schema.Type.STRING -> encodeStringUnchecked(Utf8(value))
+            Schema.Type.STRING -> encodeStringUnchecked(value)
             Schema.Type.BYTES -> encodeBytesUnchecked(value.encodeToByteArray())
             Schema.Type.FIXED -> encodeFixedUnchecked(ensureFixedSize(value.encodeToByteArray()))
             Schema.Type.ENUM -> encodeEnumUnchecked(value)
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/internal/encoder/direct/AbstractAvroDirectEncoder.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/encoder/direct/AbstractAvroDirectEncoder.kt
index 57e42aa..b556cc5 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/internal/encoder/direct/AbstractAvroDirectEncoder.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/encoder/direct/AbstractAvroDirectEncoder.kt
@@ -82,6 +82,10 @@ internal sealed class AbstractAvroDirectEncoder(
         binaryEncoder.writeString(value)
     }
 
+    override fun encodeStringUnchecked(value: String) {
+        binaryEncoder.writeString(value)
+    }
+
     override fun encodeEnumUnchecked(symbol: String) {
         binaryEncoder.writeEnum(currentWriterSchema.getEnumOrdinalChecked(symbol))
     }
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/internal/encoder/direct/KotlinxIoEncoder.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/encoder/direct/KotlinxIoEncoder.kt
new file mode 100644
index 0000000..20d1712
--- /dev/null
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/encoder/direct/KotlinxIoEncoder.kt
@@ -0,0 +1,74 @@
+package com.github.avrokotlin.avro4k.internal.encoder.direct
+
+import com.github.avrokotlin.avro4k.internal.toIntExact
+import kotlinx.io.DelicateIoApi
+import kotlinx.io.InternalIoApi
+import kotlinx.io.Sink
+import kotlinx.io.UnsafeIoApi
+import kotlinx.io.unsafe.UnsafeBufferOperations
+import kotlinx.io.writeDoubleLe
+import kotlinx.io.writeFloatLe
+import kotlinx.io.writeString
+import kotlinx.io.writeToInternalBuffer
+import org.apache.avro.io.BinaryData
+import org.apache.avro.io.BinaryEncoder
+
+internal class KotlinxIoEncoder(
+    private val sink: Sink,
+) : BinaryEncoder() {
+    override fun flush() {
+        sink.flush()
+    }
+
+    override fun writeBoolean(b: Boolean) {
+        sink.writeByte(if (b) 1 else 0)
+    }
+
+    @OptIn(DelicateIoApi::class, UnsafeIoApi::class)
+    override fun writeInt(n: Int) {
+        sink.writeToInternalBuffer { buffer ->
+            UnsafeBufferOperations.writeToTail(buffer, 5) { bytes, offset, _ ->
+                BinaryData.encodeInt(n, bytes, offset)
+            }
+        }
+    }
+
+    @OptIn(DelicateIoApi::class, UnsafeIoApi::class)
+    override fun writeLong(n: Long) {
+        sink.writeToInternalBuffer { buffer ->
+            UnsafeBufferOperations.writeToTail(buffer, 10) { bytes, offset, _ ->
+                BinaryData.encodeLong(n, bytes, offset)
+            }
+        }
+    }
+
+    override fun writeString(string: String) {
+        if (string.isEmpty()) {
+            writeZero()
+            return
+        }
+        writeInt(string.length)
+        sink.writeString(string)
+    }
+
+    override fun writeFloat(f: Float) {
+        sink.writeFloatLe(f)
+    }
+
+    override fun writeDouble(d: Double) {
+        sink.writeDoubleLe(d)
+    }
+
+    override fun writeFixed(bytes: ByteArray, start: Int, len: Int) {
+        sink.write(bytes, start, len)
+    }
+
+    override fun writeZero() {
+        sink.writeByte(0)
+    }
+
+    @OptIn(InternalIoApi::class)
+    override fun bytesBuffered(): Int {
+        return sink.buffer.size.toIntExact()
+    }
+}
\ No newline at end of file
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/internal/encoder/generic/AbstractAvroGenericEncoder.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/encoder/generic/AbstractAvroGenericEncoder.kt
index a10eab4..70263f6 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/internal/encoder/generic/AbstractAvroGenericEncoder.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/encoder/generic/AbstractAvroGenericEncoder.kt
@@ -51,6 +51,10 @@ internal abstract class AbstractAvroGenericEncoder : AbstractAvroEncoder() {
         encodeValue(value)
     }
 
+    override fun encodeStringUnchecked(value: String) {
+        encodeValue(Utf8(value))
+    }
+
     override fun encodeUnionIndexInternal(index: Int) {
         // nothing to do
     }
diff --git a/src/main/kotlin/com/github/avrokotlin/avro4k/internal/helpers.kt b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/helpers.kt
index 02f17b1..8d187d5 100644
--- a/src/main/kotlin/com/github/avrokotlin/avro4k/internal/helpers.kt
+++ b/src/main/kotlin/com/github/avrokotlin/avro4k/internal/helpers.kt
@@ -5,6 +5,9 @@ import com.fasterxml.jackson.databind.ObjectMapper
 import com.fasterxml.jackson.databind.node.TextNode
 import com.github.avrokotlin.avro4k.AvroAlias
 import com.github.avrokotlin.avro4k.AvroProp
+import kotlinx.io.Buffer
+import kotlinx.io.UnsafeIoApi
+import kotlinx.io.unsafe.UnsafeBufferOperations
 import kotlinx.serialization.ExperimentalSerializationApi
 import kotlinx.serialization.InternalSerializationApi
 import kotlinx.serialization.descriptors.PolymorphicKind
@@ -183,4 +186,8 @@ internal val AvroProp.jsonNode: JsonNode
             return objectMapper.readTree(value)
         }
         return TextNode.valueOf(value)
-    }
\ No newline at end of file
+    }
+
+@OptIn(UnsafeIoApi::class)
+internal fun Buffer(bytes: ByteArray): Buffer =
+    Buffer().apply { UnsafeBufferOperations.moveToTail(this, bytes) }
\ No newline at end of file


Test patch:
diff --git a/src/test/kotlin/com/github/avrokotlin/avro4k/encoding/RecordEncodingTest.kt b/src/test/kotlin/com/github/avrokotlin/avro4k/encoding/RecordEncodingTest.kt
index 79741e7..3b13153 100644
--- a/src/test/kotlin/com/github/avrokotlin/avro4k/encoding/RecordEncodingTest.kt
+++ b/src/test/kotlin/com/github/avrokotlin/avro4k/encoding/RecordEncodingTest.kt
@@ -2,19 +2,61 @@ package com.github.avrokotlin.avro4k.encoding
 
 import com.github.avrokotlin.avro4k.Avro
 import com.github.avrokotlin.avro4k.AvroAssertions
+import com.github.avrokotlin.avro4k.decodeFromSource
 import com.github.avrokotlin.avro4k.encodeToByteArray
 import com.github.avrokotlin.avro4k.internal.nullable
 import com.github.avrokotlin.avro4k.record
 import com.github.avrokotlin.avro4k.schema
 import io.kotest.assertions.throwables.shouldThrow
 import io.kotest.core.spec.style.StringSpec
+import kotlinx.io.asSource
+import kotlinx.io.buffered
 import kotlinx.serialization.SerialName
 import kotlinx.serialization.Serializable
 import kotlinx.serialization.SerializationException
+import kotlinx.serialization.encodeToByteArray
 import org.apache.avro.Schema
 import org.apache.avro.SchemaBuilder
+import java.io.OutputStream
+import java.io.PipedInputStream
+import java.io.PipedOutputStream
+import java.time.Instant
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.TimeUnit
 
 internal class RecordEncodingTest : StringSpec({
+    "Reading from an input stream is not waiting for the buffer to be full before decoding" {
+        @Serializable
+        data class DataS(
+            val id: Int,
+            val timestamp: Long,
+        )
+
+        val latch = CountDownLatch(1)
+        val latch2 = CountDownLatch(1)
+
+        class ProduceTask(private val out: OutputStream) : Thread() {
+            override fun run() {
+                out.write(Avro.encodeToByteArray(DataS(1, Instant.now().toEpochMilli())))
+                out.flush()
+                latch.countDown()
+                latch2.await(2, TimeUnit.SECONDS)
+            }
+        }
+
+        val inStream = PipedInputStream()
+        val outStream = PipedOutputStream(inStream)
+
+        val produceTask = ProduceTask(outStream)
+        produceTask.start()
+        latch.await(1, TimeUnit.SECONDS)
+        val source = inStream.asSource().buffered()
+
+        // It should throw an exception if the data is not available yet
+        // Thanks to kotlinx.io, we can read less than the buffer size, so the data is immediately available for small chunks
+        Avro.decodeFromSource<DataS>(source)
+        latch2.countDown()
+    }
     "encoding basic data class" {
         val input =
             Foo(