AsyncSequence Dependencies

[rhysm94]

Hey all,

Perhaps slightly off topic, but I think it's on-topic to how dependencies are typically defined when using TCA.

Just wondering if anyone has a dependency which returns an AsyncSequence, and if so, how you're handling the fact it's missing a primary associated type?

For example, I'm using GRDB as my database layer in this application. There's a certain table, say of Player types, that I want to observe. So in my DatabaseManager type, I would implement my observePlayers method like this:

ValueObservation.tracking(Player.fetchAll).values(in: writer)

However, this returns a GRDB specific type - AsyncValueObservation<Player>. Obviously, I don't want any GRDB types to bleed into my DatabaseManager interface, as I don't want to be building GRDB each time I build my non-live DatabaseManager implementation (even though it's a small dependency and doesn't take long to build).

If I was doing this with Combine (which I probably will go back to), the type for my observePlayers method would be @Sendable () -> AnyPublisher<[Player], Error>. Ideally, I'd like to do quite similar with async/await instead of Combine.

Is there a way around this? How are you designing AsyncSequence-returning dependencies?

[mbrandonw]

You can kinda think of AsyncStream as a type erased AsyncSequence (with caveats), and in fact the library ships with an initializer on AsyncStream that takes a sequence to do the erasing:

swift-composable-architecture/Sources/ComposableArchitecture/Effects/ConcurrencySupport.swift

Lines 1 to 76 in ed3a380

	extension AsyncStream {
	/// Initializes an `AsyncStream` from any `AsyncSequence`.
	///
	/// Useful as a type eraser for live `AsyncSequence`-based dependencies.
	///
	/// For example, your feature may want to subscribe to screenshot notifications. You can model
	/// this as a dependency client that returns an `AsyncStream`:
	///
	/// ```swift
	/// struct ScreenshotsClient {
	/// var screenshots: () -> AsyncStream<Void>
	/// func callAsFunction() -> AsyncStream<Void> { self.screenshots() }
	/// }
	/// ```
	///
	/// The "live" implementation of the dependency can supply a stream by erasing the appropriate
	/// `NotificationCenter.Notifications` async sequence:
	///
	/// ```swift
	/// extension ScreenshotsClient {
	/// static let live = Self(
	/// screenshots: {
	/// AsyncStream(
	/// NotificationCenter.default
	/// .notifications(named: UIApplication.userDidTakeScreenshotNotification)
	/// .map { _ in }
	/// )
	/// }
	/// )
	/// }
	/// ```
	///
	/// While your tests can use `AsyncStream.streamWithContinuation` to spin up a controllable stream
	/// for tests:
	///
	/// ```swift
	/// let screenshots = AsyncStream<Void>.streamWithContinuation()
	///
	/// let store = TestStore(
	/// initialState: Feature.State(),
	/// reducer: Feature()
	/// )
	///
	/// store.dependencies.screenshots.screenshots = { screenshots.stream }
	///
	/// screenshots.continuation.yield() // Simulate a screenshot being taken.
	///
	/// await store.receive(.screenshotTaken) { ... }
	/// ```
	///
	/// - Parameters:
	/// - sequence: An `AsyncSequence`.
	/// - limit: The maximum number of elements to hold in the buffer. By default, this value is
	/// unlimited. Use a `Continuation.BufferingPolicy` to buffer a specified number of oldest or
	/// newest elements.
	public init<S: AsyncSequence & Sendable>(
	_ sequence: S,
	bufferingPolicy limit: Continuation.BufferingPolicy = .unbounded
	) where S.Element == Element {
	self.init(bufferingPolicy: limit) { (continuation: Continuation) in
	let task = Task {
	do {
	for try await element in sequence {
	continuation.yield(element)
	}
	} catch {}
	continuation.finish()
	}
	continuation.onTermination =
	{ _ in
	task.cancel()
	}
	// NB: This explicit cast is needed to work around a compiler bug in Swift 5.5.2
	as @Sendable (Continuation.Termination) -> Void
	}
	}

We have an example of using this in the long-living effect case study, where we erase the NotificationCenter async sequence to be an AsyncStream:

swift-composable-architecture/Examples/CaseStudies/SwiftUICaseStudies/02-Effects-LongLiving.swift

Lines 57 to 63 in ed3a380

	static let liveValue: @Sendable () async -> AsyncStream<Void> = {
	await AsyncStream(
	NotificationCenter.default
	.notifications(named: UIApplication.userDidTakeScreenshotNotification)
	.map { _ in }
	)
	}

The biggest caveat is that AsyncStream does not support multiple subscribers. So, it only works if you are willing to vend a fresh AsyncStream for each time your dependency endpoint is called. That's why the screenshot dependency above is a function () -> AsyncStream rather than a bare AsyncStream.

[tgrapperon]

There is another caveat with the AsyncStream when initialized from any AsyncSequence, as it starts observing the sequence as soon as it is created and will replay these values when you start iterating over it. So it is better to create this value at the last moment, and not store it around to start iterating later.

[JacksonUtsch]

These caveats make it hard for usage. Especially no CurrentValueSubject equivalent without using a third party AsyncSequence implementation. Hoping good native support comes soon, meanwhile I'll be sticking to Combine.

[rhysm94]

These caveats make it hard for usage. Especially no CurrentValueSubject equivalent without using a third party AsyncSequence implementation. Hoping good native support comes soon, meanwhile I'll be sticking to Combine.

I believe there's an AsyncChannel type in Async Algorithms from Apple, but iirc, it's more like a PassthroughSubject than a CurrentValueSubject. Is that right?

[tgrapperon]

AsyncChannel has the same "Unique consumer" issue than any AsyncSequence unfortunately.
The potential replacement will likely involve AsyncBroadcastSequence to allow multiple consumers like CurrentValueSubject, but I'm afraid we'll have to wait for a long time for official support.

[corysullivan]

You can also have a look at AsyncExtensions. There is some overlap with apples Async Algorithms, but it seems to have some useful operators like multicast and a AsyncCurrentValueSubject