Repository: avro-kotlin/avro4k
Pull request #75 (2020-10-19 18:33:59)

Reworked Avro encoding and decoding API
Reworked Avro encoding and decoding API in order to have type safety for required encoding / decoding settings.

The following code would perfectly compile:
```kotlin
val input = openInputStream<T>(serializer) {
    format = AvroFormat.BinaryFormat
}
```

but at runtime an exception will be thrown because the readerSchema needs to be set for BinaryFormat.

To prevent this wrong API usage, the API has been redesigned so that the compiler can detect these errors. The call would look like this with the new API:
```kotlin
val input = openInputStream<T>(serializer) {
   decodeFormat = AvroDecodeFormat.Binary(mySchema)
}


Patch:
diff --git a/buildSrc/src/main/kotlin/Ci.kt b/buildSrc/src/main/kotlin/Ci.kt
index 4256be5..f771f7e 100644
--- a/buildSrc/src/main/kotlin/Ci.kt
+++ b/buildSrc/src/main/kotlin/Ci.kt
@@ -2,7 +2,7 @@ object Ci {
 
    // this is the version used for building snapshots
    // .buildnumber-snapshot will be appended
-   private const val snapshotBase = "0.41.0"
+   private const val snapshotBase = "0.50.0"
 
    private val githubBuildNumber = System.getenv("GITHUB_RUN_NUMBER")
 
diff --git a/src/main/kotlin/com/sksamuel/avro4k/Avro.kt b/src/main/kotlin/com/sksamuel/avro4k/Avro.kt
index 351383a..1a66ce4 100644
--- a/src/main/kotlin/com/sksamuel/avro4k/Avro.kt
+++ b/src/main/kotlin/com/sksamuel/avro4k/Avro.kt
@@ -2,14 +2,10 @@ package com.sksamuel.avro4k
 
 import com.sksamuel.avro4k.decoder.RootRecordDecoder
 import com.sksamuel.avro4k.encoder.RootRecordEncoder
-import com.sksamuel.avro4k.io.AvroBinaryInputStream
-import com.sksamuel.avro4k.io.AvroBinaryOutputStream
-import com.sksamuel.avro4k.io.AvroDataInputStream
-import com.sksamuel.avro4k.io.AvroDataOutputStream
+import com.sksamuel.avro4k.io.AvroDecodeFormat
+import com.sksamuel.avro4k.io.AvroEncodeFormat
 import com.sksamuel.avro4k.io.AvroFormat
 import com.sksamuel.avro4k.io.AvroInputStream
-import com.sksamuel.avro4k.io.AvroJsonInputStream
-import com.sksamuel.avro4k.io.AvroJsonOutputStream
 import com.sksamuel.avro4k.io.AvroOutputStream
 import com.sksamuel.avro4k.schema.DefaultNamingStrategy
 import com.sksamuel.avro4k.schema.schemaFor
@@ -18,8 +14,8 @@ import kotlinx.serialization.BinaryFormat
 import kotlinx.serialization.DeserializationStrategy
 import kotlinx.serialization.ExperimentalSerializationApi
 import kotlinx.serialization.SerialFormat
-import kotlinx.serialization.SerializationException
 import kotlinx.serialization.SerializationStrategy
+import kotlinx.serialization.descriptors.SerialDescriptor
 import kotlinx.serialization.modules.EmptySerializersModule
 import kotlinx.serialization.modules.SerializersModule
 import org.apache.avro.Schema
@@ -36,16 +32,41 @@ import java.nio.file.Path
 import java.nio.file.Paths
 import java.util.*
 
-data class AvroConfiguration constructor(
-   @JvmField internal val encodeDefaults: Boolean = true
-)
-
-class AvroInputStreamBuilder<T>(private val converter: (Any) -> T) {
-
+open class AvroInputStreamBuilder<T>(
+   private val converter: (Any) -> T
+   ) {
+   /**
+    * The format that should be used to decode the bytes from the input stream.
+    */
+   var decodeFormat : AvroDecodeFormat = defaultDecodeFormat
+   @Deprecated("please use decodeFormat to specify the format")
+   @Suppress("DEPRECATION")
    var format: AvroFormat = AvroFormat.DataFormat
+   @Deprecated("please use decodeFormat to specify the format")
    var writerSchema: Schema? = null
+   @Deprecated("please use decodeFormat to specify the format")
    var readerSchema: Schema? = null
 
+   companion object {
+      val defaultDecodeFormat = AvroDecodeFormat.Data(null, null)
+   }
+
+   @Suppress("DEPRECATION")
+   private val derivedDecodeFormat : AvroDecodeFormat
+      get() {
+         return when(format){
+            is AvroFormat.JsonFormat -> {
+               val wschema = writerSchema ?: throw IllegalArgumentException("Writer schema needs to be supplied for Json format")
+               AvroDecodeFormat.Json(wschema, readerSchema?:wschema)
+            }
+            is AvroFormat.BinaryFormat -> {
+               val wschema = writerSchema ?: throw IllegalArgumentException("Writer schema needs to be supplied for Binary format")
+               AvroDecodeFormat.Binary(wschema, readerSchema?:wschema)
+            }
+            is AvroFormat.DataFormat ->  AvroDecodeFormat.Data(writerSchema, readerSchema)
+         }
+      }
+
    fun from(path: Path): AvroInputStream<T> = from(Files.newInputStream(path))
    fun from(path: String): AvroInputStream<T> = from(Paths.get(path))
    fun from(file: File): AvroInputStream<T> = from(file.toPath())
@@ -53,39 +74,45 @@ class AvroInputStreamBuilder<T>(private val converter: (Any) -> T) {
    fun from(buffer: ByteBuffer): AvroInputStream<T> = from(ByteArrayInputStream(buffer.array()))
 
    fun from(source: InputStream): AvroInputStream<T> {
-      return when (format) {
-         AvroFormat.BinaryFormat -> {
-            val wschema = writerSchema
-               ?: throw SerializationException("Cannot read from binary unless a writer schema is specified")
-            AvroBinaryInputStream(source, converter, wschema, readerSchema)
-         }
-         AvroFormat.JsonFormat -> {
-            val wschema = writerSchema
-               ?: throw SerializationException("Cannot read from json unless a writer schema is specified")
-            AvroJsonInputStream(source, converter, wschema, readerSchema)
-         }
-         AvroFormat.DataFormat -> when {
-            writerSchema != null && readerSchema != null ->
-               AvroDataInputStream(source, converter, writerSchema, readerSchema)
-            writerSchema != null ->
-               AvroDataInputStream(source, converter, writerSchema, readerSchema)
-            readerSchema != null ->
-               AvroDataInputStream(source,converter,null, readerSchema)
-            else ->
-               AvroDataInputStream(source, converter, null, null)
-         }
+      val currentDerivedFormat = derivedDecodeFormat
+      val decodeFormatToUse = if(defaultDecodeFormat != currentDerivedFormat){
+         currentDerivedFormat
+      }else {
+         decodeFormat
       }
+      return decodeFormatToUse.createInputStream(source, converter)
    }
 }
+class AvroDeserializerInputStreamBuilder<T>(
+   private val deserializer: DeserializationStrategy<T>,
+   private val avro : Avro,
+   converter: (Any) -> T
+) : AvroInputStreamBuilder<T>(converter) {
+   val defaultReadSchema : Schema by lazy { avro.schema(deserializer.descriptor) }
+}
 
 class AvroOutputStreamBuilder<T>(private val serializer: SerializationStrategy<T>,
                                  private val avro: Avro,
-                                 private val converterFn: (Schema) -> (T) -> GenericRecord) {
-
+                                 private val converterFn: (Schema) -> (T) -> GenericRecord
+){
+   var encodeFormat : AvroEncodeFormat = defaultEncodeFormat
+   @Deprecated("please use AvroEncodeFormat to specify the format")
+   @Suppress("DEPRECATION")
    var format: AvroFormat = AvroFormat.DataFormat
    var schema: Schema? = null
+   @Deprecated("please use AvroEncodeFormat to specify the format")
    var codec: CodecFactory = CodecFactory.nullCodec()
 
+   companion object {
+      val defaultEncodeFormat = AvroEncodeFormat.Data()
+   }
+   @Suppress("DEPRECATION")
+   private val derivedEncodeFormat : AvroEncodeFormat
+      get() = when(format){
+         AvroFormat.JsonFormat -> AvroEncodeFormat.Json
+         AvroFormat.BinaryFormat -> AvroEncodeFormat.Binary
+         AvroFormat.DataFormat -> AvroEncodeFormat.Data(codec)
+      }
    fun to(path: Path): AvroOutputStream<T> = to(Files.newOutputStream(path))
    fun to(path: String): AvroOutputStream<T> = to(Paths.get(path))
    fun to(file: File): AvroOutputStream<T> = to(file.toPath())
@@ -93,11 +120,16 @@ class AvroOutputStreamBuilder<T>(private val serializer: SerializationStrategy<T
    fun to(output: OutputStream): AvroOutputStream<T> {
       val schema = schema ?: avro.schema(serializer)
       val converter = converterFn(schema)
-      return when (format) {
-         AvroFormat.DataFormat -> AvroDataOutputStream(output, converter, schema, codec)
-         AvroFormat.JsonFormat -> AvroJsonOutputStream(output, converter, schema)
-         AvroFormat.BinaryFormat -> AvroBinaryOutputStream(output, converter, schema)
+      return createOutputStream(output, schema, converter)
+   }
+   private fun createOutputStream(output: OutputStream, schema : Schema, converter: (T) -> GenericRecord): AvroOutputStream<T> {
+      val currentDirectEncodeFormat = derivedEncodeFormat
+      val encodeFormatToUse = if(currentDirectEncodeFormat != defaultEncodeFormat) {
+         currentDirectEncodeFormat
+      } else{
+         encodeFormat
       }
+      return encodeFormatToUse.createOutputStream(output, schema, converter)
    }
 }
 @OptIn(ExperimentalSerializationApi::class)
@@ -119,11 +151,11 @@ class Avro(override val serializersModule: SerializersModule = EmptySerializersM
 
    /**
     * Loads an instance of <T> from the given ByteArray, with the assumption that the record was stored
-    * using [AvroFormat.DataFormat]. The schema used will be the embedded schema.
+    * using [AvroEncodeFormat.Data]. The schema used will be the embedded schema.
     */
    override fun <T> decodeFromByteArray(deserializer: DeserializationStrategy<T>, bytes: ByteArray): T =
       openInputStream(deserializer) {
-         format = AvroFormat.DataFormat
+         decodeFormat = AvroDecodeFormat.Data(null, null)
       }.from(bytes).nextOrThrow()
 
    /**
@@ -132,13 +164,9 @@ class Avro(override val serializersModule: SerializersModule = EmptySerializersM
     *
     * <pre>
     * val input = openInputStream<T>(serializer) {
-    *    format = AvroFormat.DataFormat
-    *    readerSchema = mySchema
+    *    decodeFormat = AvroDecodeFormat.Data(writerSchema = null, readerSchema = mySchema)
     * }
     * </pre>
-    *
-    * When using formats [AvroFormat.JsonFormat] or [AvroFormat.BinaryFormat] then the
-    * readerSchema must be set in the configuration function.
     */
    fun openInputStream(f: AvroInputStreamBuilder<Any>.() -> Unit = {}): AvroInputStreamBuilder<Any> {
       val builder = AvroInputStreamBuilder { it }
@@ -146,36 +174,34 @@ class Avro(override val serializersModule: SerializersModule = EmptySerializersM
       return builder
    }
 
+
    /**
     * Creates an [AvroInputStreamBuilder] that will read instances of <T>.
     * Supply a function to this method to configure the builder, eg
     *
     * <pre>
     * val input = openInputStream<T>(serializer) {
-    *    format = AvroFormat.DataFormat
-    *    readerSchema = mySchema
+    *    decodeFormat = AvroDecodeFormat.Data(writerSchema = null, readerSchema = mySchema)
     * }
     * </pre>
-    *
-    * When using formats [AvroFormat.JsonFormat] or [AvroFormat.BinaryFormat] then the
-    * readerSchema must be set in the configuration function.
     */
-   fun <T> openInputStream(serializer: DeserializationStrategy<T>,
-                           f: AvroInputStreamBuilder<T>.() -> Unit = {}): AvroInputStreamBuilder<T> {
-      val builder = AvroInputStreamBuilder { fromRecord(serializer, it as GenericRecord) }
+   fun <T> openInputStream(deserializer: DeserializationStrategy<T>,
+                           f: AvroDeserializerInputStreamBuilder<T>.() -> Unit = {}): AvroDeserializerInputStreamBuilder<T> {
+      val builder = AvroDeserializerInputStreamBuilder(deserializer,this) { fromRecord(deserializer, it as GenericRecord) }
       builder.f()
       return builder
    }
 
+
    /**
     * Writes an instance of <T> using a [Schema] derived from the type.
-    * This method will use the [AvroFormat.DataFormat] format.
+    * This method will use the [AvroEncodeFormat.Data] format without a codec.
     * The written object will be returned as a [ByteArray].
     */
    override fun <T> encodeToByteArray(serializer: SerializationStrategy<T>, value: T): ByteArray {
       val baos = ByteArrayOutputStream()
       openOutputStream(serializer) {
-         format = AvroFormat.DataFormat
+         encodeFormat = AvroEncodeFormat.Data()
       }.to(baos).write(value).close()
       return baos.toByteArray()
    }
@@ -185,8 +211,8 @@ class Avro(override val serializersModule: SerializersModule = EmptySerializersM
     * Supply a function to this method to configure the builder, eg
     *
     * <pre>
-    * val output = dump<T>(serializer) {
-    *    format = AvroFormat.DataFormat
+    * val output = openOutputStream<T>(serializer) {
+    *    encodeFormat = AvroEncodeFormat.Data()
     *    schema = mySchema
     * }
     * </pre>
@@ -230,15 +256,18 @@ class Avro(override val serializersModule: SerializersModule = EmptySerializersM
       return RootRecordDecoder(record, serializersModule).decodeSerializableValue(deserializer)
    }
 
-   fun <T> schema(serializer: SerializationStrategy<T>): Schema {
+   fun schema(descriptor : SerialDescriptor) : Schema {
       return schemaFor(
          serializersModule,
-         serializer.descriptor,
-         serializer.descriptor.annotations,
+         descriptor,
+         descriptor.annotations,
          DefaultNamingStrategy,
          mutableMapOf()
       ).schema()
    }
+   fun <T> schema(serializer: SerializationStrategy<T>): Schema {
+      return schema(serializer.descriptor)
+   }
 
 
 }
\ No newline at end of file
diff --git a/src/main/kotlin/com/sksamuel/avro4k/io/AvroDecodeFormat.kt b/src/main/kotlin/com/sksamuel/avro4k/io/AvroDecodeFormat.kt
new file mode 100644
index 0000000..4a4d98a
--- /dev/null
+++ b/src/main/kotlin/com/sksamuel/avro4k/io/AvroDecodeFormat.kt
@@ -0,0 +1,49 @@
+package com.sksamuel.avro4k.io
+
+import org.apache.avro.Schema
+import java.io.InputStream
+
+/**
+ * Formats that can be used to decode the contents of an [InputStream].
+ */
+sealed class AvroDecodeFormat {
+    abstract fun <T> createInputStream(source : InputStream, converter: (Any) -> T) : AvroInputStream<T>
+    /**
+     * Decoding a binary format with a header that contains the full schema, this is the format usually used for Avro files.
+     *
+     * See https://avro.apache.org/docs/current/spec.html#Data+Serialization+and+Deserialization
+     */
+    data class Data(val writerSchema : Schema?, val readerSchema : Schema?) : AvroDecodeFormat() {
+        constructor(readWriteSchema : Schema?) : this(readWriteSchema, readWriteSchema)
+        override fun <T> createInputStream(source : InputStream, converter: (Any) -> T) = when{
+                writerSchema != null && readerSchema != null ->
+                    AvroDataInputStream(source, converter, writerSchema, readerSchema)
+                writerSchema != null ->
+                    AvroDataInputStream(source, converter, writerSchema, readerSchema)
+                readerSchema != null ->
+                    AvroDataInputStream(source, converter, null, readerSchema)
+                else ->
+                    AvroDataInputStream(source, converter, null, null)
+        }
+    }
+    /**
+     * Decodes the binary format without the header, the most compact format.
+     *
+     * See https://avro.apache.org/docs/current/spec.html#binary_encoding
+     */
+    data class Binary(val writerSchema: Schema, val readerSchema: Schema) : AvroDecodeFormat() {
+        constructor(readWriteSchema : Schema) : this(readWriteSchema, readWriteSchema)
+        override fun <T> createInputStream(source : InputStream, converter: (Any) -> T) =  AvroBinaryInputStream(source, converter, writerSchema, readerSchema)
+    }
+    /**
+     * Decodes avro records that have been encoded in JSON. The most verbose format, but easy for a human to read.
+     *
+     * The avro json format does not include the schema. Thus the write and read schema is needed for decoding.
+     *
+     * See https://avro.apache.org/docs/current/spec.html#json_encoding
+     */
+    data class Json(val writerSchema: Schema, val readerSchema: Schema) : AvroDecodeFormat() {
+        constructor(readWriteSchema : Schema) : this(readWriteSchema, readWriteSchema)
+        override fun <T> createInputStream(source : InputStream, converter: (Any) -> T) =  AvroJsonInputStream(source, converter, writerSchema, readerSchema)
+    }
+}
\ No newline at end of file
diff --git a/src/main/kotlin/com/sksamuel/avro4k/io/AvroEncodeFormat.kt b/src/main/kotlin/com/sksamuel/avro4k/io/AvroEncodeFormat.kt
new file mode 100644
index 0000000..ddfebd1
--- /dev/null
+++ b/src/main/kotlin/com/sksamuel/avro4k/io/AvroEncodeFormat.kt
@@ -0,0 +1,54 @@
+package com.sksamuel.avro4k.io
+
+import org.apache.avro.Schema
+import org.apache.avro.file.CodecFactory
+import org.apache.avro.generic.GenericRecord
+import java.io.OutputStream
+
+/**
+ * Formats that can be used to encode a record to an [OutputStream].
+ */
+sealed class AvroEncodeFormat {
+    abstract fun <T> createOutputStream(output : OutputStream, schema : Schema, converter : (T) -> GenericRecord) : AvroOutputStream<T>
+    /**
+     * Encodes a record in a binary format with a header that contains the full schema, this is the format usually used when writing Avro files.
+     *
+     * See https://avro.apache.org/docs/current/spec.html#Data+Serialization+and+Deserialization
+     */
+    data class Data(private val codecFactory: CodecFactory = CodecFactory.nullCodec()) : AvroEncodeFormat() {
+        override fun <T> createOutputStream(
+            output: OutputStream,
+            schema: Schema,
+            converter: (T) -> GenericRecord
+        ): AvroOutputStream<T> {
+            return AvroDataOutputStream(output, converter, schema, codecFactory)
+        }
+    }
+    /**
+     * Encodes the record in a binary format without schema information, the most compact format.
+     *
+     * See https://avro.apache.org/docs/current/spec.html#binary_encoding
+     */
+    object Binary : AvroEncodeFormat() {
+        override fun <T> createOutputStream(
+            output: OutputStream,
+            schema: Schema,
+            converter: (T) -> GenericRecord
+        ) = AvroBinaryOutputStream(output, converter, schema)
+
+    }
+    /**
+     * Encodes teh avro records as JSON text. The most verbose format, but easy for a human to read.
+     *
+     * This format does not include the schema.
+     *
+     * See https://avro.apache.org/docs/current/spec.html#json_encoding
+     */
+    object Json : AvroEncodeFormat() {
+        override fun <T> createOutputStream(
+            output: OutputStream,
+            schema: Schema,
+            converter: (T) -> GenericRecord
+        ) = AvroJsonOutputStream(output, converter, schema)
+    }
+}
\ No newline at end of file
diff --git a/src/main/kotlin/com/sksamuel/avro4k/io/AvroFormat.kt b/src/main/kotlin/com/sksamuel/avro4k/io/AvroFormat.kt
index 52d9cd4..1991fe4 100644
--- a/src/main/kotlin/com/sksamuel/avro4k/io/AvroFormat.kt
+++ b/src/main/kotlin/com/sksamuel/avro4k/io/AvroFormat.kt
@@ -1,5 +1,8 @@
+@file:Suppress("DEPRECATION")
+
 package com.sksamuel.avro4k.io
 
+@Deprecated("use AvroEncodeFormat and AvroDecodeFormat")
 sealed class AvroFormat {
 
    /**
@@ -7,19 +10,24 @@ sealed class AvroFormat {
     *
     * See https://avro.apache.org/docs/current/spec.html#binary_encoding
     */
+   @Deprecated("use AvroEncodeFormat and AvroDecodeFormat")
    object BinaryFormat : AvroFormat()
 
    /**
     * Text format encoded as JSON. The most verbose format, but easy for a human to read.
     *
+    * This format does not include the schema. Thus the write and read schema is needed for decoding.
+    *
     * See https://avro.apache.org/docs/current/spec.html#json_encoding
     */
+   @Deprecated("use AvroEncodeFormat and AvroDecodeFormat")
    object JsonFormat : AvroFormat()
 
    /**
-    * Binary format with an header that contains the full schema, this is the format usually used when writing Avro files.
+    * Binary format with a header that contains the full schema, this is the format usually used when writing Avro files.
     *
     * See https://avro.apache.org/docs/current/spec.html#Data+Serialization+and+Deserialization
     */
+   @Deprecated("use AvroEncodeFormat and AvroDecodeFormat")
    object DataFormat : AvroFormat()
 }
\ No newline at end of file
diff --git a/src/main/kotlin/com/sksamuel/avro4k/io/AvroInputStream.kt b/src/main/kotlin/com/sksamuel/avro4k/io/AvroInputStream.kt
index b102eb8..edc25d3 100644
--- a/src/main/kotlin/com/sksamuel/avro4k/io/AvroInputStream.kt
+++ b/src/main/kotlin/com/sksamuel/avro4k/io/AvroInputStream.kt
@@ -1,18 +1,6 @@
 package com.sksamuel.avro4k.io
 
-import com.sksamuel.avro4k.Avro
-import kotlinx.serialization.DeserializationStrategy
-import kotlinx.serialization.KSerializer
 import kotlinx.serialization.SerializationException
-import org.apache.avro.Schema
-import org.apache.avro.generic.GenericRecord
-import java.io.ByteArrayInputStream
-import java.io.File
-import java.io.InputStream
-import java.nio.ByteBuffer
-import java.nio.file.Files
-import java.nio.file.Path
-import java.nio.file.Paths
 
 interface AvroInputStream<T> : AutoCloseable {
 
@@ -35,113 +23,4 @@ interface AvroInputStream<T> : AutoCloseable {
    fun next(): T?
 
    fun nextOrThrow(): T = next() ?: throw SerializationException("No next entity found")
-
-   companion object {
-
-      /**
-       * Creates an [AvroInputStreamBuilder] that will read instances of T
-       * from binary encoded files.
-       *
-       * Binary encoded files do not include the schema, and therefore
-       * require a schema to make sense of the data. This is known as the
-       * 'writer schema' or in other words, the schema that was originally
-       * used when creating the data.
-       *
-       * This writer schema can be supplied here, or if omitted, then a schema
-       * will be generated from the class using [Avro.schema]. This of course
-       * requires that the types have not changed between writing the original
-       * data and reading it back in here. Otherwise the schemas will not match.
-       */
-      @Deprecated("Create an instance of [AvroInputStream] using the openInputStream method on Avro.default")
-      fun <T> binary(serializer: KSerializer<T>,
-                     writerSchema: Schema? = null,
-                     avro: Avro = Avro.default): AvroInputStreamBuilder<T> {
-         val writer = writerSchema ?: avro.schema(serializer)
-         val converter = { record: Any -> avro.fromRecord(serializer, record as GenericRecord) }
-         return AvroInputStreamBuilder(converter, AvroFormat.BinaryFormat, writer, null)
-      }
-
-      @Deprecated("Create an instance of [AvroInputStream] using the openInputStream method on Avro.default")
-      fun binary(writerSchema: Schema): AvroInputStreamBuilder<Any> {
-         return AvroInputStreamBuilder({ it }, AvroFormat.BinaryFormat, writerSchema, null)
-      }
-
-      /**
-       * Creates an [AvroInputStreamBuilder] that will read instances of T from binary
-       * encoded files with the schema present. To convert to instance of T, the
-       * data source must contain [GenericRecord]s.
-       */
-      @Deprecated("Create an instance of [AvroInputStream] using the openInputStream method on Avro.default")
-      fun <T> data(serializer: DeserializationStrategy<T>,
-                   writerSchema: Schema? = null,
-                   avro: Avro = Avro.default): AvroInputStreamBuilder<T> {
-         val converter = { record: Any -> avro.fromRecord(serializer, record as GenericRecord) }
-         return AvroInputStreamBuilder(converter, AvroFormat.DataFormat, writerSchema, null)
-      }
-
-      /**
-       * Creates an [AvroInputStreamBuilder] that will read Avro values from binary
-       * encoded files with the schema present.
-       */
-      @Deprecated("Create an instance of [AvroInputStream] using the openInputStream method on Avro.default")
-      fun data(writerSchema: Schema? = null): AvroInputStreamBuilder<Any> =
-         AvroInputStreamBuilder({ it }, AvroFormat.DataFormat, writerSchema, null)
-
-      /**
-       * Creates an [AvroInputStreamBuilder] that will read instances of T from json
-       * encoded files. To convert to instance of T, the data source must contain [GenericRecord]s.
-       *
-       * JSON encoded files do not include the schema, and therefore
-       * require a schema to make sense of the data. This is the writer schema, which is
-       * the schema that was originally used when writing the data.
-       *
-       * This writer schema can be supplied here, or if omitted, then a schema
-       * will be generated from the class using [Avro.schema]. This of course
-       * requires that the types have not changed between writing the original
-       * data and reading it back in here. Otherwise the schemas will not match.
-       */
-      @Deprecated("Create an instance of [AvroInputStream] using the openInputStream method on Avro")
-      fun <T> json(serializer: KSerializer<T>,
-                   writerSchema: Schema? = null,
-                   avro: Avro = Avro.default): AvroInputStreamBuilder<T> {
-         val wschema = writerSchema ?: avro.schema(serializer)
-         val converter = { record: Any -> avro.fromRecord(serializer, record as GenericRecord) }
-         return AvroInputStreamBuilder(converter, AvroFormat.JsonFormat, wschema, null)
-      }
-
-      /**
-       * Creates an [AvroInputStreamBuilder] that will read Avro values from json
-       * encoded files.
-       */
-      @Deprecated("Create an instance of [AvroInputStream] using the openInputStream method on Avro.default")
-      fun json(writerSchema: Schema): AvroInputStreamBuilder<Any> =
-         AvroInputStreamBuilder({ it }, AvroFormat.DataFormat, writerSchema, null)
-   }
-}
-
-class AvroInputStreamBuilder<T>(private val converter: (Any) -> T,
-                                private val format: AvroFormat,
-                                private val wschema: Schema?,
-                                private val rschema: Schema?) {
-
-   fun withWriterSchema(schema: Schema) = AvroInputStreamBuilder(converter, format, schema, rschema)
-   fun withReaderSchema(schema: Schema) = AvroInputStreamBuilder(converter, format, wschema, schema)
-
-   fun from(path: Path): AvroInputStream<T> = from(Files.newInputStream(path))
-   fun from(path: String): AvroInputStream<T> = from(Paths.get(path))
-   fun from(file: File): AvroInputStream<T> = from(file.toPath())
-   fun from(bytes: ByteArray): AvroInputStream<T> = from(ByteArrayInputStream(bytes))
-   fun from(buffer: ByteBuffer): AvroInputStream<T> = from(ByteArrayInputStream(buffer.array()))
-
-   fun from(source: InputStream): AvroInputStream<T> {
-      return when (format) {
-         AvroFormat.BinaryFormat -> AvroBinaryInputStream(source, converter, wschema!!, rschema)
-         AvroFormat.JsonFormat -> AvroJsonInputStream(source, converter, wschema!!, rschema)
-         AvroFormat.DataFormat -> when {
-            wschema != null && rschema != null -> AvroDataInputStream(source, converter, wschema, rschema)
-            wschema != null -> AvroDataInputStream(source, converter, wschema, wschema)
-            else -> AvroDataInputStream(source, converter, null, null)
-         }
-      }
-   }
 }
\ No newline at end of file
diff --git a/src/main/kotlin/com/sksamuel/avro4k/io/AvroOutputStream.kt b/src/main/kotlin/com/sksamuel/avro4k/io/AvroOutputStream.kt
index 66eeb63..a350da6 100644
--- a/src/main/kotlin/com/sksamuel/avro4k/io/AvroOutputStream.kt
+++ b/src/main/kotlin/com/sksamuel/avro4k/io/AvroOutputStream.kt
@@ -1,16 +1,5 @@
 package com.sksamuel.avro4k.io
 
-import com.sksamuel.avro4k.Avro
-import kotlinx.serialization.SerializationStrategy
-import org.apache.avro.Schema
-import org.apache.avro.file.CodecFactory
-import org.apache.avro.generic.GenericRecord
-import java.io.File
-import java.io.OutputStream
-import java.nio.file.Files
-import java.nio.file.Path
-import java.nio.file.Paths
-
 /**
  * An [AvroOutputStream] will write instances of T to an underlying
  * representation.
@@ -33,109 +22,4 @@ interface AvroOutputStream<T> : AutoCloseable {
       ts.forEach { write(it) }
       return this
    }
-
-   companion object {
-
-      /**
-       * An [AvroOutputStream] that writes [GenericRecord]s as binary without the schema. Use this when
-       * you want the smallest messages possible at the cost of not having the schema available
-       * in the messages for downstream clients.
-       */
-      @Deprecated("Create an instance of [AvroOutputStream] using the openOutputStream method on Avro.default")
-      fun binary(schema: Schema): AvroOutputStreamBuilder<GenericRecord> =
-         AvroOutputStreamBuilder(schema, { it }, AvroFormat.BinaryFormat)
-
-      /**
-       * An [AvroOutputStream] that writes as binary without the schema. Use this when
-       * you want the smallest messages possible at the cost of not having the schema available
-       * in the messages for downstream clients.
-       */
-      @Deprecated("Create an instance of [AvroOutputStream] using the openOutputStream method on Avro.default")
-      fun <T> binary(schema: Schema,
-                     serializer: SerializationStrategy<T>,
-                     avro: Avro = Avro.default): AvroOutputStreamBuilder<T> =
-         AvroOutputStreamBuilder(schema, { avro.toRecord(serializer, it) }, AvroFormat.BinaryFormat)
-
-      /**
-       * An [AvroOutputStream] that writes as binary without the schema. Use this when
-       * you want the smallest messages possible at the cost of not having the schema available
-       * in the messages for downstream clients.
-       */
-      @Deprecated("Create an instance of [AvroOutputStream] using the openOutputStream method on Avro.default")
-      fun <T> binary(serializer: SerializationStrategy<T>,
-                     avro: Avro = Avro.default): AvroOutputStreamBuilder<T> =
-         binary(avro.schema(serializer), serializer, avro)
-
-      /**
-       * An [AvroOutputStream] that writes [GenericRecord]s as JSON.
-       * The schema is not included in the output.
-       */
-      @Deprecated("Create an instance of [AvroOutputStream] using the openOutputStream method on Avro.default")
-      fun json(schema: Schema): AvroOutputStreamBuilder<GenericRecord> =
-         AvroOutputStreamBuilder(schema, { it }, AvroFormat.JsonFormat)
-
-      /**
-       * An [AvroOutputStream] that writes values of T as JSON.
-       * The schema is not included in the output.
-       */
-      @Deprecated("Create an instance of [AvroOutputStream] using the openOutputStream method on Avro.default")
-      fun <T> json(schema: Schema,
-                   serializer: SerializationStrategy<T>,
-                   avro: Avro = Avro.default): AvroOutputStreamBuilder<T> =
-         AvroOutputStreamBuilder(schema, { avro.toRecord(serializer, it) }, AvroFormat.JsonFormat)
-
-      /**
-       * An [AvroOutputStream] that writes values of T as JSON.
-       * The schema is not included in the output.
-       */
-      @Deprecated("Create an instance of [AvroOutputStream] using the openOutputStream method on Avro.default")
-      fun <T> json(serializer: SerializationStrategy<T>,
-                   avro: Avro = Avro.default): AvroOutputStreamBuilder<T> =
-         json(avro.schema(serializer), serializer, avro)
-
-      /**
-       * An [AvroOutputStream] that writes [GenericRecord]s as binary with the inclusion of the schema.
-       */
-      @Deprecated("Create an instance of [AvroOutputStream] using the openOutputStream method on Avro")
-      fun data(schema: Schema): AvroOutputStreamBuilder<GenericRecord> =
-         AvroOutputStreamBuilder(schema, { it }, AvroFormat.DataFormat)
-
-      /**
-       * An [AvroOutputStream] that writes values of <T> as binary with the inclusion of the schema.
-       * The serializer must be supplied so that Ts can be converted to Avro records.
-       * The schema is generated from the serializer.
-       */
-      @Deprecated("Create an instance of [AvroOutputStream] using the openOutputStream method on Avro.default")
-      fun <T> data(serializer: SerializationStrategy<T>,
-                   avro: Avro = Avro.default) =
-         data(avro.schema(serializer), serializer, avro)
-
-      /**
-       * An [AvroOutputStream] that writes as binary with the inclusion of the given schema.
-       */
-      @Deprecated("Create an instance of [AvroOutputStream] using the openOutputStream method on Avro.default")
-      fun <T> data(schema: Schema,
-                   serializer: SerializationStrategy<T>,
-                   avro: Avro = Avro.default) =
-         AvroOutputStreamBuilder<T>(schema, { avro.toRecord(serializer, it) }, AvroFormat.DataFormat)
-   }
-}
-
-@Deprecated("Create an instance of [AvroOutputStream] using the openOutputStream method on Avro.default")
-class AvroOutputStreamBuilder<T>(private val schema: Schema,
-                                 private val converter: (T) -> GenericRecord,
-                                 private val format: AvroFormat,
-                                 private val codec: CodecFactory = CodecFactory.nullCodec()) {
-
-   fun withCodec(codec: CodecFactory) = AvroOutputStreamBuilder(schema, converter, format, codec)
-
-   fun to(path: Path): AvroOutputStream<T> = to(Files.newOutputStream(path))
-   fun to(path: String): AvroOutputStream<T> = to(Paths.get(path))
-   fun to(file: File): AvroOutputStream<T> = to(file.toPath())
-
-   fun to(output: OutputStream): AvroOutputStream<T> = when (format) {
-      AvroFormat.DataFormat -> AvroDataOutputStream(output, converter, schema, codec)
-      AvroFormat.JsonFormat -> AvroJsonOutputStream(output, converter, schema)
-      AvroFormat.BinaryFormat -> AvroBinaryOutputStream(output, converter, schema)
-   }
 }
\ No newline at end of file
diff --git a/src/main/kotlin/com/sksamuel/avro4k/io/SchemalessAvroInputStream.kt b/src/main/kotlin/com/sksamuel/avro4k/io/SchemalessAvroInputStream.kt
index a057c8d..cb647c0 100644
--- a/src/main/kotlin/com/sksamuel/avro4k/io/SchemalessAvroInputStream.kt
+++ b/src/main/kotlin/com/sksamuel/avro4k/io/SchemalessAvroInputStream.kt
@@ -18,12 +18,9 @@ import java.io.InputStream
 abstract class SchemalessAvroInputStream<T>(private val input: InputStream,
                                             private val converter: (Any) -> T,
                                             writerSchema: Schema,
-                                            readerSchema: Schema?) : AvroInputStream<T> {
+                                            readerSchema: Schema) : AvroInputStream<T> {
 
-   private val datumReader = when (readerSchema) {
-      null -> GenericDatumReader<GenericRecord>(writerSchema)
-      else -> GenericDatumReader(writerSchema, readerSchema)
-   }
+   private val datumReader = GenericDatumReader<GenericRecord>(writerSchema, readerSchema)
 
    abstract val decoder: Decoder
 
@@ -45,9 +42,9 @@ abstract class SchemalessAvroInputStream<T>(private val input: InputStream,
 class AvroJsonInputStream<T>(input: InputStream,
                              converter: (Any) -> T,
                              writerSchema: Schema,
-                             readerSchema: Schema?) :
+                             readerSchema: Schema) :
    SchemalessAvroInputStream<T>(input, converter, writerSchema, readerSchema) {
-   override val decoder: JsonDecoder = DecoderFactory.get().jsonDecoder(writerSchema, input)
+   override val decoder: JsonDecoder = DecoderFactory.get().jsonDecoder(readerSchema, input)
 }
 
 
@@ -59,7 +56,7 @@ class AvroJsonInputStream<T>(input: InputStream,
 class AvroBinaryInputStream<T>(input: InputStream,
                                converter: (Any) -> T,
                                writerSchema: Schema,
-                               readerSchema: Schema?) :
+                               readerSchema: Schema) :
    SchemalessAvroInputStream<T>(input, converter, writerSchema, readerSchema) {
    override val decoder: BinaryDecoder = DecoderFactory.get().binaryDecoder(input, null)
 }
\ No newline at end of file


Test patch:
diff --git a/src/test/kotlin/com/sksamuel/avro4k/decoder/AvroDefaultValuesDecoderTest.kt b/src/test/kotlin/com/sksamuel/avro4k/decoder/AvroDefaultValuesDecoderTest.kt
index 9d02a41..db971f9 100644
--- a/src/test/kotlin/com/sksamuel/avro4k/decoder/AvroDefaultValuesDecoderTest.kt
+++ b/src/test/kotlin/com/sksamuel/avro4k/decoder/AvroDefaultValuesDecoderTest.kt
@@ -4,7 +4,7 @@ import com.sksamuel.avro4k.Avro
 import com.sksamuel.avro4k.AvroDefault
 import com.sksamuel.avro4k.AvroName
 import com.sksamuel.avro4k.ScalePrecision
-import com.sksamuel.avro4k.io.AvroFormat
+import com.sksamuel.avro4k.io.AvroDecodeFormat
 import com.sksamuel.avro4k.serializer.BigDecimalSerializer
 import io.kotest.core.spec.style.FunSpec
 import io.kotest.matchers.collections.shouldBeEmpty
@@ -55,14 +55,13 @@ data class ContainerWithDefaultFields(
 class AvroDefaultValuesDecoderTest : FunSpec({
    test("test default values correctly decoded") {
       val name = "abc"
-      val schema = Avro.default.schema(ContainerWithoutDefaultFields.serializer())
-      val record = GenericData.Record(schema)
+      val writerSchema = Avro.default.schema(ContainerWithoutDefaultFields.serializer())
+      val record = GenericData.Record(writerSchema)
       record.put("name", name)
 
       val byteArray = Avro.default.encodeToByteArray(ContainerWithoutDefaultFields("abc"))
       val deserialized = Avro.default.openInputStream(ContainerWithDefaultFields.serializer()){
-         format = AvroFormat.DataFormat
-         readerSchema = Avro.default.schema(ContainerWithDefaultFields.serializer())
+         decodeFormat = AvroDecodeFormat.Data(writerSchema, defaultReadSchema)
       }.from(byteArray).nextOrThrow()
       deserialized.name.shouldBe("abc")
       deserialized.strDefault.shouldBe("hello")
diff --git a/src/test/kotlin/com/sksamuel/avro4k/io/AvroBinaryOutputStreamTest.kt b/src/test/kotlin/com/sksamuel/avro4k/io/AvroBinaryOutputStreamTest.kt
index 1a315c7..4c092ca 100644
--- a/src/test/kotlin/com/sksamuel/avro4k/io/AvroBinaryOutputStreamTest.kt
+++ b/src/test/kotlin/com/sksamuel/avro4k/io/AvroBinaryOutputStreamTest.kt
@@ -23,7 +23,7 @@ class AvroBinaryOutputStreamTest : StringSpec({
 
       val baos = ByteArrayOutputStream()
       Avro.default.openOutputStream(Composer.serializer()) {
-         format = AvroFormat.BinaryFormat
+         encodeFormat = AvroEncodeFormat.Binary
       }.to(baos).write(ennio).write(hans).close()
 
       // the schema should not be written in a binary stream
diff --git a/src/test/kotlin/com/sksamuel/avro4k/io/AvroDataOutputStreamCodecTest.kt b/src/test/kotlin/com/sksamuel/avro4k/io/AvroDataOutputStreamCodecTest.kt
index e686992..49a5ae2 100644
--- a/src/test/kotlin/com/sksamuel/avro4k/io/AvroDataOutputStreamCodecTest.kt
+++ b/src/test/kotlin/com/sksamuel/avro4k/io/AvroDataOutputStreamCodecTest.kt
@@ -1,5 +1,6 @@
 package com.sksamuel.avro4k.io
 
+import com.sksamuel.avro4k.Avro
 import io.kotest.core.spec.style.StringSpec
 import io.kotest.matchers.should
 import io.kotest.matchers.shouldNot
@@ -17,9 +18,9 @@ class AvroDataOutputStreamCodecTest : StringSpec({
 
    "include schema"  {
       val baos = ByteArrayOutputStream()
-      val output = AvroOutputStream.data(Composer.serializer()).to(baos)
-      output.write(ennio)
-      output.close()
+      Avro.default.openOutputStream(Composer.serializer()) {
+         encodeFormat = AvroEncodeFormat.Data()
+      }.to(baos).write(ennio).close()
       String(baos.toByteArray()) should contain("birthplace")
       String(baos.toByteArray()) should contain("compositions")
    }
@@ -27,11 +28,9 @@ class AvroDataOutputStreamCodecTest : StringSpec({
    "include snappy coded in metadata when serialized with snappy"  {
 
       val baos = ByteArrayOutputStream()
-      val output = AvroOutputStream.data(Composer.serializer())
-         .withCodec(CodecFactory.snappyCodec())
-         .to(baos)
-      output.write(ennio)
-      output.close()
+      Avro.default.openOutputStream(Composer.serializer()) {
+         encodeFormat = AvroEncodeFormat.Data(CodecFactory.snappyCodec())
+      }.to(baos).write(ennio).close()
       String(baos.toByteArray()) should contain("snappy")
       String(baos.toByteArray()) shouldNot contain("bzip2")
       String(baos.toByteArray()) shouldNot contain("deflate")
@@ -39,11 +38,9 @@ class AvroDataOutputStreamCodecTest : StringSpec({
 
    "include deflate coded in metadata when serialized with deflate"  {
       val baos = ByteArrayOutputStream()
-      val output = AvroOutputStream.data(Composer.serializer())
-         .withCodec(CodecFactory.deflateCodec(CodecFactory.DEFAULT_DEFLATE_LEVEL))
-         .to(baos)
-      output.write(ennio)
-      output.close()
+      Avro.default.openOutputStream(Composer.serializer()) {
+         encodeFormat = AvroEncodeFormat.Data(CodecFactory.deflateCodec(CodecFactory.DEFAULT_DEFLATE_LEVEL))
+      }.to(baos).write(ennio).close()
       String(baos.toByteArray()) should contain("deflate")
       String(baos.toByteArray()) shouldNot contain("bzip2")
       String(baos.toByteArray()) shouldNot contain("snappy")
@@ -51,11 +48,9 @@ class AvroDataOutputStreamCodecTest : StringSpec({
 
    "include bzip2 coded in metadata when serialized with bzip2"  {
       val baos = ByteArrayOutputStream()
-      val output = AvroOutputStream.data(Composer.serializer())
-         .withCodec(CodecFactory.bzip2Codec())
-         .to(baos)
-      output.write(ennio)
-      output.close()
+      Avro.default.openOutputStream(Composer.serializer()) {
+         encodeFormat = AvroEncodeFormat.Data(CodecFactory.bzip2Codec())
+      }.to(baos).write(ennio).close()
       String(baos.toByteArray()) should contain("bzip2")
       String(baos.toByteArray()) shouldNot contain("deflate")
       String(baos.toByteArray()) shouldNot contain("snappy")
diff --git a/src/test/kotlin/com/sksamuel/avro4k/io/AvroJsonOutputStreamTest.kt b/src/test/kotlin/com/sksamuel/avro4k/io/AvroJsonOutputStreamTest.kt
new file mode 100644
index 0000000..c83e501
--- /dev/null
+++ b/src/test/kotlin/com/sksamuel/avro4k/io/AvroJsonOutputStreamTest.kt
@@ -0,0 +1,35 @@
+package com.sksamuel.avro4k.io
+
+import com.sksamuel.avro4k.Avro
+import io.kotest.core.spec.style.StringSpec
+import io.kotest.inspectors.forAll
+import io.kotest.matchers.string.shouldContain
+import kotlinx.serialization.Serializable
+import java.io.ByteArrayOutputStream
+
+class AvroJsonOutputStreamTest : StringSpec({
+
+    @Serializable
+    data class Work(val name: String, val year: Int)
+
+    @Serializable
+    data class Composer(val name: String, val birthplace: String, val works: List<Work>)
+
+    val ennio = Composer("ennio morricone", "rome", listOf(Work("legend of 1900", 1986), Work("ecstasy of gold", 1969)))
+
+    val hans = Composer("hans zimmer", "frankfurt", listOf(Work("batman begins", 2007), Work("dunkirk", 2017)))
+
+    "AvroJsonOutputStream should write schemas"  {
+
+        val baos = ByteArrayOutputStream()
+        Avro.default.openOutputStream(Composer.serializer()) {
+            encodeFormat = AvroEncodeFormat.Json
+        }.to(baos).write(ennio).write(hans).close()
+        val jsonString = String(baos.toByteArray())
+        // the schema should be written in a json stream
+        listOf("name", "birthplace", "works", "year").forAll {
+            jsonString.shouldContain(it)
+        }
+    }
+
+})
\ No newline at end of file
diff --git a/src/test/kotlin/com/sksamuel/avro4k/io/AvroNameIoTest.kt b/src/test/kotlin/com/sksamuel/avro4k/io/AvroNameIoTest.kt
index 76ccf30..22c8441 100644
--- a/src/test/kotlin/com/sksamuel/avro4k/io/AvroNameIoTest.kt
+++ b/src/test/kotlin/com/sksamuel/avro4k/io/AvroNameIoTest.kt
@@ -2,13 +2,14 @@ package com.sksamuel.avro4k.io
 
 import com.sksamuel.avro4k.Avro
 import com.sksamuel.avro4k.AvroName
-import io.kotest.matchers.shouldBe
 import io.kotest.core.spec.style.StringSpec
+import io.kotest.matchers.shouldBe
 import kotlinx.serialization.Serializable
 import org.apache.avro.Schema
 import org.apache.avro.SchemaBuilder
 import org.apache.avro.generic.GenericData
-import java.io.File
+import org.apache.avro.generic.GenericRecord
+import java.nio.file.Files
 
 class AvroNameIoTest : StringSpec({
 
@@ -28,8 +29,9 @@ class AvroNameIoTest : StringSpec({
          .name("status").type(Schema.create(Schema.Type.STRING)).noDefault()
          .endRecord()
 
-      AvroInputStream.data(Composer.serializer(), schema).from(bytes).nextOrThrow() shouldBe ennio
-
+      Avro.default.openInputStream(Composer.serializer()){
+         decodeFormat = AvroDecodeFormat.Data(schema, defaultReadSchema)
+      }.from(bytes).nextOrThrow() shouldBe ennio
    }
 
    "using @AvroName to read a record back in" {
@@ -43,22 +45,26 @@ class AvroNameIoTest : StringSpec({
       record.put("fullname", "Ennio Morricone")
       record.put("status", "Maestro")
 
-      val file = File("avroname.avro")
+      val file = Files.createTempFile("avroname.avro","")
 
-      val output = AvroOutputStream.binary(schema1).to(file)
-      output.write(record)
-      output.close()
+      val outputStream = Files.newOutputStream(file)
+      AvroBinaryOutputStream<GenericRecord>(outputStream, {it}, schema1).write(record).close()
 
       @Serializable
       data class Composer(@AvroName("fullname") val name: String, val status: String)
 
       val schema2 = Avro.default.schema(Composer.serializer())
-      val input = AvroInputStream.binary(Composer.serializer(), schema2).from(file)
+      val input = Avro.default.openInputStream(Composer.serializer()) {
+         decodeFormat = AvroDecodeFormat.Binary(
+            writerSchema = schema1,
+            readerSchema = schema2
+         )
+      }.from(file)
 
       input.next() shouldBe Composer("Ennio Morricone", "Maestro")
       input.close()
 
-      file.delete()
+      Files.delete(file)
    }
 
 })
\ No newline at end of file
diff --git a/src/test/kotlin/com/sksamuel/avro4k/io/StreamTests.kt b/src/test/kotlin/com/sksamuel/avro4k/io/StreamTests.kt
index 1a63fe7..9653614 100644
--- a/src/test/kotlin/com/sksamuel/avro4k/io/StreamTests.kt
+++ b/src/test/kotlin/com/sksamuel/avro4k/io/StreamTests.kt
@@ -1,5 +1,6 @@
 package com.sksamuel.avro4k.io
 
+
 import com.sksamuel.avro4k.Avro
 import io.kotest.matchers.shouldBe
 import kotlinx.serialization.KSerializer
@@ -60,7 +61,7 @@ fun <T> writeData(t: T, serializer: SerializationStrategy<T>): ByteArray {
    val schema = Avro.default.schema(serializer)
    val out = ByteArrayOutputStream()
    val avro = Avro.default.openOutputStream(serializer) {
-      format = AvroFormat.DataFormat
+      encodeFormat = AvroEncodeFormat.Data()
       this.schema = schema
    }.to(out)
    avro.write(t)
@@ -79,7 +80,7 @@ fun <T> writeJson(t: T, serializer: KSerializer<T>): ByteArray {
    val schema = Avro.default.schema(serializer)
    val baos = ByteArrayOutputStream()
    val output = Avro.default.openOutputStream(serializer) {
-      format = AvroFormat.JsonFormat
+      encodeFormat = AvroEncodeFormat.Json
       this.schema = schema
    }.to(baos)
    output.write(t)
@@ -90,8 +91,7 @@ fun <T> writeJson(t: T, serializer: KSerializer<T>): ByteArray {
 fun <T> readData(bytes: ByteArray, serializer: KSerializer<T>): GenericRecord {
    val schema = Avro.default.schema(serializer)
    val avro = Avro.default.openInputStream {
-      format = AvroFormat.DataFormat
-      readerSchema = schema
+      decodeFormat = AvroDecodeFormat.Data(schema)
    }.from(bytes)
    return avro.next() as GenericRecord
 }
@@ -100,7 +100,7 @@ fun <T> writeBinary(t: T, serializer: SerializationStrategy<T>): ByteArray {
    val schema = Avro.default.schema(serializer)
    val out = ByteArrayOutputStream()
    val avro = Avro.default.openOutputStream(serializer) {
-      format = AvroFormat.BinaryFormat
+      encodeFormat = AvroEncodeFormat.Binary
       this.schema = schema
    }.to(out)
    avro.write(t)
diff --git a/src/test/kotlin/com/sksamuel/avro4k/schema/AvroDefaultSchemaTest.kt b/src/test/kotlin/com/sksamuel/avro4k/schema/AvroDefaultSchemaTest.kt
index 6acb916..428b1c9 100644
--- a/src/test/kotlin/com/sksamuel/avro4k/schema/AvroDefaultSchemaTest.kt
+++ b/src/test/kotlin/com/sksamuel/avro4k/schema/AvroDefaultSchemaTest.kt
@@ -8,7 +8,6 @@ import io.kotest.core.spec.style.FunSpec
 import io.kotest.matchers.shouldBe
 import kotlinx.serialization.Serializable
 import org.apache.avro.AvroTypeException
-import org.apache.avro.Conversions
 import java.math.BigDecimal
 
 @Suppress("BlockingMethodInNonBlockingContext")
@@ -34,7 +33,6 @@ class AvroDefaultSchemaTest : FunSpec() {
 
       test("schema for data class with @AvroDefault should include default value as a BigDecimal") {
          val expected = org.apache.avro.Schema.Parser().parse(javaClass.getResourceAsStream("/avro_default_annotation_big_decimal.json"))
-         val conversion = Conversions.DecimalConversion()
          val schema = Avro.default.schema(BarDecimal.serializer())
          schema.toString(true) shouldBe expected.toString(true)
       }
diff --git a/src/test/kotlin/com/sksamuel/avro4k/schema/BasicSchemaTest.kt b/src/test/kotlin/com/sksamuel/avro4k/schema/BasicSchemaTest.kt
index 5e250e4..3f36492 100644
--- a/src/test/kotlin/com/sksamuel/avro4k/schema/BasicSchemaTest.kt
+++ b/src/test/kotlin/com/sksamuel/avro4k/schema/BasicSchemaTest.kt
@@ -1,10 +1,10 @@
 package com.sksamuel.avro4k.schema
 
 import com.sksamuel.avro4k.Avro
-import com.sksamuel.avro4k.io.AvroInputStream
-import com.sksamuel.avro4k.io.AvroOutputStream
-import io.kotest.matchers.shouldBe
+import com.sksamuel.avro4k.io.AvroDecodeFormat
+import com.sksamuel.avro4k.io.AvroEncodeFormat
 import io.kotest.core.spec.style.FunSpec
+import io.kotest.matchers.shouldBe
 import kotlinx.serialization.Serializable
 import java.io.File
 
@@ -19,13 +19,18 @@ fun main() {
    val veg = Pizza("veg", listOf(Ingredient("peppers", 0.1, 0.3), Ingredient("onion", 1.0, 0.4)), true, 265)
    val hawaiian = Pizza("hawaiian", listOf(Ingredient("ham", 1.5, 5.6), Ingredient("pineapple", 5.2, 0.2)), false, 391)
 
-   val schema = Avro.default.schema(Pizza.serializer())
+   val writeSchema = Avro.default.schema(Pizza.serializer())
 
-   val output = AvroOutputStream.binary(schema, Pizza.serializer()).to(File("pizzas.avro"))
+   val output = Avro.default.openOutputStream(Pizza.serializer()){
+       encodeFormat = AvroEncodeFormat.Binary
+       schema = writeSchema
+   }.to(File("pizzas.avro"))
    output.write(listOf(veg, hawaiian))
    output.close()
 
-   val input = AvroInputStream.binary(Pizza.serializer(), schema).from(File("pizzas.avro"))
+   val input = Avro.default.openInputStream(Pizza.serializer()){
+       decodeFormat = AvroDecodeFormat.Binary(writeSchema, defaultReadSchema)
+   }.from(File("pizzas.avro"))
    input.iterator().forEach { println(it) }
    input.close()
 }