Buffer implementation

Author: lauzadis

runtime/runtime-core/api/runtime-core.api

@@ -1031,6 +1031,9 @@ public final class aws/smithy/kotlin/runtime/io/SdkSourceJVMKt {
 
 public final class aws/smithy/kotlin/runtime/io/SdkSourceKt {
 	public static final fun readFully (Laws/smithy/kotlin/runtime/io/SdkSource;Laws/smithy/kotlin/runtime/io/SdkBuffer;J)V
+	public static final fun readToByteArray (Laws/smithy/kotlin/runtime/io/SdkSource;Lkotlin/coroutines/Continuation;)Ljava/lang/Object;
+	public static final fun toSdkByteReadChannel (Laws/smithy/kotlin/runtime/io/SdkSource;Lkotlinx/coroutines/CoroutineScope;)Laws/smithy/kotlin/runtime/io/SdkByteReadChannel;
+	public static synthetic fun toSdkByteReadChannel$default (Laws/smithy/kotlin/runtime/io/SdkSource;Lkotlinx/coroutines/CoroutineScope;ILjava/lang/Object;)Laws/smithy/kotlin/runtime/io/SdkByteReadChannel;
 }
 
 public final class aws/smithy/kotlin/runtime/io/internal/ConvertKt {

runtime/runtime-core/common/src/aws/smithy/kotlin/runtime/io/SdkSource.kt

@@ -6,7 +6,16 @@
 package aws.smithy.kotlin.runtime.io
 
 import aws.smithy.kotlin.runtime.InternalApi
+import aws.smithy.kotlin.runtime.io.internal.JobChannel
+import kotlinx.coroutines.CoroutineName
 import kotlinx.coroutines.CoroutineScope
+import kotlinx.coroutines.DelicateCoroutinesApi
+import kotlinx.coroutines.Dispatchers
+import kotlinx.coroutines.GlobalScope
+import kotlinx.coroutines.IO
+import kotlinx.coroutines.ensureActive
+import kotlinx.coroutines.launch
+import kotlinx.coroutines.withContext
 
 /**
  * A source for reading a stream of bytes (e.g. from file, network, or in-memory buffer). Sources may
@@ -43,16 +52,42 @@ public interface SdkSource : Closeable {
  * Consume the [SdkSource] and pull the entire contents into memory as a [ByteArray].
  */
 @InternalApi
-public expect suspend fun SdkSource.readToByteArray(): ByteArray
+public suspend fun SdkSource.readToByteArray(): ByteArray = withContext(Dispatchers.IO) {
+    use { it.buffer().readByteArray() }
+}
 
 /**
  * Convert the [SdkSource] to an [SdkByteReadChannel]. Content is read from the source and forwarded
  * to the channel.
  * @param coroutineScope the coroutine scope to use to launch a background reader channel responsible for propagating data
  * between source and the returned channel
  */
+@OptIn(DelicateCoroutinesApi::class)
 @InternalApi
-public expect fun SdkSource.toSdkByteReadChannel(coroutineScope: CoroutineScope? = null): SdkByteReadChannel
+public fun SdkSource.toSdkByteReadChannel(coroutineScope: CoroutineScope? = null): SdkByteReadChannel {
+    val source = this
+    val ch = JobChannel()
+    val scope = coroutineScope ?: GlobalScope
+    val job = scope.launch(Dispatchers.IO + CoroutineName("sdk-source-reader")) {
+        val buffer = SdkBuffer()
+        val result = runCatching {
+            source.use {
+                while (true) {
+                    ensureActive()
+                    val rc = source.read(buffer, DEFAULT_BYTE_CHANNEL_MAX_BUFFER_SIZE.toLong())
+                    if (rc == -1L) break
+                    ch.write(buffer)
+                }
+            }
+        }
+
+        ch.close(result.exceptionOrNull())
+    }
+
+    ch.attachJob(job)
+
+    return ch
+}
 
 /**
  * Remove exactly [byteCount] bytes from this source and appends them to [sink].

runtime/runtime-core/common/test/aws/smithy/kotlin/runtime/io/SdkBufferedSinkTest.kt

@@ -25,7 +25,6 @@ abstract class AbstractBufferedSinkTest(
     private val data = SdkBuffer()
     private val sink = factory(data)
 
-    @IgnoreNative // FIXME Re-enable after Kotlin/Native implementation
     @Test
     fun testWriteByte() {
         sink.writeByte(0xDE.toByte())
@@ -36,7 +35,6 @@ abstract class AbstractBufferedSinkTest(
         assertEquals("[hex=deadbeef]", data.toString())
     }
 
-    @IgnoreNative // FIXME Re-enable after Kotlin/Native implementation
     @Test
     fun testWriteShort() {
         sink.writeShort(0xdead.toShort())
@@ -45,7 +43,6 @@ abstract class AbstractBufferedSinkTest(
         assertEquals("[hex=deadbeef]", data.toString())
     }
 
-    @IgnoreNative // FIXME Re-enable after Kotlin/Native implementation
     @Test
     fun testWriteShortLe() {
         sink.writeShortLe(0xdead.toShort())
@@ -54,55 +51,48 @@ abstract class AbstractBufferedSinkTest(
         assertEquals("[hex=addeefbe]", data.toString())
     }
 
-    @IgnoreNative // FIXME Re-enable after Kotlin/Native implementation
     @Test
     fun testWriteInt() {
         sink.writeInt(0xdeadbeef.toInt())
         sink.flush()
         assertEquals("[hex=deadbeef]", data.toString())
     }
 
-    @IgnoreNative // FIXME Re-enable after Kotlin/Native implementation
     @Test
     fun testWriteLe() {
         sink.writeIntLe(0xdeadbeef.toInt())
         sink.flush()
         assertEquals("[hex=efbeadde]", data.toString())
     }
 
-    @IgnoreNative // FIXME Re-enable after Kotlin/Native implementation
     @Test
     fun testWriteLong() {
         sink.writeLong(-2401053092341600192)
         sink.flush()
         assertEquals("[hex=deadbeef10203040]", data.toString())
     }
 
-    @IgnoreNative // FIXME Re-enable after Kotlin/Native implementation
     @Test
     fun testWriteLongLe() {
         sink.writeLongLe(4625232074423315934)
         sink.flush()
         assertEquals("[hex=deadbeef10203040]", data.toString())
     }
 
-    @IgnoreNative // FIXME Re-enable after Kotlin/Native implementation
     @Test
     fun testWriteString() {
         sink.writeUtf8("レップはボールです")
         sink.flush()
         assertEquals("[text=レップはボールです]", data.toString())
     }
 
-    @IgnoreNative // FIXME Re-enable after Kotlin/Native implementation
     @Test
     fun testWriteSubstring() {
         sink.writeUtf8("a lep is a ball", start = 2, endExclusive = 10)
         sink.flush()
         assertEquals("lep is a", data.readUtf8())
     }
 
-    @IgnoreNative // FIXME Re-enable after Kotlin/Native implementation
     @Test
     fun testWriteAll() {
         val contents = "a tay is a hammer"
@@ -113,7 +103,6 @@ abstract class AbstractBufferedSinkTest(
         assertEquals(contents.length.toLong(), rc)
     }
 
-    @IgnoreNative // FIXME Re-enable after Kotlin/Native implementation
     @Test
     fun testReadSourceFully() {
         val source = object : SdkSource by SdkBuffer() {
@@ -128,7 +117,7 @@ abstract class AbstractBufferedSinkTest(
         assertEquals("12341234", data.readUtf8())
     }
 
-    @IgnoreNative // FIXME Re-enable after Kotlin/Native implementation
+    @IgnoreNative // FIXME "Expected an exception of aws.smithy.kotlin.runtime.io.EOFException to be thrown, but was okio.EOFException"
     @Test
     fun testWriteEof() {
         val source: SdkSource = SdkBuffer().apply { writeUtf8("1234") }
@@ -137,15 +126,13 @@ abstract class AbstractBufferedSinkTest(
         assertEquals("1234", data.readUtf8())
     }
 
-    @IgnoreNative // FIXME Re-enable after Kotlin/Native implementation
     @Test
     fun testWriteExhausted() {
         val source: SdkSource = SdkBuffer()
         assertEquals(0, sink.writeAll(source))
         assertEquals(0, data.size)
     }
 
-    @IgnoreNative // FIXME Re-enable after Kotlin/Native implementation
     @Test
     fun testWriteExplicitZero() {
         val source = object : SdkSource by SdkBuffer() {
@@ -156,15 +143,13 @@ abstract class AbstractBufferedSinkTest(
         assertEquals(0, data.size)
     }
 
-    @IgnoreNative // FIXME Re-enable after Kotlin/Native implementation
     @Test
     fun testCloseFlushes() {
         sink.writeUtf8("a flix is a comb")
         sink.close()
         assertEquals("a flix is a comb", data.readUtf8())
     }
 
-    @IgnoreNative // FIXME Re-enable after Kotlin/Native implementation
     @Test
     fun testWriteByteArray() {
         val expected = bytes(0xde, 0xad, 0xbe, 0xef)
@@ -174,7 +159,6 @@ abstract class AbstractBufferedSinkTest(
         assertContentEquals(expected, actual)
     }
 
-    @IgnoreNative // FIXME Re-enable after Kotlin/Native implementation
     @Test
     fun testWriteByteArrayOffset() {
         val expected = bytes(0xde, 0xad, 0xbe, 0xef)
@@ -184,7 +168,6 @@ abstract class AbstractBufferedSinkTest(
         assertContentEquals(expected.sliceArray(2..3), actual)
     }
 
-    @IgnoreNative // FIXME Re-enable after Kotlin/Native implementation
     @Test
     fun testWriteByteArrayOffsetAndLimit() {
         val expected = bytes(0xde, 0xad, 0xbe, 0xef)

runtime/runtime-core/jvm/src/aws/smithy/kotlin/runtime/io/SdkSourceJVM.kt

@@ -4,67 +4,4 @@
  */
 package aws.smithy.kotlin.runtime.io
 
-internal actual class BufferedSinkAdapter actual constructor(sink: okio.BufferedSink) : SdkBufferedSink {
-    actual override val buffer: SdkBuffer
-        get() = TODO("Not yet implemented")
-
-    actual override fun write(source: ByteArray, offset: Int, limit: Int) {
-        TODO("Not yet implemented")
-    }
-
-    actual override fun writeAll(source: SdkSource): Long {
-        TODO("Not yet implemented")
-    }
-
-    actual override fun write(source: SdkSource, byteCount: Long) {
-        TODO("Not yet implemented")
-    }
-
-    actual override fun writeUtf8(string: String, start: Int, endExclusive: Int) {
-        TODO("Not yet implemented")
-    }
-
-    actual override fun writeByte(x: Byte) {
-        TODO("Not yet implemented")
-    }
-
-    actual override fun writeShort(x: Short) {
-        TODO("Not yet implemented")
-    }
-
-    actual override fun writeShortLe(x: Short) {
-        TODO("Not yet implemented")
-    }
-
-    actual override fun writeInt(x: Int) {
-        TODO("Not yet implemented")
-    }
-
-    actual override fun writeIntLe(x: Int) {
-        TODO("Not yet implemented")
-    }
-
-    actual override fun writeLong(x: Long) {
-        TODO("Not yet implemented")
-    }
-
-    actual override fun writeLongLe(x: Long) {
-        TODO("Not yet implemented")
-    }
-
-    actual override fun flush() {
-        TODO("Not yet implemented")
-    }
-
-    actual override fun emit() {
-        TODO("Not yet implemented")
-    }
-
-    actual override fun write(source: SdkBuffer, byteCount: Long) {
-        TODO("Not yet implemented")
-    }
-
-    actual override fun close() {
-        TODO("Not yet implemented")
-    }
-}
+internal actual class BufferedSinkAdapter actual constructor(sink: okio.BufferedSink) : SdkBufferedSink, AbstractBufferedSinkAdapter(sink)