Adds Reactive Extensions

Author: Jimmy Byrd

paket.dependencies

@@ -16,7 +16,7 @@ nuget StackExchange.Redis
 nuget Hopac
 nuget Akka.Streams
 nuget TaskBuilder.fs
-
+nuget System.Reactive
 
 // [ FAKE GROUP ]
 group Build

paket.lock

@@ -0,0 +1,23 @@
+// Auto-Generated by FAKE; do not edit
+namespace System
+open System.Reflection
+
+[<assembly: AssemblyTitleAttribute("Redis.Streams")>]
+[<assembly: AssemblyProductAttribute("Redis.Streams")>]
+[<assembly: AssemblyVersionAttribute("0.1.0")>]
+[<assembly: AssemblyMetadataAttribute("ReleaseDate","2017-03-17T00:00:00.0000000")>]
+[<assembly: AssemblyFileVersionAttribute("0.1.0")>]
+[<assembly: AssemblyInformationalVersionAttribute("0.1.0")>]
+[<assembly: AssemblyMetadataAttribute("ReleaseChannel","release")>]
+[<assembly: AssemblyMetadataAttribute("GitHash","bb8964b54bee133e9af64d316dc2cfee16df7f72")>]
+do ()
+
+module internal AssemblyVersionInformation =
+    let [<Literal>] AssemblyTitle = "Redis.Streams"
+    let [<Literal>] AssemblyProduct = "Redis.Streams"
+    let [<Literal>] AssemblyVersion = "0.1.0"
+    let [<Literal>] AssemblyMetadata_ReleaseDate = "2017-03-17T00:00:00.0000000"
+    let [<Literal>] AssemblyFileVersion = "0.1.0"
+    let [<Literal>] AssemblyInformationalVersion = "0.1.0"
+    let [<Literal>] AssemblyMetadata_ReleaseChannel = "release"
+    let [<Literal>] AssemblyMetadata_GitHash = "bb8964b54bee133e9af64d316dc2cfee16df7f72"

src/FSharp.Control.Redis.Streams.Reactive/AssemblyInfo.fs

@@ -0,0 +1,49 @@
+namespace FSharp.Control.Redis.Streams
+
+module Hopac =
+    open System
+    open System.Threading
+    open System.Threading.Tasks
+    open System.Reactive
+    open System.Reactive.Linq
+    open StackExchange.Redis
+    open FSharp.Control.Redis.Streams.Core
+    open FSharp.Control.Tasks.V2.ContextInsensitive
+
+
+    let pollStreamForever (redisdb : IDatabase) (streamName : RedisKey) (startingPosition : RedisValue) (pollOptions : PollOptions) =
+        let calculateNextPollDelay (nextPollDelay) =
+            let increment = (float pollOptions.MaxPollDelay.Ticks / pollOptions.MaxPollDelayBuckets)
+            let nextPollDelay = nextPollDelay + TimeSpan.FromTicks(int64 increment)
+            TimeSpan.Min nextPollDelay pollOptions.MaxPollDelay
+
+        Observable.Create(fun (obs : IObserver<_>) ->
+            let cts = new CancellationTokenSource()
+            let ct = cts.Token
+            task {
+                let mutable nextPollDelay = TimeSpan.Zero
+                let mutable nextPosition = startingPosition
+                try
+                    try
+                        while not ct.IsCancellationRequested do
+                            let! (response : StreamEntry []) = redisdb.StreamRangeAsync(streamName, minId = Nullable(nextPosition), count = (Option.toNullable pollOptions.CountToPullATime))
+                            match response with
+                            | EmptySeq ->
+                                nextPollDelay <- calculateNextPollDelay nextPollDelay
+                                do! Task.Delay(nextPollDelay, ct)
+                            | entries ->
+                                let lastEntry = Seq.last entries
+                                nextPosition <- EntryId.CalculateNextPosition lastEntry.Id
+                                nextPollDelay <- TimeSpan.Zero
+                                entries |> Array.iter obs.OnNext
+                        obs.OnCompleted()
+                    with e ->
+                        obs.OnError e
+                finally
+                    cts.Dispose()
+            } |> ignore
+
+            new Disposables.CancellationDisposable(cts) :> IDisposable
+        )
+
+

src/FSharp.Control.Redis.Streams.Reactive/FSharp.Control.Redis.Streams.Reactive.fs

@@ -0,0 +1,27 @@
+<?xml version="1.0" encoding="utf-8"?>
+<Project Sdk="Microsoft.NET.Sdk">
+  <PropertyGroup>
+    <TargetFrameworks>netstandard2.0;net461</TargetFrameworks>
+    <GenerateDocumentationFile>true</GenerateDocumentationFile>
+  </PropertyGroup>
+  <PropertyGroup>
+    <Title>FSharp.Control.Redis.Streams.Reactive</Title>
+    <Description>FSharp.Control.Redis.Streams.Reactive does the thing!</Description>
+
+  </PropertyGroup>
+  <PropertyGroup Condition="'$(Configuration)'=='Release'">
+    <Optimize>true</Optimize>
+    <Tailcalls>true</Tailcalls>
+    
+  </PropertyGroup>
+  <ItemGroup>
+    <Compile Include="AssemblyInfo.fs" />
+    <Compile Include="FSharp.Control.Redis.Streams.Reactive.fs" />
+  </ItemGroup>
+  <ItemGroup>
+    <ProjectReference Include="..\FSharp.Control.Redis.Streams\FSharp.Control.Redis.Streams.fsproj" />
+  </ItemGroup>
+  <!-- <Import Project="..\..\fsc.props" /> -->
+  <Import Project="..\..\netfx.props" />
+  <Import Project="..\..\.paket\Paket.Restore.targets" />
+</Project>

src/FSharp.Control.Redis.Streams.Reactive/FSharp.Control.Redis.Streams.Reactive.fsproj

@@ -0,0 +1,5 @@
+FSharp.Core
+Microsoft.SourceLink.GitHub
+StackExchange.Redis
+System.Reactive
+TaskBuilder.fs