From 6a64d71bca17770e6a194c4af41e3bf04c76a141 Mon Sep 17 00:00:00 2001 From: Pierluigi Cifani Date: Tue, 3 Nov 2026 17:22:02 +0100 Subject: [PATCH 1/2] JNI: AsyncThrowingStream results, with errors delivered to onError Functions returning `AsyncThrowingStream` were skipped: the type wasn't in the modeled standard library. - The generator treats them like `AsyncStream`: Java gets the same `SwiftAsyncStream` and the Swift thunk calls `_attachAsyncStream`, which gains an `AsyncThrowingStream` overload. - When the stream throws, the error is converted like any thrown Swift error and delivered to the new `Subscriber.onError` instead of `onComplete`. Neither is called after the subscription is cancelled. - SwiftKitKotlin's `asFlow()` closes the flow with that error, so `collect` throws it. Co-Authored-By: Claude Opus 5.5 --- .../Sources/MySwiftLibrary/AsyncStreams.swift | 14 +++++ .../com/example/swift/AsyncStreamTest.java | 29 +++++++++ .../JNI/JNIJavaTypeTranslator.swift | 6 +- ...ISwift2JavaGenerator+JavaTranslation.swift | 13 ++-- ...wift2JavaGenerator+NativeTranslation.swift | 2 +- .../SwiftTypes/SwiftKnownModules.swift | 1 + .../SwiftTypes/SwiftKnownTypeDecls.swift | 6 ++ .../AsyncStreamBridge.swift | 61 ++++++++++++++++--- .../swift/swiftkit/core/SwiftAsyncStream.java | 22 ++++++- .../swiftkit/kotlin/SwiftAsyncStreamFlow.kt | 9 ++- .../JNI/JNIAsyncStreamTests.swift | 26 +++++++- 11 files changed, 166 insertions(+), 23 deletions(-) diff --git a/Samples/SwiftJavaExtractJNISampleApp/Sources/MySwiftLibrary/AsyncStreams.swift b/Samples/SwiftJavaExtractJNISampleApp/Sources/MySwiftLibrary/AsyncStreams.swift index 9c79fdea0..74cfa4049 100644 --- a/Samples/SwiftJavaExtractJNISampleApp/Sources/MySwiftLibrary/AsyncStreams.swift +++ b/Samples/SwiftJavaExtractJNISampleApp/Sources/MySwiftLibrary/AsyncStreams.swift @@ -30,3 +30,17 @@ public func shapes(count: Int64) -> AsyncStream { continuation.finish() } } + +public struct CountdownInterrupted: Error, CustomStringConvertible { + public var description: String { "interrupted" } +} + +/// `AsyncThrowingStream` results too; an error thrown by the stream reaches `onError`. +public func interruptedCountdown(from start: Int64, failingAt failure: Int64) -> AsyncThrowingStream { + AsyncThrowingStream { continuation in + for value in stride(from: start, through: failure, by: -1) { + continuation.yield(value) + } + continuation.finish(throwing: CountdownInterrupted()) + } +} diff --git a/Samples/SwiftJavaExtractJNISampleApp/src/test/java/com/example/swift/AsyncStreamTest.java b/Samples/SwiftJavaExtractJNISampleApp/src/test/java/com/example/swift/AsyncStreamTest.java index 580aafd33..54bdd108f 100644 --- a/Samples/SwiftJavaExtractJNISampleApp/src/test/java/com/example/swift/AsyncStreamTest.java +++ b/Samples/SwiftJavaExtractJNISampleApp/src/test/java/com/example/swift/AsyncStreamTest.java @@ -23,6 +23,7 @@ import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; import static org.junit.jupiter.api.Assertions.*; @@ -59,6 +60,34 @@ void jextractedElements() throws Exception { } } + @Test + void throwingStreamDeliversItsError() throws Exception { + List values = Collections.synchronizedList(new ArrayList<>()); + AtomicReference error = new AtomicReference<>(); + CountDownLatch done = new CountDownLatch(1); + MySwiftLibrary.interruptedCountdown(3, 2).subscribe(new SwiftAsyncStream.Subscriber() { + @Override + public void onNext(Long value) { + values.add(value); + } + + @Override + public void onComplete() { + done.countDown(); + } + + @Override + public void onError(Throwable failure) { + error.set(failure); + done.countDown(); + } + }); + assertTrue(done.await(5, TimeUnit.SECONDS)); + assertEquals(List.of(3L, 2L), values); + assertNotNull(error.get()); + assertEquals("interrupted", error.get().getMessage()); + } + @Test void canOnlyBeSubscribedOnce() throws Exception { SwiftAsyncStream stream = MySwiftLibrary.countdown(1); diff --git a/Sources/JExtractSwiftLib/JNI/JNIJavaTypeTranslator.swift b/Sources/JExtractSwiftLib/JNI/JNIJavaTypeTranslator.swift index d52823d30..7761b4dff 100644 --- a/Sources/JExtractSwiftLib/JNI/JNIJavaTypeTranslator.swift +++ b/Sources/JExtractSwiftLib/JNI/JNIJavaTypeTranslator.swift @@ -53,7 +53,7 @@ enum JNIJavaTypeTranslator { .array, .dictionary, .set, - .asyncStream, + .asyncStream, .asyncThrowingStream, .foundationDate, .essentialsDate, .foundationUUID, .essentialsUUID, .foundationURL, .essentialsURL: @@ -82,7 +82,7 @@ enum JNIJavaTypeTranslator { .array, .dictionary, .set, - .asyncStream, + .asyncStream, .asyncThrowingStream, .foundationDate, .essentialsDate, .foundationUUID, .essentialsUUID, .foundationURL, .essentialsURL: @@ -111,7 +111,7 @@ enum JNIJavaTypeTranslator { .array, .dictionary, .set, - .asyncStream, + .asyncStream, .asyncThrowingStream, .foundationDate, .essentialsDate, .foundationUUID, .essentialsUUID, .foundationURL, .essentialsURL: diff --git a/Sources/JExtractSwiftLib/JNI/JNISwift2JavaGenerator+JavaTranslation.swift b/Sources/JExtractSwiftLib/JNI/JNISwift2JavaGenerator+JavaTranslation.swift index 6570958a8..e607a4483 100644 --- a/Sources/JExtractSwiftLib/JNI/JNISwift2JavaGenerator+JavaTranslation.swift +++ b/Sources/JExtractSwiftLib/JNI/JNISwift2JavaGenerator+JavaTranslation.swift @@ -227,7 +227,7 @@ extension JNISwift2JavaGenerator { // Name. let javaName = javaIdentifiers.makeJavaMethodName(decl) - // A function returning `AsyncStream` is translated as returning `T`, then converted to a stream. + // A function returning `AsyncStream` or `AsyncThrowingStream` 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 { @@ -725,12 +725,17 @@ extension JNISwift2JavaGenerator { ) } - /// The element type of an `AsyncStream`, or nil. + /// The element type of an `AsyncStream` or `AsyncThrowingStream`, or nil. static func asyncStreamElementType(_ type: SwiftType) -> SwiftType? { - guard case .nominal(let nominal) = type, case .asyncStream(let element) = nominal.asKnownType else { + guard case .nominal(let nominal) = type else { + return nil + } + switch nominal.asKnownType { + case .asyncStream(let element), .asyncThrowingStream(let element): + return element + default: return nil } - return element } /// Turns a function translated as returning `T` into one returning a `SwiftAsyncStream`: Java diff --git a/Sources/JExtractSwiftLib/JNI/JNISwift2JavaGenerator+NativeTranslation.swift b/Sources/JExtractSwiftLib/JNI/JNISwift2JavaGenerator+NativeTranslation.swift index 0cd6f88e0..47bda010a 100644 --- a/Sources/JExtractSwiftLib/JNI/JNISwift2JavaGenerator+NativeTranslation.swift +++ b/Sources/JExtractSwiftLib/JNI/JNISwift2JavaGenerator+NativeTranslation.swift @@ -1460,7 +1460,7 @@ extension JNISwift2JavaGenerator { indirect case unwrapOptional(NativeSwiftConversionStep, name: String, fatalErrorMessage: String) - /// Attaches the `AsyncStream` returned by the call to the Java `SwiftAsyncStream` passed in as + /// Attaches the `AsyncStream` (or `AsyncThrowingStream`) 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) diff --git a/Sources/SwiftExtract/SwiftTypes/SwiftKnownModules.swift b/Sources/SwiftExtract/SwiftTypes/SwiftKnownModules.swift index 37e074f2d..72739cf60 100644 --- a/Sources/SwiftExtract/SwiftTypes/SwiftKnownModules.swift +++ b/Sources/SwiftExtract/SwiftTypes/SwiftKnownModules.swift @@ -106,6 +106,7 @@ private let swiftSourceFile: SourceFileSyntax = """ // `_Concurrency` in reality; modeled here with the other standard library types. public struct AsyncStream {} + public struct AsyncThrowingStream {} public protocol Error {} diff --git a/Sources/SwiftExtract/SwiftTypes/SwiftKnownTypeDecls.swift b/Sources/SwiftExtract/SwiftTypes/SwiftKnownTypeDecls.swift index eb351c1b9..5ecc2cbd5 100644 --- a/Sources/SwiftExtract/SwiftTypes/SwiftKnownTypeDecls.swift +++ b/Sources/SwiftExtract/SwiftTypes/SwiftKnownTypeDecls.swift @@ -44,6 +44,7 @@ public enum SwiftKnownType: Equatable { case dictionary(_ key: SwiftType, _ value: SwiftType) case set(_ element: SwiftType) case asyncStream(_ element: SwiftType) + case asyncThrowingStream(_ element: SwiftType) // Foundation case foundationDataProtocol @@ -106,6 +107,9 @@ public enum SwiftKnownType: Equatable { case .asyncStream: guard let arg = genericArguments?.first else { return nil } self = .asyncStream(arg) + case .asyncThrowingStream: + guard let arg = genericArguments?.first else { return nil } + self = .asyncThrowingStream(arg) case .foundationDataProtocol: self = .foundationDataProtocol case .essentialsDataProtocol: self = .essentialsDataProtocol case .foundationData: self = .foundationData @@ -150,6 +154,7 @@ public enum SwiftKnownType: Equatable { case .dictionary: .dictionary case .set: .set case .asyncStream: .asyncStream + case .asyncThrowingStream: .asyncThrowingStream case .foundationDataProtocol: .foundationDataProtocol case .essentialsDataProtocol: .essentialsDataProtocol case .foundationData: .foundationData @@ -195,6 +200,7 @@ public enum SwiftKnownTypeDeclKind: String, Hashable { case dictionary = "Swift.Dictionary" case set = "Swift.Set" case asyncStream = "Swift.AsyncStream" + case asyncThrowingStream = "Swift.AsyncThrowingStream" // Foundation case foundationDataProtocol = "Foundation.DataProtocol" diff --git a/Sources/SwiftJavaRuntimeSupport/AsyncStreamBridge.swift b/Sources/SwiftJavaRuntimeSupport/AsyncStreamBridge.swift index 71be72a05..b8b7fe0e7 100644 --- a/Sources/SwiftJavaRuntimeSupport/AsyncStreamBridge.swift +++ b/Sources/SwiftJavaRuntimeSupport/AsyncStreamBridge.swift @@ -28,6 +28,9 @@ class JavaSwiftAsyncStreamDelivery: JavaObject { @JavaMethod func finish(_ completed: Bool) + + @JavaMethod + func fail(_ failure: Throwable?) } /// A Swift `AsyncStream` handed to Java, waiting for its single subscriber. @@ -51,22 +54,62 @@ public func _attachAsyncStream( environment: JNIEnvironment, convert: @escaping @Sendable (Element, JNIEnvironment) -> jobject? ) { - guard let javaStream else { return } nonisolated(unsafe) let stream = stream + attachSource(to: javaStream, environment: environment, convert: convert) { deliver in + for await element in stream { + deliver(element) + } + } +} + +/// Attaches a throwing stream like the `AsyncStream` overload; an error it throws is delivered to +/// the subscriber's `onError`. +public func _attachAsyncStream( + _ stream: AsyncThrowingStream, + to javaStream: jobject?, + environment: JNIEnvironment, + convert: @escaping @Sendable (Element, JNIEnvironment) -> jobject? +) { + nonisolated(unsafe) let stream = stream + attachSource(to: javaStream, environment: environment, convert: convert) { deliver in + for try await element in stream { + deliver(element) + } + } +} + +private func attachSource( + to javaStream: jobject?, + environment: JNIEnvironment, + convert: @escaping @Sendable (Element, JNIEnvironment) -> jobject?, + iterate: @escaping @Sendable (_ deliver: (Element) -> Void) async throws -> Void +) { + guard let javaStream else { return } 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) + var failure: (any Error)? + do { + try await iterate { element in + 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) + } } + } catch { + failure = error } let environment = try! JavaVirtualMachine.shared().environment() - JavaSwiftAsyncStreamDelivery(javaThis: delivery, environment: environment).finish(!Task.isCancelled) + let javaDelivery = JavaSwiftAsyncStreamDelivery(javaThis: delivery, environment: environment) + if let failure, !Task.isCancelled, let throwable = environment.makeJavaThrowable(for: failure) { + javaDelivery.fail(Throwable(javaThis: throwable, environment: environment)) + environment.interface.DeleteLocalRef(environment, throwable) + } else { + javaDelivery.finish(!Task.isCancelled) + } environment.interface.DeleteGlobalRef(environment, delivery) } } diff --git a/SwiftKitCore/src/main/java/org/swift/swiftkit/core/SwiftAsyncStream.java b/SwiftKitCore/src/main/java/org/swift/swiftkit/core/SwiftAsyncStream.java index 2a11e1942..82e169411 100644 --- a/SwiftKitCore/src/main/java/org/swift/swiftkit/core/SwiftAsyncStream.java +++ b/SwiftKitCore/src/main/java/org/swift/swiftkit/core/SwiftAsyncStream.java @@ -19,7 +19,7 @@ import java.util.function.Function; /** - * A Swift {@code AsyncStream} returned to Java. + * A Swift {@code AsyncStream} or {@code AsyncThrowingStream} 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 @@ -34,6 +34,12 @@ public interface Subscriber { /** The stream finished (it is not called when the subscription is cancelled). */ default void onComplete() {} + + /** + * A throwing stream failed with {@code error}, instead of finishing. It is not called when the + * subscription is cancelled. + */ + default void onError(Throwable error) {} } /** Cancels the Swift task iterating the stream; elements stop arriving. */ @@ -79,7 +85,8 @@ public Subscription subscribe(Subscriber subscriber) { 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 delivery = + new Delivery(value -> subscriber.onNext(mapper.apply(value)), subscriber::onComplete, subscriber::onError); delivery.started($subscribe(handle, delivery)); return delivery::cancel; } @@ -88,12 +95,14 @@ public Subscription subscribe(Subscriber subscriber) { private static final class Delivery { private final java.util.function.Consumer next; private final Runnable complete; + private final java.util.function.Consumer error; private final AtomicLong task = new AtomicLong(0); private final AtomicBoolean finished = new AtomicBoolean(false); - Delivery(java.util.function.Consumer next, Runnable complete) { + Delivery(java.util.function.Consumer next, Runnable complete, java.util.function.Consumer error) { this.next = next; this.complete = complete; + this.error = error; } void next(Object value) { @@ -109,6 +118,13 @@ void finish(boolean completed) { } } + void fail(Throwable failure) { + if (finished.compareAndSet(false, true)) { + release(); + error.accept(failure); + } + } + void started(long handle) { task.set(handle); // Finished or cancelled before $subscribe returned: cancel (harmless if already done) and release. diff --git a/SwiftKitKotlin/src/main/kotlin/org/swift/swiftkit/kotlin/SwiftAsyncStreamFlow.kt b/SwiftKitKotlin/src/main/kotlin/org/swift/swiftkit/kotlin/SwiftAsyncStreamFlow.kt index 2121f7586..2eed2e1d0 100644 --- a/SwiftKitKotlin/src/main/kotlin/org/swift/swiftkit/kotlin/SwiftAsyncStreamFlow.kt +++ b/SwiftKitKotlin/src/main/kotlin/org/swift/swiftkit/kotlin/SwiftAsyncStreamFlow.kt @@ -21,8 +21,9 @@ 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. + * A Swift `AsyncStream` or `AsyncThrowingStream` 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. An error thrown by the stream is thrown by `collect`. * * When the collector falls behind, the Swift task waits for buffer space before its next element. */ @@ -35,6 +36,10 @@ fun SwiftAsyncStream.asFlow(): Flow = callbackFlow { override fun onComplete() { close() } + + override fun onError(error: Throwable) { + close(error) + } }) awaitClose { subscription.cancel() } } diff --git a/Tests/JExtractSwiftTests/JNI/JNIAsyncStreamTests.swift b/Tests/JExtractSwiftTests/JNI/JNIAsyncStreamTests.swift index c0cb783cf..39811c464 100644 --- a/Tests/JExtractSwiftTests/JNI/JNIAsyncStreamTests.swift +++ b/Tests/JExtractSwiftTests/JNI/JNIAsyncStreamTests.swift @@ -18,7 +18,7 @@ import SwiftExtract import SwiftJavaConfigurationShared import Testing -/// Functions returning `AsyncStream` return a `SwiftAsyncStream` to Java. +/// Functions returning `AsyncStream` or `AsyncThrowingStream` return a `SwiftAsyncStream` to Java. @Suite struct JNIAsyncStreamTests { @Test @@ -61,6 +61,30 @@ struct JNIAsyncStreamTests { ) } + @Test(arguments: ["Error", "any Error"]) + func throwingStreamsAreAttachedLikeStreams(failure: String) throws { + let input = "public func ticks(count: Int64) -> AsyncThrowingStream { fatalError() }" + try assertOutput( + input: input, + .jni, + .java, + detectChunkByInitialLines: 1, + expectedChunks: [ + "public static org.swift.swiftkit.core.SwiftAsyncStream ticks(long count) {", + "private static native void $ticks(long count, org.swift.swiftkit.core.SwiftAsyncStream result_stream);", + ] + ) + try assertOutput( + input: input, + .jni, + .swift, + detectChunkByInitialLines: 1, + expectedChunks: [ + "SwiftJavaRuntimeSupport._attachAsyncStream(SwiftModule.ticks(count: Int64(fromJNI: count, in: environment)), to: result_stream, environment: environment) { element$, environment in" + ] + ) + } + @Test func jextractedElementsAreMappedToTheirJavaClass() throws { try assertOutput( From ce59fd826464be8738a064d75af39d691d422f1b Mon Sep 17 00:00:00 2001 From: Pierluigi Cifani Date: Tue, 3 Nov 2026 17:22:03 +0100 Subject: [PATCH 2/2] Samples: a Swift stream's error thrown from Kotlin's Flow collect Co-Authored-By: Claude Opus 5.5 --- .../Sources/KotlinSampleLibrary/Streams.swift | 11 +++++++++++ .../test/kotlin/com/example/kotlin/StreamsTest.kt | 12 ++++++++++++ 2 files changed, 23 insertions(+) diff --git a/Samples/SwiftJavaExtractJNIKotlinSampleApp/Sources/KotlinSampleLibrary/Streams.swift b/Samples/SwiftJavaExtractJNIKotlinSampleApp/Sources/KotlinSampleLibrary/Streams.swift index 20d071f48..3f660a6b4 100644 --- a/Samples/SwiftJavaExtractJNIKotlinSampleApp/Sources/KotlinSampleLibrary/Streams.swift +++ b/Samples/SwiftJavaExtractJNIKotlinSampleApp/Sources/KotlinSampleLibrary/Streams.swift @@ -12,6 +12,7 @@ // //===----------------------------------------------------------------------===// +import KotlinSampleModels import Synchronization /// A view model's stream of updates, collected in Kotlin as a Flow. @@ -30,6 +31,16 @@ public final class PriceFeed: Sendable { } } + /// Prices until the connection drops: the stream then fails with `ClientError.offline`. + public func pricesUntilOffline(count: Int64) -> AsyncThrowingStream { + AsyncThrowingStream { continuation in + for i in 0.. AsyncStream { AsyncStream { continuation in 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 ecbd6bb77..398ac425b 100644 --- a/Samples/SwiftJavaExtractJNIKotlinSampleApp/src/test/kotlin/com/example/kotlin/StreamsTest.kt +++ b/Samples/SwiftJavaExtractJNIKotlinSampleApp/src/test/kotlin/com/example/kotlin/StreamsTest.kt @@ -14,6 +14,7 @@ package com.example.kotlin +import com.example.kotlin.models.ClientError import kotlinx.coroutines.flow.map import kotlinx.coroutines.flow.take import kotlinx.coroutines.flow.toList @@ -21,6 +22,7 @@ import kotlinx.coroutines.runBlocking import org.swift.swiftkit.kotlin.asFlow import kotlin.test.Test import kotlin.test.assertEquals +import kotlin.test.assertFailsWith import kotlin.test.assertTrue class StreamsTest { @@ -30,6 +32,16 @@ class StreamsTest { assertEquals(listOf("price-0", "price-1", "price-2"), ids) } + @Test + fun aThrowingStreamFailsTheFlowWithTheSwiftError() = runBlocking { + val ids = mutableListOf() + val error = assertFailsWith { + PriceFeed.init().pricesUntilOffline(2).asFlow().collect { ids += it.id } + } + assertEquals(listOf("price-0", "price-1"), ids) + assertTrue(error.case is ClientError.Case.Offline) + } + @Test fun cancellingTheFlowStopsTheSwiftStream() = runBlocking { val feed = PriceFeed.init()