Okio NSInputeStream Source extensions

Until square/okio#1123 is merged and released.

couchbase-lite/src/appleMain/kotlin/com/couchbase/lite/kmp/Blob.kt

@@ -7,6 +7,8 @@ import com.udobny.kmp.DelegatedClass
 import com.udobny.kmp.ext.toByteArray
 import com.udobny.kmp.ext.toNSData
 import okio.*
+import okio.temp.inputStream
+import okio.temp.source
 import platform.Foundation.*
 
 public actual class Blob

couchbase-lite/src/appleMain/kotlin/okio/temp/BufferedSource.kt

@@ -0,0 +1,134 @@
+/*
+ * Copyright (C) 2020 Square, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// TODO: workaround until these extensions are merged and released in Okio
+//  https://github.com/square/okio/pull/1123
+@file:Suppress("INVISIBLE_REFERENCE", "INVISIBLE_MEMBER", "INVISIBLE_SETTER")
+
+package okio.temp
+
+import kotlinx.cinterop.CPointer
+import kotlinx.cinterop.CPointerVar
+import kotlinx.cinterop.Pinned
+import kotlinx.cinterop.UnsafeNumber
+import kotlinx.cinterop.addressOf
+import kotlinx.cinterop.convert
+import kotlinx.cinterop.pin
+import kotlinx.cinterop.pointed
+import kotlinx.cinterop.reinterpret
+import kotlinx.cinterop.usePinned
+import kotlinx.cinterop.value
+import okio.*
+import platform.Foundation.NSData
+import platform.Foundation.NSError
+import platform.Foundation.NSInputStream
+import platform.Foundation.NSLocalizedDescriptionKey
+import platform.Foundation.NSUnderlyingErrorKey
+import platform.darwin.NSInteger
+import platform.darwin.NSUInteger
+import platform.darwin.NSUIntegerVar
+import platform.posix.memcpy
+import platform.posix.uint8_tVar
+
+/** Returns an input stream that reads from this source. */
+fun BufferedSource.inputStream(): NSInputStream = BufferedSourceInputStream(this)
+
+@OptIn(UnsafeNumber::class)
+private class BufferedSourceInputStream(
+    private val bufferedSource: BufferedSource,
+) : NSInputStream(NSData()) {
+
+    private var error: NSError? = null
+    private var pinnedBuffer: Pinned<ByteArray>? = null
+
+    override fun streamError(): NSError? = error
+
+    override fun open() {
+        // no-op
+    }
+
+    override fun read(buffer: CPointer<uint8_tVar>?, maxLength: NSUInteger): NSInteger {
+        try {
+            if (bufferedSource is RealBufferedSource) {
+                if (bufferedSource.closed) throw IOException("closed")
+                if (bufferedSource.exhausted()) return 0
+            }
+
+            val toRead = minOf(maxLength.toInt(), bufferedSource.buffer.size).toInt()
+            return bufferedSource.buffer.readNative(buffer, toRead).convert()
+        } catch (e: Exception) {
+            error = e.toNSError()
+            return -1
+        }
+    }
+
+    override fun getBuffer(
+        buffer: CPointer<CPointerVar<uint8_tVar>>?,
+        length: CPointer<NSUIntegerVar>?,
+    ): Boolean {
+        if (bufferedSource.buffer.size > 0) {
+            bufferedSource.buffer.head?.let { s ->
+                pinnedBuffer?.unpin()
+                s.data.pin().let {
+                    pinnedBuffer = it
+                    buffer?.pointed?.value = it.addressOf(s.pos).reinterpret()
+                    length?.pointed?.value = (s.limit - s.pos).convert()
+                    return true
+                }
+            }
+        }
+        return false
+    }
+
+    override fun hasBytesAvailable(): Boolean = bufferedSource.buffer.size > 0
+
+    override fun close() {
+        pinnedBuffer?.unpin()
+        pinnedBuffer = null
+        bufferedSource.close()
+    }
+
+    override fun description(): String = "$bufferedSource.inputStream()"
+
+    private fun Exception.toNSError(): NSError {
+        return NSError(
+            "Kotlin",
+            0,
+            mapOf(
+                NSLocalizedDescriptionKey to message,
+                NSUnderlyingErrorKey to this,
+            ),
+        )
+    }
+
+    private fun Buffer.readNative(sink: CPointer<uint8_tVar>?, maxLength: Int): Int {
+        val s = head ?: return 0
+        val toCopy = minOf(maxLength, s.limit - s.pos)
+        s.data.usePinned {
+            memcpy(sink, it.addressOf(s.pos), toCopy.convert())
+        }
+
+        s.pos += toCopy
+        size -= toCopy.toLong()
+
+        if (s.pos == s.limit) {
+            head = s.pop()
+            SegmentPool.recycle(s)
+        }
+
+        return toCopy
+    }
+}

couchbase-lite/src/appleMain/kotlin/okio/temp/NSInputStreamSource.kt

@@ -0,0 +1,72 @@
+/*
+ * Copyright (C) 2020 Square, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// TODO: workaround until these extensions are merged and released in Okio
+//  https://github.com/square/okio/pull/1123
+@file:Suppress("INVISIBLE_REFERENCE", "INVISIBLE_MEMBER", "INVISIBLE_SETTER")
+
+package okio.temp
+
+import kotlinx.cinterop.UnsafeNumber
+import kotlinx.cinterop.addressOf
+import kotlinx.cinterop.convert
+import kotlinx.cinterop.reinterpret
+import kotlinx.cinterop.usePinned
+import okio.*
+import platform.Foundation.NSInputStream
+import platform.darwin.UInt8Var
+
+/** Returns a source that reads from `in`. */
+fun NSInputStream.source(): Source = NSInputStreamSource(this)
+
+@OptIn(UnsafeNumber::class)
+private open class NSInputStreamSource(
+    private val input: NSInputStream,
+) : Source {
+
+    init {
+        input.open()
+    }
+
+    override fun read(sink: Buffer, byteCount: Long): Long {
+        if (byteCount == 0L) return 0L
+        require(byteCount >= 0L) { "byteCount < 0: $byteCount" }
+        val tail = sink.writableSegment(1)
+        val maxToCopy = minOf(byteCount, Segment.SIZE - tail.limit)
+        val bytesRead = tail.data.usePinned {
+            val bytes = it.addressOf(tail.limit).reinterpret<UInt8Var>()
+            input.read(bytes, maxToCopy.convert()).toLong()
+        }
+        if (bytesRead < 0) throw IOException(input.streamError?.localizedDescription)
+        if (bytesRead == 0L) {
+            if (tail.pos == tail.limit) {
+                // We allocated a tail segment, but didn't end up needing it. Recycle!
+                sink.head = tail.pop()
+                SegmentPool.recycle(tail)
+            }
+            return -1
+        }
+        tail.limit += bytesRead.toInt()
+        sink.size += bytesRead
+        return bytesRead.convert()
+    }
+
+    override fun close() = input.close()
+
+    override fun timeout() = Timeout.NONE
+
+    override fun toString() = "source($input)"
+}

couchbase-lite/src/appleTest/kotlin/okio/temp/AppleBufferedSourceTest.kt

@@ -0,0 +1,119 @@
+/*
+ * Copyright (C) 2020 Square, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// TODO: workaround until these extensions are merged and released in Okio
+//  https://github.com/square/okio/pull/1123
+@file:Suppress("INVISIBLE_MEMBER")
+
+package okio.temp
+
+import kotlin.test.Test
+import kotlin.test.assertEquals
+import kotlin.test.assertFalse
+import kotlin.test.assertNotNull
+import kotlin.test.assertTrue
+import kotlinx.cinterop.CPointerVar
+import kotlinx.cinterop.addressOf
+import kotlinx.cinterop.alloc
+import kotlinx.cinterop.convert
+import kotlinx.cinterop.get
+import kotlinx.cinterop.memScoped
+import kotlinx.cinterop.ptr
+import kotlinx.cinterop.reinterpret
+import kotlinx.cinterop.usePinned
+import kotlinx.cinterop.value
+import okio.*
+import platform.Foundation.NSInputStream
+import platform.darwin.NSUIntegerVar
+import platform.darwin.UInt8Var
+
+class AppleBufferedSourceTest {
+    @Test fun bufferInputStream() {
+        val source = Buffer()
+        source.writeUtf8("abc")
+        testInputStream(source.inputStream())
+    }
+
+    @Test fun realBufferedSourceInputStream() {
+        val source = Buffer()
+        source.writeUtf8("abc")
+        testInputStream(RealBufferedSource(source).inputStream())
+    }
+
+    private fun testInputStream(nsis: NSInputStream) {
+        nsis.open()
+        val byteArray = ByteArray(4)
+        byteArray.usePinned {
+            val cPtr = it.addressOf(0).reinterpret<UInt8Var>()
+
+            byteArray.fill(-5)
+            assertEquals(3, nsis.read(cPtr, 4U))
+            assertEquals("[97, 98, 99, -5]", byteArray.contentToString())
+
+            byteArray.fill(-7)
+            assertEquals(0, nsis.read(cPtr, 4U))
+            assertEquals("[-7, -7, -7, -7]", byteArray.contentToString())
+        }
+    }
+
+    @Test fun nsInputStreamGetBuffer() {
+        val source = Buffer()
+        source.writeUtf8("abc")
+
+        val nsis = source.inputStream()
+        nsis.open()
+        assertTrue(nsis.hasBytesAvailable)
+
+        memScoped {
+            val bufferPtr = alloc<CPointerVar<UInt8Var>>()
+            val lengthPtr = alloc<NSUIntegerVar>()
+            assertTrue(nsis.getBuffer(bufferPtr.ptr, lengthPtr.ptr))
+
+            val length = lengthPtr.value
+            assertNotNull(length)
+            assertEquals(3.convert(), length)
+
+            val buffer = bufferPtr.value
+            assertNotNull(buffer)
+            assertEquals('a'.code.convert(), buffer[0])
+            assertEquals('b'.code.convert(), buffer[1])
+            assertEquals('c'.code.convert(), buffer[2])
+        }
+    }
+
+    @Test fun nsInputStreamClose() {
+        val buffer = Buffer()
+        buffer.writeUtf8("abc")
+        val source = RealBufferedSource(buffer)
+        assertFalse(source.closed)
+
+        val nsis = source.inputStream()
+        nsis.open()
+        nsis.close()
+        assertTrue(source.closed)
+
+        val byteArray = ByteArray(4)
+        byteArray.usePinned {
+            val cPtr = it.addressOf(0).reinterpret<UInt8Var>()
+
+            byteArray.fill(-5)
+            assertEquals(-1, nsis.read(cPtr, 4U))
+            assertNotNull(nsis.streamError)
+            assertEquals("closed", nsis.streamError?.localizedDescription)
+            assertEquals("[-5, -5, -5, -5]", byteArray.contentToString())
+        }
+    }
+}