Add async stream interfaces

[haeseoklee]

Summary

Add Swift Concurrency-friendly interfaces on top of the existing RxSwift-based RIBs APIs.

This change keeps RxSwift as the source of truth and adds additive async/await conveniences for observing lifecycle streams, building and consuming workflow steps, confining async sequences to RIB lifecycle, and binding Swift Tasks to existing RIB lifecycle scopes.

Changes

• Add Observable to async sequence bridges:

  • ObservableType.asAsyncStream(...)
  • ObservableType.asAsyncThrowingStream(...)

• Add async lifecycle stream conveniences:

  • InteractorScope.isActiveSequence
  • RouterScope.lifecycleSequence
  • Working.isStartedSequence
  • LeakDetector.statusSequence

• Add Workflow async APIs:

  • Workflow.onAsyncStep(...)
  • Step.onAsyncStep(...)
  • Step.asAsyncSequence(...)
  • AsyncSequence.fork(_:)

• Add lifecycle-scoped task helpers:

  • Interactor.taskOnDeactivate(...)
  • Interactor.throwingTaskOnDeactivate(...)
  • Worker.taskOnStop(...)
  • Worker.throwingTaskOnStop(...)
  • Workflow.task(...)
  • Workflow.throwingTask(...)

• Add async sequence lifecycle confinement:

  • AsyncSequence.confineTo(_:)
  • Returns an AsyncThrowingStream<Element, Error>
  • Uses the existing RxSwift confineTo(_:) behavior internally by bridging the AsyncSequence to an Observable

• Add tests covering

Notes

The async APIs are additive and preserve existing RxSwift APIs.

RxSwift remains the internal runtime for lifecycle and workflow behavior. The async interfaces bridge from the existing observable streams and disposable lifecycle hooks rather than replacing them.

Close #6

[CLAassistant]

All committers have signed the CLA.

[alexvbush]

thank you for your submission @haeseoklee ! This looks very interesting and useful and might be a great addition to the framework! That said I will be thoroughly reviewing this this/next week to check if there is any conflicts with incoming #49

[alexvbush]

@haeseoklee Thanks for submitting this PR again!

I checked out your branch and ran some tests myself and it seems like your changes will merge cleanly with #49 when it eventually lands. No issues there and as you said your changes are just additive.

I'll still done more thorough testing and review of your code soon but so far here's what I found:

While running the test suite I hit two failures in the new ConcurrencyTests that look independent of any environment setup — they reproduce on this branch with a clean checkout:

1. test_asyncSequenceFork_createsWorkflowStep — fails deterministically

ConcurrencyTests.swift:116. The assertion XCTAssertEqual(workflow.completeCallCount, 1) (line 127) fails with 0 != 1 on every run.

The test body is synchronous, but fork(_:) bridges the AsyncSequence into an Observable by consuming it on a Task:

func test_asyncSequenceFork_createsWorkflowStep() {   // not async
    ...
    let step: Step<(), (), ()> = asyncSequence.fork(workflow)
    _ = step.commit().subscribe(())
    XCTAssertEqual(workflow.forkCallCount, 1)       // passes (didFork is sync)
    XCTAssertEqual(workflow.completeCallCount, 1)   // always 0 — async work hasn't run yet
}

forkCallCount passes because didFork() is called synchronously, but the stream is drained asynchronously, so the completion hasn't propagated by the time the assertion runs. The test needs to await the completion (e.g. make it async and await an expectation / the step's output) rather than asserting immediately.

2. test_asyncSequenceConfineTo_yieldsLatestElementWhenInteractorBecomesActive — flaky

ConcurrencyTests.swift:23. Passes intermittently (~1 in 3 for me); when it fails it's thirdValue (line 43) returning Optional(2) instead of Optional(3).

The test interleaves sourceContinuation.yield(...), interactor.activate()/deactivate(), and await iterator.next() synchronously, but confineTo is built on combineLatest(interactorScope.isActiveStream.values, self) — there's no ordering guarantee between the activeness stream and the source after a deactivate→yield→activate sequence, so "the latest value re-emitted on reactivation" races. It'd be worth either making the confinement semantics deterministic or restructuring the test to synchronize on the expected emissions instead of assuming step-by-step ordering.

[haeseoklee]

@alexvbush
Thank you for checking this out and for the detailed feedback!

1. test_asyncSequenceFork_createsWorkflowStep

fork(_:) consumes the underlying AsyncSequence asynchronously, so the previous test was asserting completeCallCount too early. I updated the test to be async and to wait for workflow completion before asserting.

2. test_asyncSequenceConfineTo_yieldsLatestElementWhenInteractorBecomesActive

I updated the confinement implementation to delegate to the existing RxSwift confineTo(_:) behavior internally, so it preserves the current push-based Rx semantics, and adjusted the tests to synchronize on expected emissions instead of assuming immediate iterator ordering.

While making these changes, I also reorganized the new concurrency tests by feature area and added a bit more lifecycle/cancellation coverage.

[alexvbush]

thank you for the fixes and the updates @haeseoklee.

I am currently testing your changes with a production codebase and replacing existing Rx implementations/calls with your swift concurrency friendly equivalents to see if everything still works and behaves the same.

after that I'll go over the code you have and leave feedback, if any, on refactoring or naming changes, etc.

I'll keep you posted.

Thank you again for your submission!

[alexvbush]

Again, thank you for submitting this PR. After reviewing it thoroughly there are some changes that we'd need to make before merging it, and a clarification on the direction we want the RIBs iOS framework to take as of today.

Direction: keep Rx as the backbone for async work in RIBs

RIBs is Rx-native. The vast majority of real interactors are built as cold Observable chains that end in subscribe(...).disposeOnDeactivate(interactor:).

So the highest-value async support we can add today is tight interop inside that model, letting you call async functions from within an Rx chain, rather than parallel async-first method calls.

Most of this PR points the opposite way. It's built to let you leave Rx and consume RIBs lifecycle and streams as AsyncSequences in async-first code (isActiveSequence and friends, AsyncSequence.confineTo, Step.asAsyncSequence).

Making RIBs async-first is not the direction we want to go because we find more value and flexibility in handling asynchronicity with Rx rather than only async/await (swift concurrency) - Rx can do everything async/await can + some, where async/await falls short on being flexible.

So I'd like to narrow this PR to the Rx-backbone interop pieces. The async-first APIs aren't something we intend to bring into the framework right now.

What we want instead: a few small conveniences for Rx chains

The idea is simple. Give people a handful of helpers so an async call drops straight into a normal Rx chain, and the whole thing still ends in one subscribe(...).disposeOnDeactivate(interactor:). There are three of them.

1. Add the missing async-function → Rx bridge (+ thin operators)

RxSwift already bridges an async sequence into Rx (AsyncSequence.asObservable()) and an Rx stream back into async (.values). What it doesn't give you is the most basic case: wrapping a single async call as a cold Observable. That's the piece that's missing here.

This is the primitive that makes the Rx-backbone model actually ergonomic:

public extension PrimitiveSequenceType where Trait == SingleTrait {
    static func fromAsync(_ work: @escaping () async throws -> Element) -> Single<Element> {
        .create { observer in
            let task = Task {
                do    { observer(.success(try await work())) }
                catch { observer(.failure(error)) }
            }
            return Disposables.create { task.cancel() }   // cold; cancels on dispose
        }
    }
}

Plus a small set of operators so that callers don't repeat flatMap { .fromAsync { … } }:

public extension ObservableType {
    /// Order-preserving (concatMap): like `map`, but `transform` may await.
    func mapAsync<T>(_ transform: @escaping (Element) async throws -> T) -> Observable<T> {
        concatMap { el in Single.fromAsync { try await transform(el) } }
    }
    /// Switch-to-latest (flatMapLatest): cancels in-flight async work on a new element.
    func flatMapLatestAsync<T>(_ transform: @escaping (Element) async throws -> T) -> Observable<T> {
        flatMapLatest { el in Single.fromAsync { try await transform(el) } }
    }
    // concatMapAsync / flatMapAsync as needed
}

Now async and Rx compose in one chain, with one subscribe, one disposeOnDeactivate, and cancellation propagating down to the in-flight Task for free:

someObservable
    .flatMapLatestAsync { id in try await service.fetch(id) }
    .observe(on: MainScheduler.instance)
    .subscribe(onNext: { … })
    .disposeOnDeactivate(interactor: self)

2. Refactor the Task lifecycle helpers into Task extensions

Task+RIBs.swift has six methods right now (taskOnDeactivate / throwingTaskOnDeactivate, taskOnStop / throwingTaskOnStop, task / throwingTask).

Because Task<Success, Failure> is already generic, the plain and throwing pairs consolidate into one. Please replace them with three extensions on Task itself, mirroring how disposeOnDeactivate extends Disposable:

public extension Task {
    @discardableResult
    func cancelOnDeactivate(interactor: Interactor) -> Task {
        Disposables.create { self.cancel() }.disposeOnDeactivate(interactor: interactor)
        return self
    }
    // cancelOnStop(_ worker:)          via disposeOnStop(_:)
    // cancel(with workflow:)           via disposeWith(workflow:)
}

Task { try await something() }.cancelOnDeactivate(interactor: self)

Why the extension is the cleaner primitive here, in my books:

• Six methods drop to three, and throwing/non-throwing consolidates for free.
• It reads exactly like the existing Disposable.disposeOnDeactivate (Task { } is the async version of subscribe, the Task is the Disposable), so it's one consistent lifecycle vocabulary.
• It decouples task construction from lifecycle binding, so it composes with Task.detached, priorities, async let, and task groups. None of which the closure-only form allows.
• The current throwing methods have a problem too. They return a @discardableResult Task<Void, Error>, and since you usually don't hold onto that returned Task, any error the async work throws goes into a result no one reads and just disappears, unless you catch it inside the closure yourself. With the extension you write the Task { } body, so you handle its errors right there in your own code.

3. Rebuild async Workflow steps on the same primitive

onAsyncStep has its own Observable.create { Task { … } } implementation, which is exactly what Single.fromAsync does.

You build a Step with onStep, and since that closure already returns an Observable, an async step is really just an onStep that returns Single.fromAsync { … }.asObservable().

Please keep onAsyncStep (root and step), just as a one-line wrapper over the shared primitive instead of a parallel bridge:

public final func onAsyncStep<NextActionableItemType, NextValueType>(
    _ work: @escaping (ActionableItemType, ValueType) async throws -> (NextActionableItemType, NextValueType)
) -> Step<WorkflowActionableItemType, NextActionableItemType, NextValueType> {
    onStep { actionableItem, value in
        Single.fromAsync { try await work(actionableItem, value) }.asObservable()
    }
}

How you call it is unchanged from what you already have. Async and Rx steps interleave freely:

workflow
    .onStep { rootItem in rootItem.waitForLogin() }                 // Rx step
    .onAsyncStep { loggedInItem, _ in                              // async step
        (loggedInItem, try await service.fetchClient(id: 42))
    }
    .onStep { item, client in item.routeToClientDetails(client) }   // Rx step
    .commit()
    .subscribe(rootActionableItem)
    .disposeOnDeactivate(interactor: self)

A step stays a pure Rx chain (Observable in → Observable out), and cancellation still works because fromAsync cancels its Task when the subscription is disposed.

What to remove from this PR

With those three in place you can call an async function from inside any Rx chain, bind that work to a RIB's lifecycle, and write async workflow steps, all without leaving Rx. So a large part of this PR (honestly, more than half) stops earning its place. Some of it solves the reverse direction we don't want, and some of it duplicates what RxSwift or RIBs already gives you.

Here's what I'd remove from this PR:

• Out of scope, not something we're adding to the framework. The async-first consumption APIs: isActiveSequence / lifecycleSequence / isStartedSequence / statusSequence, AsyncSequence.confineTo(_:), and Step.asAsyncSequence.
• Drop as redundant. asAsyncStream / asAsyncThrowingStream (the throwing one duplicates RxSwift 6.9's own ObservableConvertibleType.values), and AsyncSequence.fork (you can already write someAsyncSequence.asObservable().fork(workflow) using RxSwift's asObservable() plus the existing Rx ObservableType.fork).
• Drop, it's redundant. WorkflowHandle + Workflow.start(_:). subscribe(_:) already returns a Disposable you cancel with dispose() (or bind with disposeOnDeactivate), same as interactors and workers. start(_:) just wraps that in a handle whose cancel() calls dispose(), so it adds nothing and breaks that consistency.

What we want to keep

The narrowed PR ends up small:

• onAsyncStep (root and step), rebuilt on fromAsync as in §3. The way you already call it stays exactly the same.
• The Task lifecycle helpers, refactored into the three Task extensions from 2. above. Same capability, smaller scope.
• The new Single.fromAsync / Observable.fromAsync plus the mapAsync / flatMapLatestAsync operators from 1. above.

Everything else in the framework stays as it is. The existing Rx confineTo, fork, subscribe, and disposeOnDeactivate already cover the rest.

Happy to have a discussion about this. Thanks.

[haeseoklee]

@alexvbush

Thank you for the thorough review.

My original intent with this PR was to start reducing the amount of RxSwift surface area that application code needs to interact with directly, and to explore whether RIBs could gradually become more approachable from Swift Concurrency-based code without requiring callers to import RxSwift everywhere. I still believe that direction may have long-term value, especially as Swift Concurrency continues to become the native concurrency model in Swift.

That said, your feedback makes sense to me. I agree that introducing async-first APIs in this PR is premature, especially given that RIBs is still fundamentally Rx-native and that the existing lifecycle and workflow model is built around Rx chains.

I’ve updated the branch to narrow the scope accordingly:

• removed the async-first consumption APIs
• removed the AsyncSequence workflow/fork/confine APIs
• removed WorkflowHandle/start
• added Single.fromAsync / Observable.fromAsync
• refactored the task lifecycle helpers into Task extensions
• rebuilt both root and step onAsyncStep on top of Single.fromAsync

I also removed mapAsync / flatMapLatestAsync after thinking more about the public API surface. While those operators can be convenient, they feel more like general RxSwift + Swift Concurrency helpers than RIBs-specific APIs, and projects can define them locally if they need those exact semantics.

With these changes, the PR is now focused on a smaller Swift Concurrency interop surface within the existing Rx workflow and lifecycle model, rather than introducing a parallel async-first API surface.

Thanks again for taking the time to review this so carefully.