some utility operators for AsyncStream inspired by RxSwift

AsyncStream+Extension.swift

import Foundation

extension AsyncStream {
  init(
    _ elementType: Element.Type = Element.self,
    bufferingPolicy limit: AsyncStream<Element>.Continuation.BufferingPolicy = .unbounded,
    _ build: @escaping (AsyncStream<Element>.Continuation) async -> Void
  ) {
    self = AsyncStream(elementType, bufferingPolicy: limit) { continuation in
      Task { await build(continuation) }
    }
  }
}

extension AsyncStream.Continuation {
  public func finish(with value: Element) {
    self.yield(value)
    self.finish()
  }
}

extension AsyncStream {
  static func empty<Element>() -> AsyncStream<Element> {
    .init { $0.finish() }
  }
  static func just<Element>(_ element: Element) -> AsyncStream<Element> {
    .init { $0.finish(with: element) }
  }
  static func of<Element>(_ elements: Element...) -> AsyncStream<Element> {
    .init { continuation in
      elements.forEach { continuation.yield($0) }
      continuation.finish()
    }
  }
  static func merge<Element>(_ streams: AsyncStream<Element>...) -> AsyncStream<Element> {
    .init { continuation in
      let tasks = streams.reduce(into: [Task<Void, Never>]()) { tasks, stream in
        tasks.append(
          Task {
            for await element in stream {
              continuation.yield(element)
            }
          }
        )
      }
      
      for task in tasks {
        await task.value
      }
      continuation.finish()
    }
  }
  static func concat<Element>(_ streams: AsyncStream<Element>...) -> AsyncStream<Element> {
    .init { continuation in
      for stream in streams {
        for await element in stream {
          continuation.yield(element)
        }
      }
      continuation.finish()
    }
  }
}