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

build-logic/src/main/kotlin/kotlinx/io/conventions/kotlinx-io-multiplatform.gradle.kts

@@ -52,9 +52,12 @@ kotlin {
     configureNativePlatforms()
 
     val nativeTargets = nativeTargets()
+    val appleTargets = appleTargets()
     sourceSets {
-        createSourceSet("nativeMain", parent = commonMain.get(), children = nativeTargets)
-        createSourceSet("nativeTest", parent = commonTest.get(), children = nativeTargets)
+        val nativeMain = createSourceSet("nativeMain", parent = commonMain.get(), children = nativeTargets)
+        val nativeTest = createSourceSet("nativeTest", parent = commonTest.get(), children = nativeTargets)
+        createSourceSet("appleMain", parent = nativeMain, children = appleTargets)
+        createSourceSet("appleTest", parent = nativeTest, children = appleTargets)
     }
 }
 
@@ -126,7 +129,7 @@ fun KotlinMultiplatformExtension.configureNativePlatforms() {
 }
 
 fun nativeTargets(): List<String> {
-    return appleTargets() + linuxTargets() + mingwTargets() + androidTargets()
+    return linuxTargets() + mingwTargets() + androidTargets()
 }
 
 fun appleTargets() = listOf(

core/apple/src/-Util.kt

@@ -0,0 +1,21 @@
+/*
+ * 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.UnsafeNumber
+import platform.Foundation.NSError
+import platform.Foundation.NSLocalizedDescriptionKey
+import platform.Foundation.NSUnderlyingErrorKey
+
+@OptIn(UnsafeNumber::class)
+internal fun Exception.toNSError() = NSError(
+    domain = "Kotlin",
+    code = 0,
+    userInfo = mapOf(
+        NSLocalizedDescriptionKey to message,
+        NSUnderlyingErrorKey to this
+    )
+)

core/apple/src/AppleCore.kt

@@ -0,0 +1,117 @@
+/*
+ * 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.
+ */
+
+@file:OptIn(UnsafeNumber::class)
+
+package kotlinx.io
+
+import kotlinx.cinterop.*
+import platform.Foundation.NSInputStream
+import platform.Foundation.NSOutputStream
+import platform.Foundation.NSStreamStatusClosed
+import platform.Foundation.NSStreamStatusNotOpen
+import platform.posix.uint8_tVar
+
+/**
+ * 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 {
+
+    init {
+        if (out.streamStatus == NSStreamStatusNotOpen) out.open()
+    }
+
+    override fun write(source: Buffer, byteCount: Long) {
+        if (out.streamStatus == NSStreamStatusClosed) throw IOException("Stream Closed")
+
+        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<uint8_tVar>()
+                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 {
+
+    init {
+        if (input.streamStatus == NSStreamStatusNotOpen) input.open()
+    }
+
+    override fun readAtMostTo(sink: Buffer, byteCount: Long): Long {
+        if (input.streamStatus == NSStreamStatusClosed) throw IOException("Stream Closed")
+
+        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<uint8_tVar>()
+            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/BuffersApple.kt

@@ -0,0 +1,74 @@
+/*
+ * 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.
+ */
+
+@file:OptIn(UnsafeNumber::class)
+
+package kotlinx.io
+
+import kotlinx.cinterop.*
+import platform.Foundation.*
+import platform.darwin.ByteVar
+import platform.darwin.NSUIntegerMax
+import platform.posix.*
+
+internal fun Buffer.write(source: CPointer<uint8_tVar>, maxLength: Int) {
+    require(maxLength >= 0) { "maxLength ($maxLength) must not be negative" }
+
+    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
+}
+
+internal fun Buffer.readAtMostTo(sink: CPointer<uint8_tVar>, maxLength: Int): Int {
+    require(maxLength >= 0) { "maxLength ($maxLength) must not be negative" }
+
+    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
+}
+
+internal fun Buffer.snapshotAsNSData(): NSData {
+    if (size == 0L) return NSData.data()
+
+    check(size.toULong() <= NSUIntegerMax) { "Buffer is too long ($size) to be converted into NSData." }
+
+    val bytes = malloc(size.convert())?.reinterpret<uint8_tVar>()
+        ?: throw Error("malloc failed: ${strerror(errno)?.toKString()}")
+    var curr = head
+    var index = 0
+    do {
+        check(curr != null) { "Current segment is null" }
+        val pos = curr.pos
+        val length = curr.limit - pos
+        curr.data.usePinned {
+            memcpy(bytes + index, it.addressOf(pos), length.convert())
+        }
+        curr = curr.next
+        index += length
+    } while (curr !== head)
+    return NSData.create(bytesNoCopy = bytes, length = size.convert())
+}

core/apple/src/SinksApple.kt

@@ -0,0 +1,157 @@
+/*
+ * 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.uint8_tVar
+import kotlin.native.ref.WeakReference
+
+/**
+ * Returns an output stream that writes to this sink. Closing the stream will also close this sink.
+ *
+ * The stream supports both polling and run-loop scheduling, please check
+ * [Apple's documentation](https://developer.apple.com/library/archive/documentation/Cocoa/Conceptual/Streams/Articles/PollingVersusRunloop.html)
+ * for information about stream events handling.
+ *
+ * The stream does not implement initializers
+ * ([NSOutputStream.initToBuffer](https://developer.apple.com/documentation/foundation/nsoutputstream/1410805-inittobuffer),
+ * [NSOutputStream.initToMemory](https://developer.apple.com/documentation/foundation/nsoutputstream/1409909-inittomemory),
+ * [NSOutputStream.initWithURL](https://developer.apple.com/documentation/foundation/nsoutputstream/1414446-initwithurl),
+ * [NSOutputStream.initToFileAtPath](https://developer.apple.com/documentation/foundation/nsoutputstream/1416367-inittofileatpath)),
+ * their use will result in a runtime error.
+ *
+ * @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), NSStreamDelegateProtocol {
+
+    private val isClosed: () -> Boolean = when (sink) {
+        is RealSink -> sink::closed
+        is Buffer -> {
+            { false }
+        }
+    }
+
+    private var status = NSStreamStatusNotOpen
+    private var error: NSError? = null
+        set(value) {
+            status = NSStreamStatusError
+            field = value
+            postEvent(NSStreamEventErrorOccurred)
+            sink.close()
+        }
+
+    override fun streamStatus() = if (status != NSStreamStatusError && isClosed()) NSStreamStatusClosed else status
+
+    override fun streamError() = error
+
+    override fun open() {
+        if (status == NSStreamStatusNotOpen) {
+            status = NSStreamStatusOpening
+            status = NSStreamStatusOpen
+            postEvent(NSStreamEventOpenCompleted)
+            postEvent(NSStreamEventHasSpaceAvailable)
+        }
+    }
+
+    override fun close() {
+        if (status == NSStreamStatusError || status == NSStreamStatusNotOpen) return
+        status = NSStreamStatusClosed
+        runLoop = null
+        runLoopModes = listOf()
+        sink.close()
+    }
+
+    @OptIn(DelicateIoApi::class)
+    override fun write(buffer: CPointer<uint8_tVar>?, maxLength: NSUInteger): NSInteger {
+        if (streamStatus != NSStreamStatusOpen || buffer == null) return -1
+        status = NSStreamStatusWriting
+        val toWrite = minOf(maxLength, Int.MAX_VALUE.convert()).toInt()
+        return try {
+            sink.writeToInternalBuffer {
+                it.write(buffer, toWrite)
+            }
+            status = NSStreamStatusOpen
+            toWrite.convert()
+        } catch (e: Exception) {
+            error = e.toNSError()
+            -1
+        }
+    }
+
+    override fun hasSpaceAvailable() = !isFinished
+
+    private val isFinished
+        get() = when (streamStatus) {
+            NSStreamStatusClosed, NSStreamStatusError -> true
+            else -> false
+        }
+
+    @OptIn(InternalIoApi::class)
+    override fun propertyForKey(key: NSStreamPropertyKey): Any? = when (key) {
+        NSStreamDataWrittenToMemoryStreamKey -> sink.buffer.snapshotAsNSData()
+        else -> null
+    }
+
+    override fun setProperty(property: Any?, forKey: NSStreamPropertyKey) = false
+
+    // WeakReference as delegate should not be retained
+    // https://developer.apple.com/documentation/foundation/nsstream/1418423-delegate
+    private var _delegate: WeakReference<NSStreamDelegateProtocol>? = null
+    private var runLoop: NSRunLoop? = null
+    private var runLoopModes = listOf<NSRunLoopMode>()
+
+    private fun postEvent(event: NSStreamEvent) {
+        val runLoop = runLoop ?: return
+        runLoop.performInModes(runLoopModes) {
+            if (runLoop == this.runLoop) {
+                delegateOrSelf.stream(this, event)
+            }
+        }
+    }
+
+    override fun delegate() = _delegate?.value
+
+    private val delegateOrSelf get() = delegate ?: this
+
+    override fun setDelegate(delegate: NSStreamDelegateProtocol?) {
+        _delegate = delegate?.let { WeakReference(it) }
+    }
+
+    override fun stream(aStream: NSStream, handleEvent: NSStreamEvent) {
+        // no-op
+    }
+
+    override fun scheduleInRunLoop(aRunLoop: NSRunLoop, forMode: NSRunLoopMode) {
+        if (runLoop == null) {
+            runLoop = aRunLoop
+        }
+        if (runLoop == aRunLoop) {
+            runLoopModes += forMode
+        }
+        if (status == NSStreamStatusOpen) {
+            postEvent(NSStreamEventHasSpaceAvailable)
+        }
+    }
+
+    override fun removeFromRunLoop(aRunLoop: NSRunLoop, forMode: NSRunLoopMode) {
+        if (aRunLoop == runLoop) {
+            runLoopModes -= forMode
+            if (runLoopModes.isEmpty()) {
+                runLoop = null
+            }
+        }
+    }
+
+    override fun description() = "$sink.asNSOutputStream()"
+}