Functions for working with AsyncSeq and Channel

[njlr]

I propose adding two functions:

AsyncSeq.toChannel : ChannelWriter<'a> -> AsyncSeq<'a> -> Async<unit>

AsyncSeq.fromChannel : ChannelReader<'a> -> AsyncSeq<'a>

AsyncSeq.toChannel would be used to fill a ChannelWriter from an AsyncSeq

AsyncSeq.fromChannel would be used to empty a ChannelReader into an AsyncSeq

Here is an initial implementation:

#r "nuget: FSharp.Control.AsyncSeq, 2.0.23"

open System.Threading.Channels
open FSharp.Control

[<RequireQualifiedAccess>]
module AsyncSeq =

  let toChannel (writer : ChannelWriter<'a>) (xs :  AsyncSeq<'a>) : Async<unit> =
    async {
      try
        for x in xs do
          if not (writer.TryWrite(x)) then
            let! ct = Async.CancellationToken

            do!
              writer.WriteAsync(x, ct).AsTask()
              |> Async.AwaitTask

        writer.Complete()
      with exn ->
        writer.Complete(error = exn)
    }

  let fromChannel (reader : ChannelReader<'a>) : AsyncSeq<'a> =
    asyncSeq {
      let mutable keepGoing = true

      while keepGoing do
        let mutable item = Unchecked.defaultof<'a>

        if reader.TryRead(&item) then
          yield item
        else
          let! ct = Async.CancellationToken

          let! hasMoreData =
            reader.WaitToReadAsync(ct).AsTask()
            |> Async.AwaitTask

          if not hasMoreData then
            keepGoing <- false
    }

// Demo
async {
  let xs = [ 0 .. 9] |> AsyncSeq.ofSeq

  let channel = Channel.CreateBounded(4)

  do!
    Async.Parallel
      [|
        AsyncSeq.toChannel channel.Writer xs

        async {
          for x in AsyncSeq.fromChannel channel.Reader do
            printfn $"%i{x}"
        }
      |]
    |> Async.Ignore<unit array>
}
|> Async.RunSynchronously

Would a PR be accepted?

Thanks!

[dsyme]

Hi @njlr!

Yes, a PR with good testing would be accepted

You might need to update the engineering in this repository a bit - I haven't checked lately but I assume we should move to latest .NET etc.

[dsyme]

Oh also make sure you bump the version numbers too (via CHANGELOG)