JNI: AsyncThrowingStream results, with errors delivered to onError

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<Item, Error> {
+    AsyncThrowingStream { continuation in
+      for i in 0..<count {
+        continuation.yield(Item(id: "price-\(i)", count: i))
+      }
+      continuation.finish(throwing: ClientError.offline)
+    }
+  }
+
   /// Live updates that never finish on their own.
   public func live() -> AsyncStream<String> {
     AsyncStream { continuation in

Samples/SwiftJavaExtractJNIKotlinSampleApp/src/test/kotlin/com/example/kotlin/StreamsTest.kt

@@ -14,13 +14,15 @@
 
 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
 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<String>()
+        val error = assertFailsWith<ClientError> {
+            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()

Samples/SwiftJavaExtractJNISampleApp/Sources/MySwiftLibrary/AsyncStreams.swift

@@ -30,3 +30,17 @@ public func shapes(count: Int64) -> AsyncStream<Square> {
     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<Int64, Error> {
+  AsyncThrowingStream { continuation in
+    for value in stride(from: start, through: failure, by: -1) {
+      continuation.yield(value)
+    }
+    continuation.finish(throwing: CountdownInterrupted())
+  }
+}

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<Long> values = Collections.synchronizedList(new ArrayList<>());
+        AtomicReference<Throwable> error = new AtomicReference<>();
+        CountDownLatch done = new CountDownLatch(1);
+        MySwiftLibrary.interruptedCountdown(3, 2).subscribe(new SwiftAsyncStream.Subscriber<Long>() {
+            @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<Long> stream = MySwiftLibrary.countdown(1);

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:

Sources/JExtractSwiftLib/JNI/JNISwift2JavaGenerator+JavaTranslation.swift

@@ -227,7 +227,7 @@ extension JNISwift2JavaGenerator {
       // Name.
       let javaName = javaIdentifiers.makeJavaMethodName(decl)
 
-      // A function returning `AsyncStream<T>` is translated as returning `T`, then converted to a stream.
+      // A function returning `AsyncStream<T>` or `AsyncThrowingStream<T, _>` 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<Element>`, or nil.
+    /// The element type of an `AsyncStream<Element>` or `AsyncThrowingStream<Element, Failure>`, 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<T>`: Java

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)
 

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<Element> {}
+  public struct AsyncThrowingStream<Element, Failure> {}
 
   public protocol Error {}
 

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"

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<Element>(
   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<Element, Failure: Error>(
+  _ stream: AsyncThrowingStream<Element, Failure>,
+  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<Element>(
+  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)
     }
   }

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.
  * <p>
  * {@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<T> {
 
         /** 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<? super T> 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<? super T> subscriber) {
     private static final class Delivery {
         private final java.util.function.Consumer<Object> next;
         private final Runnable complete;
+        private final java.util.function.Consumer<Throwable> error;
         private final AtomicLong task = new AtomicLong(0);
         private final AtomicBoolean finished = new AtomicBoolean(false);
 
-        Delivery(java.util.function.Consumer<Object> next, Runnable complete) {
+        Delivery(java.util.function.Consumer<Object> next, Runnable complete, java.util.function.Consumer<Throwable> 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.

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 <T> SwiftAsyncStream<T>.asFlow(): Flow<T> = callbackFlow {
         override fun onComplete() {
             close()
         }
+
+        override fun onError(error: Throwable) {
+            close(error)
+        }
     })
     awaitClose { subscription.cancel() }
 }