AsyncStreamReducer.swift

import ComposableArchitecture
import Foundation

@Reducer
struct AsyncStreamReducer<Element>
where Element: Sendable,
      Element: Equatable
{
  struct State: Equatable {
    let id = UUID()
    var element: Element?
  }

  enum Action {
    case observe
    case element(Element)
    case finished
    case cancel
  }

  enum CancelID: Hashable {
    case observe(id: UUID)
  }

  let stream: @Sendable () -> AsyncStream<Element>

  var body: some ReducerOf<Self> {
    Reduce { state, action in
      switch action {
      case .observe:
        return .run { send in
          for await element in stream() {
            await send(.element(element))
          }
          await send(.finished)
        }
        .cancellable(
          id: CancelID.observe(id: state.id),
          cancelInFlight: true
        )

      case .element(let element):
        state.element = element
        return .none

      case .finished:
        return .none

      case .cancel:
        return .cancel(
          id: CancelID.observe(id: state.id)
        )
      }
    }
  }
}

AsyncThrowingStreamReducer.swift

import ComposableArchitecture
import Foundation

@Reducer
struct AsyncThrowingStreamReducer<Element, Failure>
where Element: Sendable,
      Element: Equatable,
      Failure: Error
{
  struct State: Equatable {
    let id = UUID()
    var element: Element?
  }

  enum Action {
    case observe
    case element(Element)
    case completion(Failure? = nil)
    case cancel
  }

  enum CancelID: Hashable {
    case observe(id: UUID)
  }

  let stream: @Sendable () -> AsyncThrowingStream<Element, Failure>

  var body: some ReducerOf<Self> {
    Reduce { state, action in
      switch action {
      case .observe:
        return .run { send in
          for try await element in stream() {
            await send(.element(element))
          }
          await send(.completion())
        } catch: { error, send in
          let error: Failure = error as! Failure
          await send(.completion(error))
        }
        .cancellable(
          id: CancelID.observe(id: state.id),
          cancelInFlight: true
        )

      case .element(let element):
        state.element = element
        return .none

      case .completion(_):
        return .none

      case .cancel:
        return .cancel(
          id: CancelID.observe(id: state.id)
        )
      }
    }
  }
}