Other option would be to simply use OnNext instead of OnError for connection issues and wrap inside an F# result

module temp.RedisRx

open System
open System.Threading
open System.Reactive.Linq
open System.Reactive
open System.Threading.Tasks
open FSharp.Control.Reactive
open FSharp.Control.Redis.Streams.Core
open FSharp.Control.Tasks
open StackExchange.Redis

module Observable =
  let internal taskUnfold (fn: 's -> CancellationToken -> Task<('s * 'e) option>) (state: 's) =
    Observable.Create(fun (obs: IObserver<_>) ->
      let cts = new CancellationTokenSource()
      let ct = cts.Token
      task {
        let mutable innerState = state
        let mutable isFinished = false
        try
            while not ct.IsCancellationRequested || not isFinished do
              try 
                let! result = fn innerState ct
                match result with
                | Some (newState, output) ->
                    innerState <- newState
                    obs.OnNext <| Ok output
                | None ->
                    isFinished <- true
              with e ->
                obs.OnNext <| Error e
            obs.OnCompleted()
        finally
          cts.Dispose()
      }
      |> ignore

      new Disposables.CancellationDisposable(cts) :> IDisposable)

  let flattenArray (observable : IObservable<Result<StreamEntry [],exn>>) =
      observable.SelectMany (fun x ->
          match x with
          | Ok o -> o |> Array.map (Ok) :> seq<_>
          | Error e -> [Error e] :> seq<_>)


let pollStreamForever (redisdb : IDatabase) (streamName : RedisKey) (startingPosition : RedisValue) (pollOptions : PollOptions) =
    Observable.taskUnfold (fun (nextPosition, pollDelay) ct -> task {
        let! (response : StreamEntry []) = redisdb.StreamRangeAsync(streamName, minId = Nullable(nextPosition), count = (Option.toNullable pollOptions.CountToPullATime))
        match response with
        | EmptyArray ->
            let nextPollDelay = pollOptions.CalculateNextPollDelay pollDelay
            do! Task.Delay pollDelay
            return Some ((nextPosition, nextPollDelay ) , Array.empty )
        | entries ->
            let lastEntry = Seq.last entries
            let nextPosition = EntryId.CalculateNextPositionIncr lastEntry.Id
            let nextPollDelay = TimeSpan.Zero
            return Some ((nextPosition, nextPollDelay), entries )

    }) (startingPosition, TimeSpan.Zero)
    |> Observable.flattenArray

Uh oh!

On error resume #2

Description

Metadata

Metadata

Assignees

Labels

Projects

Milestone

Relationships

Development

Issue actions