1: type AsyncSeq<'T> = Async<Chunk<'T>> 
  2: and Chunk<'T> = 
  3:     | Done
  4:     | Value of 'T * AsyncSeq<'T>
  5:  
  6: module AsyncSeq =
  7:     let map f (seq : AsyncSeq<_>) : AsyncSeq<_> = 
  8:         let rec doMap s = async {
  9:             let! chunk = s
 10:             match chunk with
 11:             | Done -> return Done
 12:             | Value (v, next) -> return Value (f v, doMap next)
 13:             }
 14:         doMap seq
 15:     
 16:     let run action (seq : AsyncSeq<_>) = 
 17:         let rec doRun s = async {
 18:             let! chunk = s
 19:             match chunk with
 20:             | Done -> return ()
 21:             | Value (v, next) -> action v; return! doRun next
 22:             }
 23:         doRun seq
 24: 
 25:     let filter f (seq : AsyncSeq<_>) : AsyncSeq<_> =
 26:         let rec doFilter s = async { 
 27:             let! chunk = s
 28:             match chunk with
 29:             | Value (value, next) -> 
 30:                 if f value then return Value(value, doFilter next)
 31:                 else return! doFilter next
 32:             | x -> return x
 33:             }
 34:         doFilter seq
 35: 
 36: [<AutoOpen>]
 37: module AsyncSeqExtensions = 
 38:     
 39:     open System.Text
 40: 
 41:     type System.IO.Stream with
 42:         member this.AsyncReadSeq(?bufferSize) : AsyncSeq<byte[]> = 
 43:             let bufferSize = defaultArg bufferSize (2 <<< 16)
 44:             let temp : byte[] = Array.zeroCreate bufferSize
 45:             let rec doRead () = async {
 46:                 let! count = this.AsyncRead(temp, 0, bufferSize)
 47:                 if count = 0 then return Done
 48:                 else 
 49:                     let buf = Array.sub temp 0 count
 50:                     return Value(buf, doRead ())
 51:                 }
 52:             doRead ()
 53:         
 54:         member this.AsyncReadLines(?bufferSize) = 
 55:             let sb = StringBuilder()
 56:             let getText = AsyncSeq.map Encoding.UTF8.GetString
 57: 
 58:             let rec doRead (s : AsyncSeq<string>) = async {
 59:                 let! chunk = s
 60:                 match chunk with
 61:                 | Done -> 
 62:                     if sb.Length <> 0 then return Value(sb.ToString(), async.Return Done )
 63:                     else return Done
 64:                 | Value(part, next) -> 
 65:                     return! doProcess part 0 next
 66:                 }
 67:             and doProcess (text : string) n next = async {
 68:                 let (|Chars|) pos = 
 69:                     if pos < text.Length - 1 then text.[pos], Some (text.[pos + 1])
 70:                     else text.[pos], None
 71:                 
 72:                 let getLine newPos = 
 73:                     let line = sb.ToString()
 74:                     sb.Length <- 0
 75:                     Some (line, newPos)
 76: 
 77:                 let rec run n =
 78:                     if n >= text.Length then None
 79:                     else 
 80:                         match n with
 81:                         | Chars ('\r', Some '\n') -> getLine (n + 2)
 82:                         | Chars ('\r', _) 
 83:                         | Chars ('\n', _) -> getLine (n + 1)
 84:                         | Chars (c, _) -> 
 85:                             sb.Append(c) |> ignore
 86:                             run(n + 1)
 87:                 match run n with
 88:                 | Some (line, pos) -> return Value (line, doProcess text pos next)
 89:                 | None -> return! doRead next
 90:                 }
 91:                 
 92:             this.AsyncReadSeq(?bufferSize=bufferSize)
 93:                 |> AsyncSeq.map Encoding.UTF8.GetString
 94:                 |> doRead
 95:         
 96:         member this.AsyncWriteSeq(seq : AsyncSeq<byte[]>) = 
 97:             let rec run s = async {
 98:                 let! chunk = s
 99:                 match chunk with
100:                 | Done -> return ()
101:                 | Value(data, next) ->
102:                     do! this.AsyncWrite(data)
103:                     return! run next
104:                 }
105:             run seq       
106: 
107: open System.IO
108: 
109: let printWithPrefix path prefix = async {
110:     use f= File.Open(path, FileMode.Open)
111:     do! f.AsyncReadLines()
112:         |> AsyncSeq.map (sprintf "%s: %s" prefix)
113:         |> AsyncSeq.run (printfn "%s")
114:     }