anchor.diff:                      |None -> DispatchReady <| (lines |> List.rev)
                       |Some li -> Processing <| li::lines)
   
  -    let receive (network:unit -> Stream) (retryDelay:TimeSpan option) =
  +    let receive (network:unit -> Stream) (retryDelay:TimeSpan seq option) =
         let rec read (observer:IObserver<_>) (sr:StreamReader) = async {
           match (sr.ReadLine()) with
           | null -> observer.OnCompleted()
  
positive.code:         |> processStrings
          |> Observable.filter SSEProcessingState.isReady
          |> Observable.map (SSEProcessingState.unwrap >> SSEEvent.fromLines)
          |> retryAfterDelay (defaultArg retryDelay (TimeSpan.FromSeconds 0.))
  
    type Connection =
      static member Receive(network:unit -> Stream, ?retryDelay: TimeSpan) =
        Observable.receive network retryDelay
negative.code:         { new IDisposable with
              member __.Dispose() = ct.Cancel() }
  
      let repeatDelay (delay:TimeSpan seq) (src:IObservable<_>) = seq{
          yield src
          let en = delay.GetEnumerator()
          while en.MoveNext() do
            yield src.DelaySubscription(en.Current)}
  
      let retryAfterDelay (delay:TimeSpan seq) (src:IObservable<_>)  =
          src |> repeatDelay delay |> Observable.catchSeq
  
      let processStrings (obs:IObservable<_>) =
        obs |> Observable.scanInit (Processing [])
  