Add Source and Sink extensions for Apple's NSInputStream and NSOutputStream

buildSrc/src/main/kotlin/Platforms.kt

@@ -32,7 +32,7 @@ fun KotlinMultiplatformExtension.configureNativePlatforms() {
     mingwX64()
 }
 
-private val appleTargets = listOf(
+val appleTargets = listOf(
     "iosArm64",
     "iosX64",
     "iosSimulatorArm64",
@@ -64,7 +64,7 @@ private val androidTargets = listOf(
     "androidNativeX86"
 )
 
-val nativeTargets = appleTargets + linuxTargets + mingwTargets + androidTargets
+val nativeTargets = linuxTargets + mingwTargets + androidTargets
 
 /**
  * Creates a source set for a directory that isn't already a built-in platform. Use this to create

core/apple/src/-Util.kt

@@ -0,0 +1,26 @@
+package kotlinx.io
+
+import kotlinx.cinterop.UnsafeNumber
+import kotlinx.cinterop.addressOf
+import kotlinx.cinterop.convert
+import kotlinx.cinterop.usePinned
+import platform.Foundation.*
+
+@OptIn(UnsafeNumber::class)
+internal fun Exception.toNSError() = NSError(
+    domain = "Kotlin",
+    code = 0,
+    userInfo = mapOf(
+        NSLocalizedDescriptionKey to message,
+        NSUnderlyingErrorKey to this
+    )
+)
+
+internal fun ByteArray.toNSData() = if (isNotEmpty()) {
+    usePinned {
+        @OptIn(UnsafeNumber::class)
+        NSData.create(bytes = it.addressOf(0), length = size.convert())
+    }
+} else {
+    NSData.data()
+}

core/apple/src/AppleCore.kt

@@ -0,0 +1,108 @@
+/*
+ * Copyright 2017-2023 JetBrains s.r.o. and respective authors and developers.
+ * Use of this source code is governed by the Apache 2.0 license that can be found in the LICENCE file.
+ */
+
+package kotlinx.io
+
+import kotlinx.cinterop.*
+import platform.Foundation.NSInputStream
+import platform.Foundation.NSOutputStream
+import platform.Foundation.NSStreamStatusNotOpen
+import platform.darwin.UInt8Var
+
+/**
+ * Returns [RawSink] that writes to an output stream.
+ *
+ * Use [RawSink.buffered] to create a buffered sink from it.
+ *
+ * @sample kotlinx.io.samples.KotlinxIoSamplesApple.outputStreamAsSink
+ */
+public fun NSOutputStream.asSink(): RawSink = OutputStreamSink(this)
+
+private open class OutputStreamSink(
+    private val out: NSOutputStream,
+) : RawSink {
+
+    @OptIn(UnsafeNumber::class)
+    override fun write(source: Buffer, byteCount: Long) {
+        if (out.streamStatus == NSStreamStatusNotOpen) out.open()
+
+        checkOffsetAndCount(source.size, 0, byteCount)
+        var remaining = byteCount
+        while (remaining > 0) {
+            val head = source.head!!
+            val toCopy = minOf(remaining, head.limit - head.pos).toInt()
+            val bytesWritten = head.data.usePinned {
+                val bytes = it.addressOf(head.pos).reinterpret<UInt8Var>()
+                out.write(bytes, toCopy.convert()).toLong()
+            }
+
+            if (bytesWritten < 0L) throw IOException(out.streamError?.localizedDescription ?: "Unknown error")
+            if (bytesWritten == 0L) throw IOException("NSOutputStream reached capacity")
+
+            head.pos += bytesWritten.toInt()
+            remaining -= bytesWritten
+            source.size -= bytesWritten
+
+            if (head.pos == head.limit) {
+                source.head = head.pop()
+                SegmentPool.recycle(head)
+            }
+        }
+    }
+
+    override fun flush() {
+        // no-op
+    }
+
+    override fun close() = out.close()
+
+    override fun toString() = "RawSink($out)"
+}
+
+/**
+ * Returns [RawSource] that reads from an input stream.
+ *
+ * Use [RawSource.buffered] to create a buffered source from it.
+ *
+ * @sample kotlinx.io.samples.KotlinxIoSamplesApple.inputStreamAsSource
+ */
+public fun NSInputStream.asSource(): RawSource = NSInputStreamSource(this)
+
+private open class NSInputStreamSource(
+    private val input: NSInputStream,
+) : RawSource {
+
+    @OptIn(UnsafeNumber::class)
+    override fun readAtMostTo(sink: Buffer, byteCount: Long): Long {
+        if (input.streamStatus == NSStreamStatusNotOpen) input.open()
+
+        if (byteCount == 0L) return 0L
+        checkByteCount(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 < 0L) throw IOException(input.streamError?.localizedDescription ?: "Unknown error")
+        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
+    }
+
+    override fun close() = input.close()
+
+    override fun toString() = "RawSource($input)"
+}

core/apple/src/SinksApple.kt

@@ -0,0 +1,101 @@
+/*
+ * Copyright 2017-2023 JetBrains s.r.o. and respective authors and developers.
+ * Use of this source code is governed by the Apache 2.0 license that can be found in the LICENCE file.
+ */
+
+package kotlinx.io
+
+import kotlinx.cinterop.*
+import platform.Foundation.*
+import platform.darwin.NSInteger
+import platform.darwin.NSUInteger
+import platform.posix.memcpy
+import platform.posix.uint8_tVar
+import kotlin.concurrent.Volatile
+
+/**
+ * Returns an output stream that writes to this sink. Closing the stream will also close this sink.
+ *
+ * @sample kotlinx.io.samples.KotlinxIoSamplesApple.asStream
+ */
+public fun Sink.asNSOutputStream(): NSOutputStream = SinkNSOutputStream(this)
+
+@OptIn(UnsafeNumber::class)
+private class SinkNSOutputStream(
+    private val sink: Sink,
+) : NSOutputStream(toMemory = Unit) {
+
+    private val isClosed: () -> Boolean = when (sink) {
+        is RealSink -> sink::closed
+        is Buffer -> {
+            { false }
+        }
+    }
+
+    @OptIn(ExperimentalStdlibApi::class)
+    @Volatile
+    private var status = NSStreamStatusNotOpen
+    private var error: NSError? = null
+        set(value) {
+            status = NSStreamStatusError
+            field = value
+        }
+
+    override fun streamStatus() = status
+
+    override fun streamError() = error
+
+    override fun open() {
+        if (status == NSStreamStatusNotOpen) {
+            status = NSStreamStatusOpen
+        }
+    }
+
+    override fun close() {
+        status = NSStreamStatusClosed
+        sink.close()
+    }
+
+    @OptIn(DelicateIoApi::class)
+    override fun write(buffer: CPointer<uint8_tVar>?, maxLength: NSUInteger): NSInteger {
+        return try {
+            if (isClosed()) throw IOException("Underlying sink is closed.")
+            if (status != NSStreamStatusOpen) return -1
+            status = NSStreamStatusWriting
+            sink.writeToInternalBuffer {
+                it.writeNative(buffer, maxLength.toInt())
+            }
+            status = NSStreamStatusOpen
+            maxLength.convert()
+        } catch (e: Exception) {
+            error = e.toNSError()
+            -1
+        }
+    }
+
+    private fun Buffer.writeNative(source: CPointer<uint8_tVar>?, maxLength: Int) {
+        var currentOffset = 0
+        while (currentOffset < maxLength) {
+            val tail = writableSegment(1)
+
+            val toCopy = minOf(maxLength - currentOffset, Segment.SIZE - tail.limit)
+            tail.data.usePinned {
+                memcpy(it.addressOf(tail.pos), source + currentOffset, toCopy.convert())
+            }
+
+            currentOffset += toCopy
+            tail.limit += toCopy
+        }
+        size += maxLength
+    }
+
+    override fun hasSpaceAvailable() = true
+
+    @OptIn(InternalIoApi::class)
+    override fun propertyForKey(key: NSStreamPropertyKey): Any? = when (key) {
+        NSStreamDataWrittenToMemoryStreamKey -> sink.buffer.readByteArray().toNSData()
+        else -> null
+    }
+
+    override fun description() = "$sink.asNSOutputStream()"
+}

core/apple/src/SourcesApple.kt

@@ -0,0 +1,121 @@
+/*
+ * Copyright 2017-2023 JetBrains s.r.o. and respective authors and developers.
+ * Use of this source code is governed by the Apache 2.0 license that can be found in the LICENCE file.
+ */
+
+package kotlinx.io
+
+import kotlinx.cinterop.*
+import platform.Foundation.*
+import platform.darwin.NSInteger
+import platform.darwin.NSUInteger
+import platform.darwin.NSUIntegerVar
+import platform.posix.memcpy
+import platform.posix.uint8_tVar
+import kotlin.concurrent.Volatile
+
+/**
+ * Returns an input stream that reads from this source. Closing the stream will also close this source.
+ *
+ * @sample kotlinx.io.samples.KotlinxIoSamplesApple.asStream
+ */
+public fun Source.asNSInputStream(): NSInputStream = SourceNSInputStream(this)
+
+@OptIn(InternalIoApi::class, UnsafeNumber::class)
+private class SourceNSInputStream(
+    private val source: Source,
+) : NSInputStream(NSData()) {
+
+    private val isClosed: () -> Boolean = when (source) {
+        is RealSource -> source::closed
+        is Buffer -> {
+            { false }
+        }
+    }
+
+    @OptIn(ExperimentalStdlibApi::class)
+    @Volatile
+    private var status = NSStreamStatusNotOpen
+    private var error: NSError? = null
+        set(value) {
+            status = NSStreamStatusError
+            field = value
+        }
+
+    private var pinnedBuffer: Pinned<ByteArray>? = null
+
+    override fun streamStatus() = if (isClosed()) NSStreamStatusClosed else status
+
+    override fun streamError() = error
+
+    override fun open() {
+        if (status == NSStreamStatusNotOpen) {
+            status = NSStreamStatusOpen
+        }
+    }
+
+    override fun close() {
+        status = NSStreamStatusClosed
+        pinnedBuffer?.unpin()
+        pinnedBuffer = null
+        source.close()
+    }
+
+    override fun read(buffer: CPointer<uint8_tVar>?, maxLength: NSUInteger): NSInteger {
+        try {
+            if (isClosed()) throw IOException("Underlying source is closed.")
+            if (status != NSStreamStatusOpen) return -1
+            status = NSStreamStatusReading
+            if (source.exhausted()) {
+                status = NSStreamStatusAtEnd
+                return 0
+            }
+            val toRead = minOf(maxLength.toInt(), source.buffer.size).toInt()
+            val read = source.buffer.readNative(buffer, toRead).convert<NSInteger>()
+            status = NSStreamStatusOpen
+            return read
+        } catch (e: Exception) {
+            error = e.toNSError()
+            return -1
+        }
+    }
+
+    override fun getBuffer(buffer: CPointer<CPointerVar<uint8_tVar>>?, length: CPointer<NSUIntegerVar>?): Boolean {
+        if (source.buffer.size > 0) {
+            source.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
+    }
+
+    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
+    }
+
+    override fun hasBytesAvailable() = !source.exhausted()
+
+    override fun propertyForKey(key: NSStreamPropertyKey): Any? = null
+
+    override fun description() = "$source.asNSInputStream()"
+}