rename and stress tests

Jimmy Byrd · Jimmy Byrd · commit 5c0647c4710c · 2019-07-01T13:50:27.000-04:00
diff --git a/.vscode/settings.json b/.vscode/settings.json
@@ -0,0 +1,5 @@
+{
+    "workbench.colorCustomizations": {
+        "tab.unfocusedActiveBorder": "#fff0"
+    }
+}
diff --git a/FSharp.Control.Redis.Streams.sln b/FSharp.Control.Redis.Streams.sln
@@ -5,7 +5,7 @@ VisualStudioVersion = 15.0.26124.0
 MinimumVisualStudioVersion = 15.0.26124.0
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "src", "src", "{C397A34C-84F1-49E7-AEBC-2F9F2B196216}"
 EndProject
-Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "Redis.Streams", "src\Redis.Streams\Redis.Streams.fsproj", "{5D30E174-2538-47AC-8443-318C8C5DC2C9}"
+Project("{F2A71F9B-5D33-465A-A702-920D77279786}") = "FSharp.Control.Redis.Streams", "src\FSharp.Control.Redis.Streams\FSharp.Control.Redis.Streams.fsproj", "{5D30E174-2538-47AC-8443-318C8C5DC2C9}"
 EndProject
 Project("{2150E333-8FDC-42A3-9474-1A3956D46DE8}") = "tests", "tests", "{ACBEE43C-7A88-4FB1-9B06-DB064D22B29F}"
 EndProject
diff --git a/build.fsx b/build.fsx
@@ -24,8 +24,8 @@ BuildServer.install [
 // Metadata and Configuration
 //-----------------------------------------------------------------------------
 
-let productName = "Redis.Streams"
-let sln = "Redis.Streams.sln"
+let productName = "FSharp.Control.Redis.Streams"
+let sln = "FSharp.Control.Redis.Streams.sln"
 
 let srcGlob =__SOURCE_DIRECTORY__  @@ "src/**/*.??proj"
 let testsGlob = __SOURCE_DIRECTORY__  @@ "tests/**/*.??proj"
@@ -43,7 +43,7 @@ let coverageThresholdPercent = 80
 let coverageReportDir =  __SOURCE_DIRECTORY__  @@ "docs" @@ "coverage"
 
 let gitOwner = "MyGithubUsername"
-let gitRepoName = "Redis.Streams"
+let gitRepoName = "FSharp.Control.Redis.Streams"
 
 let releaseBranch = "master"
 let releaseNotes = Fake.Core.ReleaseNotes.load "RELEASE_NOTES.md"
diff --git a/docker-compose.yml b/docker-compose.yml
@@ -0,0 +1,6 @@
+version: "3.7"
+services:
+    redis:
+        image: redis:latest
+        ports:
+        - "6379:6379"
diff --git a/paket.dependencies b/paket.dependencies
@@ -12,6 +12,8 @@ nuget YoloDev.Expecto.TestSdk 0.8.0
 nuget Microsoft.NET.Test.Sdk 15.7.2
 nuget altcover ~> 5
 
+nuget Hopac
+nuget StackExchange.Redis
 // [ FAKE GROUP ]
 group Build
 
@@ -29,4 +31,4 @@ group Build
     nuget Fake.DotNet.Paket
     nuget Fake.Api.GitHub
     nuget Fake.BuildServer.AppVeyor
-    nuget Fake.BuildServer.Travis
+    nuget Fake.BuildServer.Travis
diff --git a/paket.lock b/paket.lock
diff --git a/src/FSharp.Control.Redis.Streams/AssemblyInfo.fs b/src/FSharp.Control.Redis.Streams/AssemblyInfo.fs
diff --git a/src/FSharp.Control.Redis.Streams/FSharp.Control.Redis.Streams.fsproj b/src/FSharp.Control.Redis.Streams/FSharp.Control.Redis.Streams.fsproj
@@ -1,7 +1,7 @@
 <?xml version="1.0" encoding="utf-8"?>
 <Project Sdk="Microsoft.NET.Sdk">
   <PropertyGroup>
-    <TargetFrameworks>netstandard2.0;net461</TargetFrameworks>
+    <TargetFrameworks>netstandard2.0</TargetFrameworks>
     <GenerateDocumentationFile>true</GenerateDocumentationFile>
   </PropertyGroup>
   <PropertyGroup>
diff --git a/src/FSharp.Control.Redis.Streams/Library.fs b/src/FSharp.Control.Redis.Streams/Library.fs
diff --git a/src/FSharp.Control.Redis.Streams/paket.references b/src/FSharp.Control.Redis.Streams/paket.references
@@ -1,2 +1,4 @@
 FSharp.Core
 Microsoft.SourceLink.GitHub
+StackExchange.Redis
+Hopac
diff --git a/tests/Redis.Streams.Tests/Tests.fs b/tests/Redis.Streams.Tests/Tests.fs
@@ -1,16 +1,240 @@
 module Tests
 
-
+open System
 open Expecto
 open Redis.Streams
+open StackExchange.Redis
+open Hopac
+
+let getUniqueKey (keyType : string) (key : string) =
+    let suffix = Guid.NewGuid().ToString()
+    sprintf "%s:%s:%s" keyType key suffix
+    |> RedisKey.op_Implicit
+
+let printNameEntryValues (nve : NameValueEntry) =
+    printfn "Key: %A - Value: %A" nve.Name nve.Value
+let printStreamEntry (se : StreamEntry) =
+    printfn "Id: %A" se.Id
+    se.Values |> Seq.iter(printNameEntryValues )
+let printRedisStream (rs : RedisStream) =
+    printfn "key: %A" rs.Key
+    rs.Entries |> Seq.iter(printStreamEntry)
+
+let rkey (s : string) = RedisKey.op_Implicit s
+let rval (s : string) = RedisValue.op_Implicit s
+let nve name value = NameValueEntry(rval name, rval value)
+
+let sp key position =
+    StreamPosition(key, position)
+
+//special keys
+
+//This special ID means that XREAD should use as last ID the maximum ID already stored in the stream mystream, so that we will receive only new messages, starting from the time we started listening. This is similar to the tail -f Unix command in some way.
+let dollar = "$"
+// Beginning
+let dash = "-"
+// End
+let plus = "+"
+
+let (|EmptySeq|_|) xs =
+    if xs |> Seq.isEmpty then Some EmptySeq
+    else None
+
+let (|Int64|_|) (str: string) =
+    match Int64.TryParse(str) with
+    | (true,v) -> Some v
+    | _ -> None
+
+type EntryId = {
+    MillisecondsTime : int64
+    SequenceNumber : int64
+}
+    with
+        member x.Unparse () =
+            sprintf "%d-%d" x.MillisecondsTime x.SequenceNumber
+
+        member x.ToRedisValue () =
+            x.Unparse() |> RedisValue.op_Implicit
+
+        member x.IncrementSequence () =
+            {x with SequenceNumber = x.SequenceNumber + 1L}
+
+        static member Parse(s : string) =
+            match s.Split('-') |> Array.toList with
+            | [Int64 ms; Int64 sn] -> {MillisecondsTime = ms; SequenceNumber = sn }
+            | [Int64 ms] ->{MillisecondsTime = ms; SequenceNumber = 0L }
+            | _ -> failwithf "invalid EntryId format, should look like 0-0 but was %s" s
+
+        static member Parse(s : RedisValue) =
+            s |> string |> EntryId.Parse
+
+
+type PollOptions = {
+    MaxPollDelay : TimeSpan
+    MaxPollDelayBuckets : float
+    CountToPullATime : int option
+}
+    with
+        static member Default = {
+            MaxPollDelay = TimeSpan.FromMilliseconds(100.)
+            MaxPollDelayBuckets = 10.
+            CountToPullATime = Some 1000
+        }
+
+let inline minTimeSpan (t1 : TimeSpan) (t2 : TimeSpan) =
+    Math.Min(t1.Ticks, t2.Ticks)
+    |> TimeSpan.FromTicks
+
+let pollStreamForever (redisdb : IDatabase) (streamName : RedisKey) (startingPosition : RedisValue) (pollOptions : PollOptions)  =
+    let incPos (rv : RedisValue) = EntryId.Parse(rv).IncrementSequence().ToRedisValue()
+    Stream.unfoldJob(fun (nextPosition, shouldDelay, nextDelayPeriod) -> job {
+        if shouldDelay then
+            do! timeOut pollOptions.MaxPollDelay
+        let! (response : StreamEntry []) = redisdb.StreamRangeAsync(streamName, minId = Nullable(nextPosition), count = (pollOptions.CountToPullATime |> Option.toNullable))
+        match response with
+        | EmptySeq ->
+            let increment = (float pollOptions.MaxPollDelay.Ticks / pollOptions.MaxPollDelayBuckets)
+            let nextPollDelay = nextDelayPeriod + TimeSpan.FromTicks(int64 increment)
+            let nextPollDelay = minTimeSpan nextPollDelay pollOptions.MaxPollDelay
+            return Some (Array.empty, (nextPosition, true, nextPollDelay ))
+        | entries ->
+            let lastEntry = entries |> Seq.last
+            let nextPosition = lastEntry.Id |> incPos
+            return Some (entries, (nextPosition, false, TimeSpan.FromMilliseconds(0.) ))
+    }) (startingPosition, false, TimeSpan.FromMilliseconds(0.))
+    |> Stream.appendMap (Stream.ofSeq)
+
+let pollStreamUntil (redisdb : IDatabase) (streamName : RedisKey) (startingPosition : RedisValue) (pollOptions : PollOptions) (until : Alt<_>) =
+    pollStreamForever redisdb streamName startingPosition pollOptions
+    |> Stream.takeUntil until
 
+
+let failDueToTimeout message (ts : TimeSpan) =
+    failtestf "%s. Expected task to complete but failed after timeout of %f ms"  message ts.TotalMilliseconds
+
+type StreamExpect<'a> (predicate : seq<'a> -> bool) =
+    let values = ResizeArray<'a>()
+    let predicateConditionMet = IVar<unit>()
+    let mutable debugPrint : Option<'a -> unit> = None
+    let checkPredicate () =
+        if predicate values then
+            IVar.tryFill predicateConditionMet () |> start
+
+    member this.Values
+        with get () = values
+
+
+    member this.DebugPrint
+        with get () = debugPrint
+        and  set (v) = debugPrint <- v
+    member this.CaptureFromStream (s : Stream<'a>) =
+        let printer =
+            match debugPrint with
+            | Some d -> d
+            | None -> ignore
+        s
+        //Debuging
+        |> Stream.mapFun(fun x -> x |> printer ; x)
+        |> Stream.iterFun(values.Add >> checkPredicate)
+        |> start
+
+    member this.Await message (timeout : TimeSpan) = job {
+
+        do! Alt.choose [
+            predicateConditionMet :> Alt<unit>
+            timeOut timeout |> Alt.afterFun (fun _ -> failDueToTimeout message timeout)
+        ]
+    }
+
+let ranStr n : string =
+    let r = Random()
+    String(Array.init n (fun _ -> char (r.Next(97,123))))
+
+
+
+let options = ConfigurationOptions.Parse("localhost", ResponseTimeout = 100000, SyncTimeout=100000, ConnectTimeout=100000)
+let redis = ConnectionMultiplexer.Connect(options)
 [<Tests>]
 let tests =
-  testList "samples" [
-    testCase "Say nothing" <| fun _ ->
-      let subject = Say.nothing ()
-      Expect.equal subject () "Not an absolute unit"
-    testCase "Say hello all" <| fun _ ->
-      let subject = Say.hello "all"
-      Expect.equal subject "Hello all" "You didn't say hello"
+    testSequenced <|
+    testList "samples" [
+    testCaseAsync "Stream should generate 2 events" <| async {
+        let db = redis.GetDatabase()
+        let key = getUniqueKey "stream" "Foo"
+        let expecter = StreamExpect<_>(fun s -> s |> Seq.length = 2)
+        pollStreamForever db key StreamPosition.Beginning PollOptions.Default
+        |> expecter.CaptureFromStream
+
+        let values =
+            [|
+                nve "Field1" "Value1"
+                nve "Field2" "Value3"
+            |]
+        let! x = db.StreamAddAsync(key, values) |> Async.AwaitTask
+        let values =
+            [|
+                nve "Field1" "Value4"
+                nve "Field2" "Value6"
+            |]
+        let! x = db.StreamAddAsync(key, values) |> Async.AwaitTask
+        do! expecter.Await "Should have 2 results" (TimeSpan.FromSeconds(1.)) |> Job.toAsync
+    }
+
+    testCaseAsync "Stream should generate 200000 events" <| async {
+        let total = 200000
+        let db = redis.GetDatabase()
+        let key = getUniqueKey "stream" "Foo"
+        let expecter = StreamExpect<_>(fun s -> s |> Seq.length = total)
+        pollStreamForever db key StreamPosition.Beginning PollOptions.Default
+        |> expecter.CaptureFromStream
+
+        job {
+            do!
+                [0..total]
+                |> Seq.map(fun i ->
+                    let values =
+                        [|
+                            NameValueEntry (RedisValue.op_Implicit "Field1", RedisValue.op_Implicit total)
+                        |]
+                    job {
+                        let! x = db.StreamAddAsync(key, values) |> Async.AwaitTask
+                        return ()
+                    })
+                |> Stream.ofSeq
+                |> Stream.mapPipelinedJob (Environment.ProcessorCount * 4096 * 2) id
+                |> Stream.iter
+        } |> start
+
+        do! expecter.Await "Should have 200000 results" (TimeSpan.FromSeconds(30.)) |> Job.toAsync
+    }
+
+
+    testCaseAsync "Stream should generate large fields" <| async {
+        let total = 20000
+        let db = redis.GetDatabase()
+        let key = getUniqueKey "stream" "Foo"
+        let expecter = StreamExpect<_>(fun s -> s |> Seq.length = total)
+        pollStreamForever db key StreamPosition.Beginning PollOptions.Default
+        |> expecter.CaptureFromStream
+
+        job {
+            do!
+                [0..total]
+                |> Seq.map(fun i ->
+                    let data = ranStr (2000000)
+                    let values =
+                        [|
+                            NameValueEntry (RedisValue.op_Implicit "Field1", RedisValue.op_Implicit data)
+                        |]
+                    job {
+                        let! x = db.StreamAddAsync(key, values) |> Async.AwaitTask
+                        return ()
+                    })
+                |> Stream.ofSeq
+                |> Stream.mapPipelinedJob (Environment.ProcessorCount * 2) id
+                |> Stream.iter
+        } |> start
+
+        do! expecter.Await "Should have 2 results" (TimeSpan.FromSeconds(300.)) |> Job.toAsync
+    }
   ]

-Original file line number
+Diff line change
@@ @@ -0,0 +1,5 @@ @@
 +{
 +    "workbench.colorCustomizations": {
 +        "tab.unfocusedActiveBorder": "#fff0"
 +    }
 +}