Introduce interoperability with java.io (#95)

Author: qwwdfsad

TODO.md

@@ -8,6 +8,7 @@
 - Implementation
     - Index preconditions
     - Prototype `PipedOutput`
+    - Benchmark overhead `java.io` integration
 - Test
     - Verify pool has no leaks
 - Documentation

core/commonMain/src/kotlinx/io/Input.kt

@@ -146,13 +146,12 @@ public abstract class Input : Closeable {
     }
 
     /**
-     * Reads the available content in current [Input] to the [destination] buffer.
+     * Reads the available content in the current [Input] to the [destination] buffer.
      *
-     * If no bytes are available in the input, [fill] method will be called directly on
-     * the [destination] buffer without an extra copy.
+     * If no bytes are available, [fill] method will be called directly on the [destination] buffer without an extra copy.
      * Otherwise, available bytes are copied to the destination.
      *
-     * @return number of bytes written in the [destination].
+     * @return number of bytes written to the [destination].
      */
     public fun readAvailableTo(
         destination: Buffer,

core/commonMain/src/kotlinx/io/InputOperations.kt

@@ -147,7 +147,6 @@ public fun Input.discardExact(count: Int): Int {
     return count
 }
 
-
 /**
  * Reads a [Byte] from this Input.
  *

core/commonMain/src/kotlinx/io/Inputs.kt renamed to core/commonMain/src/kotlinx/io/Inputs.common.kt

@@ -54,5 +54,4 @@ class LimitingInputTest {
         assertTrue(closed)
     }
 
-    private fun StringInput(str: String) = ByteArrayInput(str.encodeToByteArray())
 }

core/commonTest/src/kotlinx/io/LimitingInputTest.kt

@@ -3,7 +3,7 @@ package kotlinx.io
 import kotlin.test.*
 
 fun assertArrayEquals(expected: ByteArray, actual: ByteArray) {
-    assertEquals(expected.size, actual.size)
+    assertEquals(expected.size, actual.size, "Expected array lengths to be equal")
     assertEquals(expected.toHexString(), actual.toHexString())
 }
 
@@ -17,4 +17,6 @@ internal fun Bytes.useInput(block: Input.() -> Unit) {
     } finally {
         close()
     }
-}
+}
+
+public fun StringInput(string: String) = ByteArrayInput(string.encodeToByteArray())

core/commonTest/src/kotlinx/io/TestUtils.kt

@@ -0,0 +1,16 @@
+package kotlinx.io
+
+import kotlinx.io.internal.*
+import java.io.*
+
+/**
+ * Returns an [InputStream] that uses the current [Input] as an underlying source of data.
+ * Closing the resulting [InputStream] will close the input.
+ */
+public fun Input.asInputStream(): InputStream = InputStreamFromInput(this)
+
+/**
+ * Returns an [Input] that uses the current [InputStream] as an underlying source of data.
+ * Closing the resulting [Input] will close the input stream.
+ */
+public fun InputStream.asInput(): Input = InputFromInputStream(this)

core/jvmMain/src/kotlinx/io/Inputs.kt

@@ -0,0 +1,16 @@
+package kotlinx.io
+
+import kotlinx.io.internal.*
+import java.io.*
+
+/**
+ * Returns an [OutputStream] that uses the current [Output] as the destination.
+ * Closing the resulting [OutputStream] will close the input.
+ */
+public fun Output.asOutputStream(): OutputStream = OutputStreamFromOutput(this)
+
+/**
+ * Returns an [Output] that uses the current [OutputStream] as the destination.
+ * Closing the resulting [Output] will close the input stream.
+ */
+public fun OutputStream.asOutput(): Output = OutputFromOutputStream(this)

core/jvmMain/src/kotlinx/io/Outputs.kt

@@ -0,0 +1,92 @@
+package kotlinx.io.internal
+
+import kotlinx.io.*
+import kotlinx.io.buffer.*
+import java.io.*
+
+internal class InputStreamFromInput(private val input: Input) : InputStream() {
+    override fun read(): Int {
+        if (input.exhausted()) {
+            return -1
+        }
+        return input.readByte().toInt() and 0xFF
+    }
+
+    override fun read(b: ByteArray): Int {
+        if (b.isEmpty()) return 0
+        val result = input.readAvailableTo(bufferOf(b))
+        if (result == 0) return -1
+        return result
+    }
+
+    override fun read(b: ByteArray, off: Int, len: Int): Int {
+        if (len == 0) return 0
+        val result = input.readAvailableTo(bufferOf(b), off, off + len)
+        if (result == 0) return -1
+        return result
+    }
+
+    override fun close() {
+        input.close()
+    }
+}
+
+internal class InputFromInputStream(private val inputStream: InputStream) : Input() {
+    override fun closeSource() {
+        inputStream.close()
+    }
+
+    override fun fill(buffer: Buffer, startIndex: Int, endIndex: Int): Int {
+        // Zero-copy attempt
+        if (buffer.buffer.hasArray()) {
+            val result = inputStream.read(buffer.buffer.array(), startIndex, endIndex - startIndex)
+            return result.coerceAtLeast(0) // -1 when IS is closed
+        }
+
+        for (i in startIndex until endIndex) {
+            val byte = inputStream.read()
+            if (byte == -1) return (i - startIndex)
+            buffer[i] = byte.toByte()
+        }
+        return endIndex - startIndex
+    }
+}
+
+internal class OutputStreamFromOutput(private val output: Output) : OutputStream() {
+    override fun write(b: Int) {
+        output.writeByte(b.toByte())
+    }
+
+    override fun write(b: ByteArray) {
+        output.writeBuffer(bufferOf(b))
+    }
+
+    override fun write(b: ByteArray, off: Int, len: Int) {
+        output.writeBuffer(bufferOf(b), off, off + len)
+    }
+
+    override fun flush() {
+        output.flush()
+    }
+
+    override fun close() {
+        output.close()
+    }
+}
+
+internal class OutputFromOutputStream(private val outputStream: OutputStream) : Output() {
+
+    override fun closeSource() {
+        outputStream.close()
+    }
+
+    override fun flush(source: Buffer, startIndex: Int, endIndex: Int) {
+        if (source.buffer.hasArray()) {
+            return outputStream.write(source.buffer.array(), startIndex, endIndex - startIndex)
+        }
+
+        for (i in startIndex until endIndex) {
+            outputStream.write(source[i].toInt())
+        }
+    }
+}

core/jvmMain/src/kotlinx/io/internal/Atomic.kt

@@ -1,6 +1,5 @@
 package kotlinx.io.pool
 
-import kotlinx.io.internal.*
 import java.util.concurrent.atomic.*
 
 private const val MULTIPLIER = 4
@@ -92,7 +91,6 @@ actual abstract class DefaultPool<T : Any> actual constructor(actual final overr
     }
 
     companion object {
-        // todo: replace with atomicfu, remove companion object
-        private val Top = longUpdater(DefaultPool<*>::top)
+        private val Top = AtomicLongFieldUpdater.newUpdater(DefaultPool::class.java, DefaultPool<*>::top.name)
     }
 }

core/jvmMain/src/kotlinx/io/internal/JavaInterop.kt

@@ -4,7 +4,7 @@ package kotlinx.io
 import kotlinx.io.buffer.*
 import kotlin.test.*
 
-class InputOutputTestJvm {
+class CustomPoolTest {
 
     @Test
     fun testCustomPools() {
@@ -38,4 +38,4 @@ class InputOutputTestJvm {
         input.readAvailableTo(output)
         output.flush()
     }
-}
+}