Ergonomics of TaskResult with AsyncThrowingStream

[rcarver]

Building on the discussion in #1847, here's an idea for improving the ergonomics of using a TaskResult with an AsyncThrowingStream

Given this client endpoint and domain action:

struct Client {
  public var fetchTemplates: @Sendable () async -> AsyncThrowingStream<[Template], Error>
}
enum Action {
  case fetchTemplatesResponse(TaskResult<[Template]>)
}

Current approach to iterating over a throwing async stream. Requires adding the catch block to run, which isn't typically needed when sending a single throwing action by using TaskResult(catching:)

return .run { send in
  for try await response in await self.templatesClient.fetchTemplates() {
    await send(.fetchTemplatesResponse(.success(response)))
  }
} catch: { error, send in
  await send(.fetchTemplatesResponse(.failure(error)))
}

Proposal

Introduce TaskResult.map that transforms a throwing stream into a non-throwing stream

return .run { send in 
  for await result in TaskResult.map(await self.templatesClient.fetchTemplates()) {
    await send(.fetchTemplatesResponse(result))
  }
}

POC implementation

extension TaskResult {
  static func map(
    _ stream: AsyncThrowingStream<Success, Error>
  ) -> AsyncStream<TaskResult<Success>> {
    AsyncStream { continuation in
      let task = Task {
        defer { continuation.finish() }
        do {
          for try await value in stream {
            continuation.yield(.success(value))
          }
        } catch let error where error is CancellationError {
          return
        } catch {
          continuation.yield(.failure(error))
        }
      }
      continuation.onTermination = { _ in task.cancel() }
    }
  }
}

Worth PRing? Happy to take this further if there's interest.

[Sajjon]

I think AsyncStream.materialize to reuse RxSwift terminology would more fitting. It is a bit misleading letting the func be on TaskResult.

[Sajjon]

Not that the term "materialize" is super clear, but it has precedence.

[rcarver]

I'm not familiar with that term, but here's what it looks like to put a method on the stream itself (name very much tbd)

return .run { send in 
  for await result in await self.templatesClient.fetchTemplates().mapTaskResult() {
    await send(.fetchTemplatesResponse(result))
  }
}

[Sajjon]

Materialize ref:

http://introtorx.com/Content/v1.0.10621.0/08_Transformation.html#MaterializeAndDematerialize

https://reactivex.io/documentation/operators/materialize-dematerialize.html

https://github.com/ReactiveX/RxSwift/blob/main/RxSwift/Observables/Materialize.swift

https://github.com/CombineCommunity/CombineExt/blob/main/Sources/Operators/Materialize.swift