AsyncChannelObserver.swift

import AsyncAlgorithms

final class AsyncChannelObserverToken {
    private let task: Task<Void, Error>
    fileprivate init(_ task: Task<Void, Error>) {
        self.task = task
    }

    func cancel() {
        if !task.isCancelled {
            task.cancel()
        }
    }

    deinit {
        cancel()
    }
}

extension AsyncChannel {

    func addObserver(@_inheritActorContext _ observer: @escaping @isolated(any) @Sendable (Element) -> Void) -> AsyncChannelObserverToken {
        AsyncChannelObserverToken(Task {
            for await value in self where !Task.isCancelled {
                await observer(value)
            }
        })
    }

}

// Usage

let observer = channel.addObserver { element in
  //
}

// later
observer = nil