Refactor asynchronous API to CsvRecordReader and CsvRecordWriter

gradle/libs.versions.toml

@@ -2,13 +2,15 @@
 dokka = "1.9.20"
 junit-jupiter = "5.11.0"
 kotlin = "2.1.0"
+kotlinx-coroutines = "1.9.0"
 kotlinx-serialization-core = "1.7.3"
 nexus-publish = "0.4.0"
 nexus-staging = "0.30.0"
 researchgate-release = "3.0.2"
 
 [libraries]
 junit-jupiter = { module = "org.junit.jupiter:junit-jupiter", version.ref = "junit-jupiter" }
+kotlinx-coroutines-test = { module = "org.jetbrains.kotlinx:kotlinx-coroutines-test", version.ref = "kotlinx-coroutines" }
 kotlinx-serialization-core = { module = "org.jetbrains.kotlinx:kotlinx-serialization-core", version.ref = "kotlinx-serialization-core" }
 
 [plugins]

library/build.gradle.kts

@@ -15,6 +15,7 @@ dependencies {
 
     testImplementation(kotlin("test-junit5"))
     testImplementation(libs.junit.jupiter)
+    testImplementation(libs.kotlinx.coroutines.test)
 }
 
 kotlin {

library/src/main/kotlin/kotlinx/serialization/csv/Csv.kt

@@ -2,24 +2,18 @@ package kotlinx.serialization.csv
 
 import kotlinx.serialization.DeserializationStrategy
 import kotlinx.serialization.ExperimentalSerializationApi
-import kotlinx.serialization.KSerializer
 import kotlinx.serialization.SerialFormat
 import kotlinx.serialization.Serializable
 import kotlinx.serialization.SerializationStrategy
 import kotlinx.serialization.StringFormat
-import kotlinx.serialization.builtins.ListSerializer
 import kotlinx.serialization.csv.config.CsvBuilder
 import kotlinx.serialization.csv.config.CsvConfig
 import kotlinx.serialization.csv.decode.CsvReader
 import kotlinx.serialization.csv.decode.FetchSource
-import kotlinx.serialization.csv.decode.RecordListCsvDecoder
 import kotlinx.serialization.csv.decode.RootCsvDecoder
 import kotlinx.serialization.csv.decode.Source
 import kotlinx.serialization.csv.decode.StringSource
-import kotlinx.serialization.csv.encode.CsvWriter
-import kotlinx.serialization.csv.encode.RecordListCsvEncoder
 import kotlinx.serialization.csv.encode.RootCsvEncoder
-import kotlinx.serialization.encoding.CompositeDecoder.Companion.DECODE_DONE
 import kotlinx.serialization.modules.SerializersModule
 import java.io.Reader
 import java.io.StringWriter
@@ -62,22 +56,6 @@ sealed class Csv(val config: CsvConfig) : StringFormat {
         output.encode(serializer, value)
     }
 
-    /**
-     * Start serializing values into CSV record(s).
-     *
-     * @param serializer The serializer used to serialize the given object.
-     * @param appendable The output where the CSV will be written.
-     * @return A function that emits a new item to the CSV.
-     */
-    @ExperimentalSerializationApi
-    fun <T> beginEncodingToAppendable(serializer: KSerializer<T>, appendable: Appendable): (T) -> Unit {
-        val encoder = RecordListCsvEncoder(this, CsvWriter(appendable, config))
-        var index = 0
-        return {
-            encoder.encodeSerializableValue(serializer, it)
-        }
-    }
-
     /**
      * Parse CSV [string] into [Serializable] object.
      *
@@ -96,49 +74,6 @@ sealed class Csv(val config: CsvConfig) : StringFormat {
     fun <T> decodeFrom(deserializer: DeserializationStrategy<T>, input: Reader): T =
         FetchSource(input).decode(deserializer)
 
-    /**
-     * Parse CSV line-by-line from the given [reader] into a sequence.
-     *
-     * @param deserializer The deserializer used to parse the given CSV string.
-     * @param reader The CSV reader to parse.  This function *does not close the reader*.
-     * @return A sequence of each element decoded.
-     */
-    @ExperimentalSerializationApi
-    fun <T> decodeSequenceFromReader(deserializer: KSerializer<T>, reader: Reader): Sequence<T> {
-        val csv = CsvReader(FetchSource(reader), config)
-        val listDescriptor = ListSerializer(deserializer).descriptor
-        val input = RecordListCsvDecoder(this, csv)
-        var previousValue: T? = null
-
-        return generateSequence {
-            val decodedIndex = input.decodeElementIndex(listDescriptor)
-            if (decodedIndex == DECODE_DONE) return@generateSequence null
-            val nextValue =
-                input.decodeSerializableElement(listDescriptor, decodedIndex, deserializer, previousValue)
-            previousValue = nextValue
-            nextValue
-        }
-    }
-
-    /**
-     * Parse CSV from the given [reader] into a sequence of [Serializable] objects.
-     * Designed to be comparable to [Reader.useLines].
-     *
-     * @param deserializer The deserializer used to parse the given CSV string.
-     * @param reader The CSV reader to parse.
-     * @param handler The code to handle the sequence of incoming values.  The sequence will not be available after the
-     * function completes.
-     */
-    fun <T> decodeFromReaderUsingSequence(
-        deserializer: KSerializer<T>,
-        reader: Reader,
-        handler: (Sequence<T>) -> Unit,
-    ) {
-        reader.use {
-            handler(decodeSequenceFromReader(deserializer, reader))
-        }
-    }
-
     /**
      * Serialize [value] into CSV record(s).
      *

library/src/main/kotlin/kotlinx/serialization/csv/CsvRecordReader.kt

@@ -0,0 +1,48 @@
+package kotlinx.serialization.csv
+
+import kotlinx.serialization.ExperimentalSerializationApi
+import kotlinx.serialization.KSerializer
+import kotlinx.serialization.builtins.ListSerializer
+import kotlinx.serialization.csv.decode.CsvReader
+import kotlinx.serialization.csv.decode.FetchSource
+import kotlinx.serialization.csv.decode.RecordListCsvDecoder
+import kotlinx.serialization.encoding.CompositeDecoder.Companion.DECODE_DONE
+import java.io.Reader
+
+/**
+ * Record reader that allows reading CSV line-by-line.
+ */
+interface CsvRecordReader<T : Any> : Iterator<T> {
+    /**
+     * Read next record
+     */
+    fun read(): T? = if (hasNext()) next() else null
+}
+
+/**
+ * Parse CSV line-by-line from the given [input].
+ *
+ * @param deserializer The deserializer used to parse the given CSV string.
+ * @param input The CSV reader to parse. This function *does not close the reader*.
+ */
+@ExperimentalSerializationApi
+fun <T : Any> Csv.recordReader(deserializer: KSerializer<T>, input: Reader): CsvRecordReader<T> {
+    val decoder = RecordListCsvDecoder(
+        csv = this,
+        reader = CsvReader(FetchSource(input), config)
+    )
+    val listDescriptor = ListSerializer(deserializer).descriptor
+    var previousValue: T? = null
+
+    return object : CsvRecordReader<T> {
+        override fun hasNext(): Boolean =
+            decoder.decodeElementIndex(listDescriptor) != DECODE_DONE
+
+        override fun next(): T {
+            val index = decoder.decodeElementIndex(listDescriptor)
+            return decoder.decodeSerializableElement(listDescriptor, index, deserializer, previousValue).also {
+                previousValue = it
+            }
+        }
+    }
+}

library/src/main/kotlin/kotlinx/serialization/csv/CsvRecordWriter.kt

@@ -0,0 +1,34 @@
+package kotlinx.serialization.csv
+
+import kotlinx.serialization.ExperimentalSerializationApi
+import kotlinx.serialization.KSerializer
+import kotlinx.serialization.csv.encode.CsvWriter
+import kotlinx.serialization.csv.encode.RecordListCsvEncoder
+
+/**
+ * Record writer that allows writing CSV line by line.
+ */
+fun interface CsvRecordWriter<T : Any> {
+    /**
+     * Write next record.
+     */
+    fun write(record: T)
+}
+
+/**
+ * Create [CsvRecordWriter] that allows writing CSV line-by-line.
+ *
+ * @param serializer The serializer used to serialize the given object.
+ * @param output The output where the CSV will be written.
+ */
+@ExperimentalSerializationApi
+fun <T : Any> Csv.recordWriter(serializer: KSerializer<T>, output: Appendable): CsvRecordWriter<T> {
+    val encoder = RecordListCsvEncoder(
+        csv = this,
+        writer = CsvWriter(output, config)
+    )
+
+    return CsvRecordWriter {
+        encoder.encodeSerializableValue(serializer, it)
+    }
+}

library/src/main/kotlin/kotlinx/serialization/csv/encode/CsvWriter.kt

@@ -9,7 +9,7 @@ import kotlinx.serialization.csv.config.QuoteMode
  * To write one CSV record, call [beginRecord], followed by multiple calls to [printColumn] and
  * finally call [endRecord] to finish the record.
  */
-internal class CsvWriter(private val sb: Appendable, private val config: CsvConfig) {
+internal class CsvWriter(private val output: Appendable, private val config: CsvConfig) {
 
     var isFirstRecord = true
     private var isFirstColumn = true
@@ -20,7 +20,7 @@ internal class CsvWriter(private val sb: Appendable, private val config: CsvConf
      */
     fun beginRecord() {
         if (!isFirstRecord) {
-            sb.append(config.recordSeparator)
+            output.append(config.recordSeparator)
         }
     }
 
@@ -64,19 +64,19 @@ internal class CsvWriter(private val sb: Appendable, private val config: CsvConf
                 escapeCharacters = "$escapeChar$delimiter$quoteChar$recordSeparator",
                 escapeChar = escapeChar
             )
-            sb.append(escapedValue)
+            output.append(escapedValue)
         } else if (mode == WriteMode.QUOTED || mode == WriteMode.ESCAPED) {
             val escapedValue = value.replace("$quoteChar", "$quoteChar$quoteChar")
-            sb.append(quoteChar).append(escapedValue).append(quoteChar)
+            output.append(quoteChar).append(escapedValue).append(quoteChar)
         } else {
-            sb.append(value)
+            output.append(value)
         }
     }
 
     /** End the current column (which writes the column delimiter). */
     private fun nextColumn() {
         if (!isFirstColumn) {
-            sb.append(config.delimiter)
+            output.append(config.delimiter)
         }
         isFirstColumn = false
     }