JNI: AsyncStream results as SwiftAsyncStream

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

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
 }
 

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

@@ -0,0 +1,42 @@
+//===----------------------------------------------------------------------===//
+//
+// 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.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.assertTrue
+
+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())
+    }
+}

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<Int64> {
+  AsyncStream { continuation in
+    for value in stride(from: start, through: 0, by: -1) {
+      continuation.yield(value)
+    }
+    continuation.finish()
+  }
+}
+
+public func shapes(count: Int64) -> AsyncStream<Square> {
+  AsyncStream { continuation in
+    for side in 1...max(count, 1) {
+      continuation.yield(Square(side: Double(side)))
+    }
+    continuation.finish()
+  }
+}

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 <T> List<T> collect(SwiftAsyncStream<T> stream) throws InterruptedException {
+        List<T> values = Collections.synchronizedList(new ArrayList<>());
+        CountDownLatch done = new CountDownLatch(1);
+        stream.subscribe(new SwiftAsyncStream.Subscriber<T>() {
+            @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<Square> squares = collect(MySwiftLibrary.shapes(2, arena));
+            assertEquals(2, squares.size());
+            assertEquals(4.0, squares.get(1).area());
+        }
+    }
+
+    @Test
+    void canOnlyBeSubscribedOnce() throws Exception {
+        SwiftAsyncStream<Long> stream = MySwiftLibrary.countdown(1);
+        collect(stream);
+        assertThrows(IllegalStateException.class, () -> stream.subscribe(value -> {}));
+    }
+}

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:

Sources/JExtractSwiftLib/JNI/JNISwift2JavaGenerator+JavaTranslation.swift

@@ -227,15 +227,25 @@ extension JNISwift2JavaGenerator {
       // Name.
       let javaName = javaIdentifiers.makeJavaMethodName(decl)
 
+      // A function returning `AsyncStream<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 {
+        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<Element>`, 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<T>`: 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
 

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,

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<T>
+  static func swiftAsyncStream(_ T: JavaType) -> JavaType {
+    .class(package: "org.swift.swiftkit.core", name: "SwiftAsyncStream", typeParameters: [T.boxedType])
+  }
 }

Sources/SwiftExtract/SwiftTypes/SwiftKnownModules.swift

@@ -104,6 +104,9 @@ private let swiftSourceFile: SourceFileSyntax = """
 
   public struct Set<Element> {}
 
+  // `_Concurrency` in reality; modeled here with the other standard library types.
+  public struct AsyncStream<Element> {}
+
   public protocol Error {}
 
   // FIXME: Support 'typealias Void = ()'