From 7572a57fecfc9bde94560cb0e3cb61c339a9360b Mon Sep 17 00:00:00 2001 From: Pierluigi Cifani Date: Tue, 3 Nov 2026 16:52:37 +0100 Subject: [PATCH 1/3] JNI: AsyncStream results as SwiftAsyncStream Functions returning `AsyncStream` were skipped: the type wasn't in the modeled standard library. - SwiftKitCore `SwiftAsyncStream`: `subscribe` starts a Swift task that delivers each element and completion without blocking a Java thread; the returned `Subscription` cancels it. `map` gives a converted view. One subscriber, as for an AsyncStream; the task handle is released exactly once. - SwiftJavaRuntimeSupport `_attachAsyncStream` stores the stream and an element converter behind a handle on the Java object. - Generator: `AsyncStream` joins the stdlib model; a function returning `AsyncStream` is translated as returning `T`, then Java passes a stream out-parameter and maps its elements with that conversion, and the Swift thunk attaches the stream. `async` functions returning streams and tuple elements are skipped with a reason. Co-Authored-By: Claude Opus 5.5 --- .../Sources/MySwiftLibrary/AsyncStreams.swift | 32 ++++ .../com/example/swift/AsyncStreamTest.java | 68 ++++++++ .../JNI/JNIJavaTypeTranslator.swift | 3 + ...ISwift2JavaGenerator+JavaTranslation.swift | 72 ++++++++- ...wift2JavaGenerator+NativeTranslation.swift | 20 +++ .../JavaTypes/JavaType+Futures.swift | 5 + .../SwiftTypes/SwiftKnownModules.swift | 3 + .../SwiftTypes/SwiftKnownTypeDecls.swift | 6 + .../AsyncStreamBridge.swift | 106 +++++++++++++ .../CompletableFutureAwait.swift | 2 +- .../swift/swiftkit/core/SwiftAsyncStream.java | 146 ++++++++++++++++++ .../JNI/JNIAsyncStreamTests.swift | 107 +++++++++++++ 12 files changed, 567 insertions(+), 3 deletions(-) create mode 100644 Samples/SwiftJavaExtractJNISampleApp/Sources/MySwiftLibrary/AsyncStreams.swift create mode 100644 Samples/SwiftJavaExtractJNISampleApp/src/test/java/com/example/swift/AsyncStreamTest.java create mode 100644 Sources/SwiftJavaRuntimeSupport/AsyncStreamBridge.swift create mode 100644 SwiftKitCore/src/main/java/org/swift/swiftkit/core/SwiftAsyncStream.java create mode 100644 Tests/JExtractSwiftTests/JNI/JNIAsyncStreamTests.swift diff --git a/Samples/SwiftJavaExtractJNISampleApp/Sources/MySwiftLibrary/AsyncStreams.swift b/Samples/SwiftJavaExtractJNISampleApp/Sources/MySwiftLibrary/AsyncStreams.swift new file mode 100644 index 000000000..9c79fdea0 --- /dev/null +++ b/Samples/SwiftJavaExtractJNISampleApp/Sources/MySwiftLibrary/AsyncStreams.swift @@ -0,0 +1,32 @@ +//===----------------------------------------------------------------------===// +// +// This source file is part of the Swift.org open source project +// +// Copyright (c) 2026 Apple Inc. and the Swift.org project authors +// Licensed under Apache License v2.0 +// +// See LICENSE.txt for license information +// See CONTRIBUTORS.txt for the list of Swift.org project authors +// +// SPDX-License-Identifier: Apache-2.0 +// +//===----------------------------------------------------------------------===// + +/// `AsyncStream` results are returned to Java as a `SwiftAsyncStream`. +public func countdown(from start: Int64) -> AsyncStream { + AsyncStream { continuation in + for value in stride(from: start, through: 0, by: -1) { + continuation.yield(value) + } + continuation.finish() + } +} + +public func shapes(count: Int64) -> AsyncStream { + AsyncStream { continuation in + for side in 1...max(count, 1) { + continuation.yield(Square(side: Double(side))) + } + continuation.finish() + } +} diff --git a/Samples/SwiftJavaExtractJNISampleApp/src/test/java/com/example/swift/AsyncStreamTest.java b/Samples/SwiftJavaExtractJNISampleApp/src/test/java/com/example/swift/AsyncStreamTest.java new file mode 100644 index 000000000..580aafd33 --- /dev/null +++ b/Samples/SwiftJavaExtractJNISampleApp/src/test/java/com/example/swift/AsyncStreamTest.java @@ -0,0 +1,68 @@ +//===----------------------------------------------------------------------===// +// +// This source file is part of the Swift.org open source project +// +// Copyright (c) 2024 Apple Inc. and the Swift.org project authors +// Licensed under Apache License v2.0 +// +// See LICENSE.txt for license information +// See CONTRIBUTORS.txt for the list of Swift.org project authors +// +// SPDX-License-Identifier: Apache-2.0 +// +//===----------------------------------------------------------------------===// + +package com.example.swift; + +import org.junit.jupiter.api.Test; +import org.swift.swiftkit.core.SwiftArena; +import org.swift.swiftkit.core.SwiftAsyncStream; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.List; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; + +import static org.junit.jupiter.api.Assertions.*; + +public class AsyncStreamTest { + static List collect(SwiftAsyncStream stream) throws InterruptedException { + List values = Collections.synchronizedList(new ArrayList<>()); + CountDownLatch done = new CountDownLatch(1); + stream.subscribe(new SwiftAsyncStream.Subscriber() { + @Override + public void onNext(T value) { + values.add(value); + } + + @Override + public void onComplete() { + done.countDown(); + } + }); + assertTrue(done.await(5, TimeUnit.SECONDS)); + return values; + } + + @Test + void primitiveElements() throws Exception { + assertEquals(List.of(3L, 2L, 1L, 0L), collect(MySwiftLibrary.countdown(3))); + } + + @Test + void jextractedElements() throws Exception { + try (var arena = SwiftArena.ofConfined()) { + List squares = collect(MySwiftLibrary.shapes(2, arena)); + assertEquals(2, squares.size()); + assertEquals(4.0, squares.get(1).area()); + } + } + + @Test + void canOnlyBeSubscribedOnce() throws Exception { + SwiftAsyncStream stream = MySwiftLibrary.countdown(1); + collect(stream); + assertThrows(IllegalStateException.class, () -> stream.subscribe(value -> {})); + } +} diff --git a/Sources/JExtractSwiftLib/JNI/JNIJavaTypeTranslator.swift b/Sources/JExtractSwiftLib/JNI/JNIJavaTypeTranslator.swift index 2019ba48a..d52823d30 100644 --- a/Sources/JExtractSwiftLib/JNI/JNIJavaTypeTranslator.swift +++ b/Sources/JExtractSwiftLib/JNI/JNIJavaTypeTranslator.swift @@ -53,6 +53,7 @@ enum JNIJavaTypeTranslator { .array, .dictionary, .set, + .asyncStream, .foundationDate, .essentialsDate, .foundationUUID, .essentialsUUID, .foundationURL, .essentialsURL: @@ -81,6 +82,7 @@ enum JNIJavaTypeTranslator { .array, .dictionary, .set, + .asyncStream, .foundationDate, .essentialsDate, .foundationUUID, .essentialsUUID, .foundationURL, .essentialsURL: @@ -109,6 +111,7 @@ enum JNIJavaTypeTranslator { .array, .dictionary, .set, + .asyncStream, .foundationDate, .essentialsDate, .foundationUUID, .essentialsUUID, .foundationURL, .essentialsURL: diff --git a/Sources/JExtractSwiftLib/JNI/JNISwift2JavaGenerator+JavaTranslation.swift b/Sources/JExtractSwiftLib/JNI/JNISwift2JavaGenerator+JavaTranslation.swift index bea4777ab..6570958a8 100644 --- a/Sources/JExtractSwiftLib/JNI/JNISwift2JavaGenerator+JavaTranslation.swift +++ b/Sources/JExtractSwiftLib/JNI/JNISwift2JavaGenerator+JavaTranslation.swift @@ -227,15 +227,25 @@ extension JNISwift2JavaGenerator { // Name. let javaName = javaIdentifiers.makeJavaMethodName(decl) + // A function returning `AsyncStream` is translated as returning `T`, then converted to a stream. + var resultSignature = decl.functionSignature + let streamElementType = Self.asyncStreamElementType(decl.functionSignature.result.type) + if let streamElementType { + if decl.functionSignature.isAsync || decl.functionSignature.isIsolated || decl.functionSignature.isImplicitlyAsync { + throw JavaTranslationError.unsupportedSwiftType(decl.functionSignature.result.type) + } + resultSignature.result = SwiftResult(convention: .direct, type: streamElementType) + } + // Swift -> Java var translatedFunctionSignature = try self.translate( - functionSignature: decl.functionSignature, + functionSignature: resultSignature, methodName: javaName, parentName: parentName, ) // Java -> Java (native) var nativeFunctionSignature = try nativeTranslation.translate( - functionSignature: decl.functionSignature, + functionSignature: resultSignature, translatedFunctionSignature: translatedFunctionSignature, methodName: javaName, parentName: parentName, @@ -272,6 +282,13 @@ extension JNISwift2JavaGenerator { } } + if streamElementType != nil { + try self.convertToAsyncStream( + translatedFunctionSignature: &translatedFunctionSignature, + nativeFunctionSignature: &nativeFunctionSignature + ) + } + // Handle async methods and isolated methods if decl.functionSignature.isAsync || decl.functionSignature.isIsolated || decl.functionSignature.isImplicitlyAsync @@ -708,6 +725,54 @@ extension JNISwift2JavaGenerator { ) } + /// The element type of an `AsyncStream`, or nil. + static func asyncStreamElementType(_ type: SwiftType) -> SwiftType? { + guard case .nominal(let nominal) = type, case .asyncStream(let element) = nominal.asKnownType else { + return nil + } + return element + } + + /// Turns a function translated as returning `T` into one returning a `SwiftAsyncStream`: Java + /// passes a stream for Swift to attach the `AsyncStream` to, and maps each delivered element + /// with the conversion it would apply to a `T` result. + func convertToAsyncStream( + translatedFunctionSignature: inout TranslatedFunctionSignature, + nativeFunctionSignature: inout NativeFunctionSignature + ) throws { + let element = translatedFunctionSignature.result + let nativeElement = nativeFunctionSignature.result + // Elements arrive one at a time as a single Java value; indirect returns (out parameters) don't fit. + guard element.outParameters.isEmpty, nativeElement.outParameters.isEmpty, !nativeElement.javaType.isVoid else { + throw JavaTranslationError.asyncStreamElementNotSupported + } + let nativeStreamType = JavaType.swiftAsyncStream(nativeElement.javaType) + + translatedFunctionSignature.result = TranslatedResult( + javaType: .swiftAsyncStream(element.javaType), + nativeJavaType: .void, + annotations: element.annotations, + outParameters: [OutParameter(name: "stream$", type: nativeStreamType, allocation: .new)], + conversion: .method( + .constant("stream$"), + function: "map", + arguments: [ + .lambda( + args: ["element$"], + body: .replacingPlaceholder(element.conversion, placeholder: "element$") + ) + ] + ) + ) + + nativeFunctionSignature.result.conversion = .attachAsyncStream( + elementConversion: nativeElement.conversion, + elementJavaType: nativeElement.javaType + ) + nativeFunctionSignature.result.javaType = .void + nativeFunctionSignature.result.outParameters.append(.init(name: "result_stream", type: nativeStreamType)) + } + func convertToAsync( translatedFunctionSignature: inout TranslatedFunctionSignature, nativeFunctionSignature: inout NativeFunctionSignature, @@ -2287,6 +2352,9 @@ extension JNISwift2JavaGenerator { /// Java can implement `async` protocol requirements only in `completableFuture` async mode. case asyncRequirementNeedsCompletableFuture + /// `AsyncStream` elements must convert to a single Java value. + case asyncStreamElementNotSupported + // FIXME: Remove once we support protocol variables case protocolVariablesNotSupported diff --git a/Sources/JExtractSwiftLib/JNI/JNISwift2JavaGenerator+NativeTranslation.swift b/Sources/JExtractSwiftLib/JNI/JNISwift2JavaGenerator+NativeTranslation.swift index 3e6cb2223..0cd6f88e0 100644 --- a/Sources/JExtractSwiftLib/JNI/JNISwift2JavaGenerator+NativeTranslation.swift +++ b/Sources/JExtractSwiftLib/JNI/JNISwift2JavaGenerator+NativeTranslation.swift @@ -1460,6 +1460,10 @@ extension JNISwift2JavaGenerator { indirect case unwrapOptional(NativeSwiftConversionStep, name: String, fatalErrorMessage: String) + /// Attaches the `AsyncStream` returned by the call to the Java `SwiftAsyncStream` passed in as + /// `result_stream`, converting each element with `elementConversion` when it is delivered. + indirect case attachAsyncStream(elementConversion: NativeSwiftConversionStep, elementJavaType: JavaType) + indirect case asyncCompleteFuture( swiftFunctionResultType: SwiftType, nativeFunctionSignature: NativeFunctionSignature, @@ -1964,6 +1968,22 @@ extension JNISwift2JavaGenerator { ) return unwrappedName + case .attachAsyncStream(let elementConversion, let elementJavaType): + var elementPrinter = SwiftPrinter() + let converted = elementConversion.render(&elementPrinter, "element$") + let value = + elementJavaType.requiresBoxing + ? "SwiftJavaRuntimeSupport._JNIBoxedConversions.box(\(converted), in: environment)" + : converted + printer.print( + """ + SwiftJavaRuntimeSupport._attachAsyncStream(\(placeholder), to: result_stream, environment: environment) { element$, environment in + \(elementPrinter.finalize())return \(value) + } + """ + ) + return "" + case .asyncCompleteFuture( let swiftFunctionResultType, let nativeFunctionSignature, diff --git a/Sources/JExtractSwiftLib/JavaTypes/JavaType+Futures.swift b/Sources/JExtractSwiftLib/JavaTypes/JavaType+Futures.swift index 3492a3980..a1bc05053 100644 --- a/Sources/JExtractSwiftLib/JavaTypes/JavaType+Futures.swift +++ b/Sources/JExtractSwiftLib/JavaTypes/JavaType+Futures.swift @@ -20,4 +20,9 @@ extension JavaType { static func simpleCompletableFuture(_ T: JavaType) -> JavaType { .class(package: "org.swift.swiftkit.core", name: "SimpleCompletableFuture", typeParameters: [T.boxedType]) } + + /// The description of the type org.swift.swiftkit.core.SwiftAsyncStream + static func swiftAsyncStream(_ T: JavaType) -> JavaType { + .class(package: "org.swift.swiftkit.core", name: "SwiftAsyncStream", typeParameters: [T.boxedType]) + } } diff --git a/Sources/SwiftExtract/SwiftTypes/SwiftKnownModules.swift b/Sources/SwiftExtract/SwiftTypes/SwiftKnownModules.swift index d3cf9ffa8..37e074f2d 100644 --- a/Sources/SwiftExtract/SwiftTypes/SwiftKnownModules.swift +++ b/Sources/SwiftExtract/SwiftTypes/SwiftKnownModules.swift @@ -104,6 +104,9 @@ private let swiftSourceFile: SourceFileSyntax = """ public struct Set {} + // `_Concurrency` in reality; modeled here with the other standard library types. + public struct AsyncStream {} + public protocol Error {} // FIXME: Support 'typealias Void = ()' diff --git a/Sources/SwiftExtract/SwiftTypes/SwiftKnownTypeDecls.swift b/Sources/SwiftExtract/SwiftTypes/SwiftKnownTypeDecls.swift index bcc167f03..eb351c1b9 100644 --- a/Sources/SwiftExtract/SwiftTypes/SwiftKnownTypeDecls.swift +++ b/Sources/SwiftExtract/SwiftTypes/SwiftKnownTypeDecls.swift @@ -43,6 +43,7 @@ public enum SwiftKnownType: Equatable { case array(_ element: SwiftType) case dictionary(_ key: SwiftType, _ value: SwiftType) case set(_ element: SwiftType) + case asyncStream(_ element: SwiftType) // Foundation case foundationDataProtocol @@ -102,6 +103,9 @@ public enum SwiftKnownType: Equatable { case .set: guard let arg = genericArguments?.first else { return nil } self = .set(arg) + case .asyncStream: + guard let arg = genericArguments?.first else { return nil } + self = .asyncStream(arg) case .foundationDataProtocol: self = .foundationDataProtocol case .essentialsDataProtocol: self = .essentialsDataProtocol case .foundationData: self = .foundationData @@ -145,6 +149,7 @@ public enum SwiftKnownType: Equatable { case .array: .array case .dictionary: .dictionary case .set: .set + case .asyncStream: .asyncStream case .foundationDataProtocol: .foundationDataProtocol case .essentialsDataProtocol: .essentialsDataProtocol case .foundationData: .foundationData @@ -189,6 +194,7 @@ public enum SwiftKnownTypeDeclKind: String, Hashable { case array = "Swift.Array" case dictionary = "Swift.Dictionary" case set = "Swift.Set" + case asyncStream = "Swift.AsyncStream" // Foundation case foundationDataProtocol = "Foundation.DataProtocol" diff --git a/Sources/SwiftJavaRuntimeSupport/AsyncStreamBridge.swift b/Sources/SwiftJavaRuntimeSupport/AsyncStreamBridge.swift new file mode 100644 index 000000000..71be72a05 --- /dev/null +++ b/Sources/SwiftJavaRuntimeSupport/AsyncStreamBridge.swift @@ -0,0 +1,106 @@ +//===----------------------------------------------------------------------===// +// +// This source file is part of the Swift.org open source project +// +// Copyright (c) 2026 Apple Inc. and the Swift.org project authors +// Licensed under Apache License v2.0 +// +// See LICENSE.txt for license information +// See CONTRIBUTORS.txt for the list of Swift.org project authors +// +// SPDX-License-Identifier: Apache-2.0 +// +//===----------------------------------------------------------------------===// + +import SwiftJava +import SwiftJavaJNICore + +@JavaClass("org.swift.swiftkit.core.SwiftAsyncStream") +class JavaSwiftAsyncStream: JavaObject { + @JavaMethod + func attachSource(_ handle: Int64) +} + +@JavaClass("org.swift.swiftkit.core.SwiftAsyncStream$Delivery") +class JavaSwiftAsyncStreamDelivery: JavaObject { + @JavaMethod + func next(_ value: JavaObject?) + + @JavaMethod + func finish(_ completed: Bool) +} + +/// A Swift `AsyncStream` handed to Java, waiting for its single subscriber. +final class SwiftAsyncStreamSource: @unchecked Sendable { + /// Starts iterating, delivering to the Java `Delivery` at the given global reference. + let subscribe: @Sendable (_ delivery: jobject) -> Task + + init(subscribe: @escaping @Sendable (_ delivery: jobject) -> Task) { + self.subscribe = subscribe + } +} + +/// Attaches `stream` to the Java `SwiftAsyncStream` the native call received. Each element is +/// converted with `convert` (to the Java value the generated `map` expects) when delivered. +/// +/// Used by generated code for functions returning `AsyncStream`. A stream that is never +/// subscribed to is not released. +public func _attachAsyncStream( + _ stream: AsyncStream, + to javaStream: jobject?, + environment: JNIEnvironment, + convert: @escaping @Sendable (Element, JNIEnvironment) -> jobject? +) { + guard let javaStream else { return } + nonisolated(unsafe) let stream = stream + let source = SwiftAsyncStreamSource { delivery in + nonisolated(unsafe) let delivery = delivery + return Task { + for await element in stream { + let environment = try! JavaVirtualMachine.shared().environment() + let value = convert(element, environment) + JavaSwiftAsyncStreamDelivery(javaThis: delivery, environment: environment) + .next(value.map { JavaObject(javaThis: $0, environment: environment) }) + if let value { + environment.interface.DeleteLocalRef(environment, value) + } + } + let environment = try! JavaVirtualMachine.shared().environment() + JavaSwiftAsyncStreamDelivery(javaThis: delivery, environment: environment).finish(!Task.isCancelled) + environment.interface.DeleteGlobalRef(environment, delivery) + } + } + let handle = Unmanaged.passRetained(source).toOpaque() + JavaSwiftAsyncStream(javaThis: javaStream, environment: environment).attachSource(Int64(Int(bitPattern: handle))) +} + +@_cdecl("Java_org_swift_swiftkit_core_SwiftAsyncStream__00024subscribe") +public func Java_org_swift_swiftkit_core_SwiftAsyncStream__00024subscribe( + environment: UnsafeMutablePointer!, + thisClass: jclass, + source: jlong, + delivery: jobject? +) -> jlong { + // The source is consumed: an AsyncStream has a single consumer. + let source = Unmanaged.fromOpaque(UnsafeRawPointer(bitPattern: Int(source))!).takeRetainedValue() + let task = source.subscribe(environment.interface.NewGlobalRef(environment, delivery)!) + return Int64(Int(bitPattern: Unmanaged.passRetained(SwiftTaskHandle(task)).toOpaque())) +} + +@_cdecl("Java_org_swift_swiftkit_core_SwiftAsyncStream__00024cancelTask") +public func Java_org_swift_swiftkit_core_SwiftAsyncStream__00024cancelTask( + environment: UnsafeMutablePointer!, + thisClass: jclass, + task: jlong +) { + Unmanaged.fromOpaque(UnsafeRawPointer(bitPattern: Int(task))!).takeUnretainedValue().task.cancel() +} + +@_cdecl("Java_org_swift_swiftkit_core_SwiftAsyncStream__00024releaseTask") +public func Java_org_swift_swiftkit_core_SwiftAsyncStream__00024releaseTask( + environment: UnsafeMutablePointer!, + thisClass: jclass, + task: jlong +) { + Unmanaged.fromOpaque(UnsafeRawPointer(bitPattern: Int(task))!).release() +} diff --git a/Sources/SwiftJavaRuntimeSupport/CompletableFutureAwait.swift b/Sources/SwiftJavaRuntimeSupport/CompletableFutureAwait.swift index fc717c177..0a03e3b7a 100644 --- a/Sources/SwiftJavaRuntimeSupport/CompletableFutureAwait.swift +++ b/Sources/SwiftJavaRuntimeSupport/CompletableFutureAwait.swift @@ -114,7 +114,7 @@ extension JavaClass { func cancelTaskWhenCancelled(_ future: JavaCompletableFuture?, _ task: Int64) } -private final class SwiftTaskHandle: @unchecked Sendable { +final class SwiftTaskHandle: @unchecked Sendable { let task: Task init(_ task: Task) { diff --git a/SwiftKitCore/src/main/java/org/swift/swiftkit/core/SwiftAsyncStream.java b/SwiftKitCore/src/main/java/org/swift/swiftkit/core/SwiftAsyncStream.java new file mode 100644 index 000000000..2a11e1942 --- /dev/null +++ b/SwiftKitCore/src/main/java/org/swift/swiftkit/core/SwiftAsyncStream.java @@ -0,0 +1,146 @@ +//===----------------------------------------------------------------------===// +// +// This source file is part of the Swift.org open source project +// +// Copyright (c) 2026 Apple Inc. and the Swift.org project authors +// Licensed under Apache License v2.0 +// +// See LICENSE.txt for license information +// See CONTRIBUTORS.txt for the list of Swift.org project authors +// +// SPDX-License-Identifier: Apache-2.0 +// +//===----------------------------------------------------------------------===// + +package org.swift.swiftkit.core; + +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Function; + +/** + * A Swift {@code AsyncStream} returned to Java. + *

+ * {@link #subscribe(Subscriber)} starts a Swift task that iterates the stream and delivers each + * element to the subscriber, without blocking a Java thread. An {@code AsyncStream} has a single + * consumer, so a stream (and any {@link #map mapped} view of it) can be subscribed to once. + * + * @param the element type + */ +public final class SwiftAsyncStream { + /** Receives the elements of a stream, on the thread of the Swift task iterating it. */ + public interface Subscriber { + void onNext(T value); + + /** The stream finished (it is not called when the subscription is cancelled). */ + default void onComplete() {} + } + + /** Cancels the Swift task iterating the stream; elements stop arriving. */ + public interface Subscription { + void cancel(); + } + + /** The Swift stream, shared by a stream and its mapped views until it is subscribed to. */ + private static final class Source { + private final AtomicLong handle = new AtomicLong(0); + } + + private final Source source; + private final Function mapper; + + @SuppressWarnings("unchecked") + public SwiftAsyncStream() { + this(new Source(), value -> (T) value); + } + + private SwiftAsyncStream(Source source, Function mapper) { + this.source = source; + this.mapper = mapper; + } + + /** Called from Swift with the handle of the Swift stream. */ + void attachSource(long handle) { + source.handle.set(handle); + } + + /** A view of this stream whose elements are transformed by {@code transform}. */ + public SwiftAsyncStream map(Function transform) { + return new SwiftAsyncStream<>(source, mapper.andThen(transform)); + } + + /** + * Starts iterating the Swift stream. + * + * @throws IllegalStateException if this stream was already subscribed to + */ + public Subscription subscribe(Subscriber subscriber) { + long handle = source.handle.getAndSet(0); + if (handle == 0) { + throw new IllegalStateException("A Swift AsyncStream can only be subscribed to once"); + } + Delivery delivery = new Delivery(value -> subscriber.onNext(mapper.apply(value)), subscriber::onComplete); + delivery.started($subscribe(handle, delivery)); + return delivery::cancel; + } + + /** Called by the Swift task; owns its task handle and releases it exactly once. */ + private static final class Delivery { + private final java.util.function.Consumer next; + private final Runnable complete; + private final AtomicLong task = new AtomicLong(0); + private final AtomicBoolean finished = new AtomicBoolean(false); + + Delivery(java.util.function.Consumer next, Runnable complete) { + this.next = next; + this.complete = complete; + } + + void next(Object value) { + next.accept(value); + } + + void finish(boolean completed) { + if (finished.compareAndSet(false, true)) { + release(); + if (completed) { + complete.run(); + } + } + } + + void started(long handle) { + task.set(handle); + // Finished or cancelled before $subscribe returned: cancel (harmless if already done) and release. + if (finished.get()) { + long started = task.getAndSet(0); + if (started != 0) { + $cancelTask(started); + $releaseTask(started); + } + } + } + + void cancel() { + long handle = task.getAndSet(0); + if (handle != 0) { + $cancelTask(handle); + $releaseTask(handle); + } + finished.set(true); + } + + private void release() { + long handle = task.getAndSet(0); + if (handle != 0) { + $releaseTask(handle); + } + } + } + + private static native long $subscribe(long source, Object delivery); + + private static native void $cancelTask(long task); + + private static native void $releaseTask(long task); +} diff --git a/Tests/JExtractSwiftTests/JNI/JNIAsyncStreamTests.swift b/Tests/JExtractSwiftTests/JNI/JNIAsyncStreamTests.swift new file mode 100644 index 000000000..c0cb783cf --- /dev/null +++ b/Tests/JExtractSwiftTests/JNI/JNIAsyncStreamTests.swift @@ -0,0 +1,107 @@ +//===----------------------------------------------------------------------===// +// +// This source file is part of the Swift.org open source project +// +// Copyright (c) 2026 Apple Inc. and the Swift.org project authors +// Licensed under Apache License v2.0 +// +// See LICENSE.txt for license information +// See CONTRIBUTORS.txt for the list of Swift.org project authors +// +// SPDX-License-Identifier: Apache-2.0 +// +//===----------------------------------------------------------------------===// + +import CodePrinting +import JExtractSwiftLib +import SwiftExtract +import SwiftJavaConfigurationShared +import Testing + +/// Functions returning `AsyncStream` return a `SwiftAsyncStream` to Java. +@Suite +struct JNIAsyncStreamTests { + @Test + func javaReturnsSwiftAsyncStream() throws { + try assertOutput( + input: "public func ticks(count: Int64) -> AsyncStream { fatalError() }", + .jni, + .java, + detectChunkByInitialLines: 1, + expectedChunks: [ + """ + public static org.swift.swiftkit.core.SwiftAsyncStream ticks(long count) { + org.swift.swiftkit.core.SwiftAsyncStream stream$ = new org.swift.swiftkit.core.SwiftAsyncStream(); + SwiftModule.$ticks(count, stream$); + return stream$.map((element$) -> { + return element$; + } + ); + } + """, + "private static native void $ticks(long count, org.swift.swiftkit.core.SwiftAsyncStream result_stream);", + ] + ) + } + + @Test + func swiftAttachesTheStreamWithAnElementConverter() throws { + try assertOutput( + input: "public func ticks(count: Int64) -> AsyncStream { fatalError() }", + .jni, + .swift, + detectChunkByInitialLines: 1, + expectedChunks: [ + """ + SwiftJavaRuntimeSupport._attachAsyncStream(SwiftModule.ticks(count: Int64(fromJNI: count, in: environment)), to: result_stream, environment: environment) { element$, environment in + return SwiftJavaRuntimeSupport._JNIBoxedConversions.box(element$.getJNILocalRefValue(in: environment), in: environment) + } + """ + ] + ) + } + + @Test + func jextractedElementsAreMappedToTheirJavaClass() throws { + try assertOutput( + input: """ + public struct Tick {} + public func ticks() -> AsyncStream { fatalError() } + """, + .jni, + .java, + detectChunkByInitialLines: 1, + expectedChunks: [ + "public static org.swift.swiftkit.core.SwiftAsyncStream ticks(SwiftArena swiftArena) {", + "return Tick.wrapMemoryAddressUnsafe(element$, swiftArena);", + ] + ) + } + + @Test + func unsupportedStreamsAreSkipped() throws { + var config = Configuration() + config.swiftModule = "SwiftModule" + let translator = makeSwiftJavaAnalyzer(config: config) + try translator.analyze( + path: "/fake/Fake.swiftinterface", + text: """ + public func later() async -> AsyncStream { fatalError() } + public func pairs() -> AsyncStream<(Int64, String)> { fatalError() } + """ + ) + let generator = JNISwift2JavaGenerator( + config: config, + translator: translator, + javaPackage: "com.example.swift", + swiftOutputDirectory: "/fake", + javaOutputDirectory: "/fake", + javaClassLookupTable: [:], + moduleJavaPackages: [:] + ) + var printer = JavaPrinter(mode: .accumulateAll) + try generator.writeExportedJavaSources(&printer) + let skipped = generator.skippedDeclarations.map { String($0.prefix { $0 != ":" }) }.sorted() + #expect(skipped == ["later()", "pairs()"], "\(generator.skippedDeclarations)") + } +} From d83f8f365d7f0583c55108edfe4218730225fc39 Mon Sep 17 00:00:00 2001 From: Pierluigi Cifani Date: Tue, 3 Nov 2026 16:52:37 +0100 Subject: [PATCH 2/3] Samples: a Swift AsyncStream collected as a Kotlin Flow Co-Authored-By: Claude Opus 5.5 --- .../Sources/KotlinSampleLibrary/Streams.swift | 55 +++++++++++++++++ .../kotlin/com/example/kotlin/StreamsTest.kt | 60 +++++++++++++++++++ 2 files changed, 115 insertions(+) create mode 100644 Samples/SwiftJavaExtractJNIKotlinSampleApp/Sources/KotlinSampleLibrary/Streams.swift create mode 100644 Samples/SwiftJavaExtractJNIKotlinSampleApp/src/test/kotlin/com/example/kotlin/StreamsTest.kt diff --git a/Samples/SwiftJavaExtractJNIKotlinSampleApp/Sources/KotlinSampleLibrary/Streams.swift b/Samples/SwiftJavaExtractJNIKotlinSampleApp/Sources/KotlinSampleLibrary/Streams.swift new file mode 100644 index 000000000..20d071f48 --- /dev/null +++ b/Samples/SwiftJavaExtractJNIKotlinSampleApp/Sources/KotlinSampleLibrary/Streams.swift @@ -0,0 +1,55 @@ +//===----------------------------------------------------------------------===// +// +// This source file is part of the Swift.org open source project +// +// Copyright (c) 2026 Apple Inc. and the Swift.org project authors +// Licensed under Apache License v2.0 +// +// See LICENSE.txt for license information +// See CONTRIBUTORS.txt for the list of Swift.org project authors +// +// SPDX-License-Identifier: Apache-2.0 +// +//===----------------------------------------------------------------------===// + +import Synchronization + +/// A view model's stream of updates, collected in Kotlin as a Flow. +public final class PriceFeed: Sendable { + private let stopped = Mutex(false) + + public init() {} + + /// A fixed series of prices. + public func prices(count: Int64) -> AsyncStream { + AsyncStream { continuation in + for i in 0.. AsyncStream { + AsyncStream { continuation in + let task = Task { + var i = 0 + while !Task.isCancelled { + continuation.yield("update \(i)") + i += 1 + try? await Task.sleep(for: .milliseconds(10)) + } + continuation.finish() + } + continuation.onTermination = { _ in + task.cancel() + self.stopped.withLock { $0 = true } + } + } + } + + public func wasStopped() -> Bool { + stopped.withLock { $0 } + } +} diff --git a/Samples/SwiftJavaExtractJNIKotlinSampleApp/src/test/kotlin/com/example/kotlin/StreamsTest.kt b/Samples/SwiftJavaExtractJNIKotlinSampleApp/src/test/kotlin/com/example/kotlin/StreamsTest.kt new file mode 100644 index 000000000..f9b8e7910 --- /dev/null +++ b/Samples/SwiftJavaExtractJNIKotlinSampleApp/src/test/kotlin/com/example/kotlin/StreamsTest.kt @@ -0,0 +1,60 @@ +//===----------------------------------------------------------------------===// +// +// This source file is part of the Swift.org open source project +// +// Copyright (c) 2026 Apple Inc. and the Swift.org project authors +// Licensed under Apache License v2.0 +// +// See LICENSE.txt for license information +// See CONTRIBUTORS.txt for the list of Swift.org project authors +// +// SPDX-License-Identifier: Apache-2.0 +// +//===----------------------------------------------------------------------===// + +package com.example.kotlin + +import kotlinx.coroutines.channels.awaitClose +import kotlinx.coroutines.channels.trySendBlocking +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.callbackFlow +import kotlinx.coroutines.flow.map +import kotlinx.coroutines.flow.take +import kotlinx.coroutines.flow.toList +import kotlinx.coroutines.runBlocking +import org.swift.swiftkit.core.SwiftAsyncStream +import kotlin.test.Test +import kotlin.test.assertEquals +import kotlin.test.assertTrue + +/** A Swift AsyncStream as a Kotlin Flow: cancelling the collector cancels the Swift task. */ +fun SwiftAsyncStream.asFlow(): Flow = callbackFlow { + val subscription = subscribe(object : SwiftAsyncStream.Subscriber { + override fun onNext(value: T) { + trySendBlocking(value) + } + + override fun onComplete() { + close() + } + }) + awaitClose { subscription.cancel() } +} + +class StreamsTest { + @Test + fun collectAsAFlow() = runBlocking { + val ids = PriceFeed.init().prices(3).asFlow().map { it.id }.toList() + assertEquals(listOf("price-0", "price-1", "price-2"), ids) + } + + @Test + fun cancellingTheFlowStopsTheSwiftStream() = runBlocking { + val feed = PriceFeed.init() + assertEquals(listOf("update 0", "update 1"), feed.live().asFlow().take(2).toList()) + + val deadline = System.currentTimeMillis() + 5_000 + while (!feed.wasStopped() && System.currentTimeMillis() < deadline) Thread.sleep(20) + assertTrue(feed.wasStopped()) + } +} From 971543b0863171eb90887da6bc5d426b29330071 Mon Sep 17 00:00:00 2001 From: Pierluigi Cifani Date: Tue, 3 Nov 2026 17:07:08 +0100 Subject: [PATCH 3/3] SwiftKitKotlin: SwiftAsyncStream as a Kotlin Flow A new `swiftkit-kotlin` artifact with `SwiftAsyncStream.asFlow()`, so Kotlin callers collect Swift streams as a `Flow` without writing the adapter themselves. It lives outside SwiftKitCore, which stays a plain Java library with no kotlinx-coroutines dependency. Targets Java 11 like SwiftKitCore. The Kotlin sample now uses it instead of its own copy. Co-Authored-By: Claude Opus 5.5 --- .../build.gradle.kts | 2 +- .../kotlin/com/example/kotlin/StreamsTest.kt | 20 +------ SwiftKitKotlin/build.gradle.kts | 59 +++++++++++++++++++ .../swiftkit/kotlin/SwiftAsyncStreamFlow.kt | 40 +++++++++++++ settings.gradle.kts | 1 + 5 files changed, 102 insertions(+), 20 deletions(-) create mode 100644 SwiftKitKotlin/build.gradle.kts create mode 100644 SwiftKitKotlin/src/main/kotlin/org/swift/swiftkit/kotlin/SwiftAsyncStreamFlow.kt diff --git a/Samples/SwiftJavaExtractJNIKotlinSampleApp/build.gradle.kts b/Samples/SwiftJavaExtractJNIKotlinSampleApp/build.gradle.kts index 349cdd89f..d59fed476 100644 --- a/Samples/SwiftJavaExtractJNIKotlinSampleApp/build.gradle.kts +++ b/Samples/SwiftJavaExtractJNIKotlinSampleApp/build.gradle.kts @@ -61,9 +61,9 @@ tasks.build { dependencies { implementation(projects.swiftKitCore) + implementation(projects.swiftKitKotlin) testImplementation(kotlin("test-junit5")) - testImplementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.11.0") testRuntimeOnly(libs.junit.platform.launcher) // TODO: workaround for not finding junit: https://github.com/gradle/gradle/issues/34512 } diff --git a/Samples/SwiftJavaExtractJNIKotlinSampleApp/src/test/kotlin/com/example/kotlin/StreamsTest.kt b/Samples/SwiftJavaExtractJNIKotlinSampleApp/src/test/kotlin/com/example/kotlin/StreamsTest.kt index f9b8e7910..ecbd6bb77 100644 --- a/Samples/SwiftJavaExtractJNIKotlinSampleApp/src/test/kotlin/com/example/kotlin/StreamsTest.kt +++ b/Samples/SwiftJavaExtractJNIKotlinSampleApp/src/test/kotlin/com/example/kotlin/StreamsTest.kt @@ -14,33 +14,15 @@ package com.example.kotlin -import kotlinx.coroutines.channels.awaitClose -import kotlinx.coroutines.channels.trySendBlocking -import kotlinx.coroutines.flow.Flow -import kotlinx.coroutines.flow.callbackFlow import kotlinx.coroutines.flow.map import kotlinx.coroutines.flow.take import kotlinx.coroutines.flow.toList import kotlinx.coroutines.runBlocking -import org.swift.swiftkit.core.SwiftAsyncStream +import org.swift.swiftkit.kotlin.asFlow import kotlin.test.Test import kotlin.test.assertEquals import kotlin.test.assertTrue -/** A Swift AsyncStream as a Kotlin Flow: cancelling the collector cancels the Swift task. */ -fun SwiftAsyncStream.asFlow(): Flow = callbackFlow { - val subscription = subscribe(object : SwiftAsyncStream.Subscriber { - override fun onNext(value: T) { - trySendBlocking(value) - } - - override fun onComplete() { - close() - } - }) - awaitClose { subscription.cancel() } -} - class StreamsTest { @Test fun collectAsAFlow() = runBlocking { diff --git a/SwiftKitKotlin/build.gradle.kts b/SwiftKitKotlin/build.gradle.kts new file mode 100644 index 000000000..b439ea368 --- /dev/null +++ b/SwiftKitKotlin/build.gradle.kts @@ -0,0 +1,59 @@ +//===----------------------------------------------------------------------===// +// +// This source file is part of the Swift.org open source project +// +// Copyright (c) 2026 Apple Inc. and the Swift.org project authors +// Licensed under Apache License v2.0 +// +// See LICENSE.txt for license information +// See CONTRIBUTORS.txt for the list of Swift.org project authors +// +// SPDX-License-Identifier: Apache-2.0 +// +//===----------------------------------------------------------------------===// + +import org.jetbrains.kotlin.gradle.dsl.JvmTarget +import org.swift.build.utils.resolveSwiftKitVersion + +// Kotlin adapters for SwiftKitCore types (for example SwiftAsyncStream as a Flow), kept out of +// SwiftKitCore so that it stays a plain Java library without a kotlinx-coroutines dependency. +plugins { + id("build-logic.java-common-conventions") + kotlin("jvm") version "2.4.21" + `maven-publish` +} + +group = "org.swift.swiftkit" + +version = resolveSwiftKitVersion() + +base { + archivesName = "swiftkit-kotlin" +} + +publishing { + publications { + create("maven") { + groupId = group as? String + artifactId = "swiftkit-kotlin" + from(components["java"]) + } + } +} + +// Consumable down to Java 11, like SwiftKitCore +tasks.withType().configureEach { + options.release.set(11) +} + +kotlin { + compilerOptions { + jvmTarget.set(JvmTarget.JVM_11) + freeCompilerArgs.add("-Xjdk-release=11") + } +} + +dependencies { + api(projects.swiftKitCore) + api("org.jetbrains.kotlinx:kotlinx-coroutines-core:1.11.0") +} diff --git a/SwiftKitKotlin/src/main/kotlin/org/swift/swiftkit/kotlin/SwiftAsyncStreamFlow.kt b/SwiftKitKotlin/src/main/kotlin/org/swift/swiftkit/kotlin/SwiftAsyncStreamFlow.kt new file mode 100644 index 000000000..2121f7586 --- /dev/null +++ b/SwiftKitKotlin/src/main/kotlin/org/swift/swiftkit/kotlin/SwiftAsyncStreamFlow.kt @@ -0,0 +1,40 @@ +//===----------------------------------------------------------------------===// +// +// This source file is part of the Swift.org open source project +// +// Copyright (c) 2026 Apple Inc. and the Swift.org project authors +// Licensed under Apache License v2.0 +// +// See LICENSE.txt for license information +// See CONTRIBUTORS.txt for the list of Swift.org project authors +// +// SPDX-License-Identifier: Apache-2.0 +// +//===----------------------------------------------------------------------===// + +package org.swift.swiftkit.kotlin + +import kotlinx.coroutines.channels.awaitClose +import kotlinx.coroutines.channels.trySendBlocking +import kotlinx.coroutines.flow.Flow +import kotlinx.coroutines.flow.callbackFlow +import org.swift.swiftkit.core.SwiftAsyncStream + +/** + * A Swift `AsyncStream` as a cold [Flow]. Collecting it subscribes to the stream, so like the + * stream it can be collected once; cancelling the collector cancels the Swift task iterating it. + * + * When the collector falls behind, the Swift task waits for buffer space before its next element. + */ +fun SwiftAsyncStream.asFlow(): Flow = callbackFlow { + val subscription = subscribe(object : SwiftAsyncStream.Subscriber { + override fun onNext(value: T) { + trySendBlocking(value) + } + + override fun onComplete() { + close() + } + }) + awaitClose { subscription.cancel() } +} diff --git a/settings.gradle.kts b/settings.gradle.kts index 1877b848e..44abbbfc7 100644 --- a/settings.gradle.kts +++ b/settings.gradle.kts @@ -69,6 +69,7 @@ val ffmModules = setOf( val skipped = mutableListOf() include("SwiftKitCore") +include("SwiftKitKotlin") if (ffmCapable) { include("SwiftKitFFM") } else {