Added Pulse facility on Projector

Author: bartelink

FunDomain.Persistence.NEventStore.Acceptance/InMemoryStore.fs

@@ -4,8 +4,7 @@ open Uno // Card Builders
 open Uno.Game // Commands, handle
 
 open FunDomain // CommandHandler, Evolution.replay
-open FunDomain.Persistence.NEventStore.NesGateway // createInMemory, StreamId
-open FunDomain.Persistence.NEventStore // NesProjector
+open FunDomain.Persistence.NEventStore // Projector, StreamId
 
 open Xunit
 open Swensen.Unquote
@@ -48,23 +47,24 @@ let gameStreamId (GameId no) = {Bucket=None; StreamId=string no }
 let [<Fact>] ``Can run a full round using NEventStore's InMemoryPersistence`` () =
     let domainHandler = CommandHandler.create replay handle 
 
-    let store = createInMemory()
-    let persistingHandler = domainHandler store.read store.append 
-    
     let monitor = DirectionMonitor()
 
+    let store = NesGateway.createInMemory()
+    let projector = Projector( store, 10, (fun batch ->
+        batch.chooseOfUnion () |> Seq.iter (fun evt ->
+            monitor.Post evt
+            logger.Post evt)))
+
+    let persistingHandler = domainHandler store.read store.append
+
     let gameId = GameId 42
     let streamId = gameStreamId gameId
     for action in fullGameActions gameId do 
         printfn "Processing %A against Stream %A" action streamId
         action |> persistingHandler streamId
+        projector.Pulse ()
 
-    NesProjector.start store 10 (fun batch ->
-        batch.chooseOfUnion () |> Seq.iter (fun evt ->
-            monitor.Post evt
-            logger.Post evt))
-
-    Async.AwaitEvent NesProjector.sleeping
+    Async.AwaitEvent projector.sleeping
     |> Async.RunSynchronously
     printfn "Projection queue empty"
 

FunDomain.Persistence.NEventStore/FunDomain.Persistence.NEventStore.fsproj

@@ -57,7 +57,7 @@
       <Link>Serialization.fs</Link>
     </Compile>
     <Compile Include="NesGateway.fs" />
-    <Compile Include="NesProjector.fs" />
+    <Compile Include="Projector.fs" />
   </ItemGroup>
   <ItemGroup>
     <Reference Include="FSharp.Core, Version=$(TargetFSharpCoreVersion), Culture=neutral, PublicKeyToken=b03f5f7f11d50a3a">

FunDomain.Persistence.NEventStore/NesGateway.fs

@@ -1,4 +1,4 @@
-module FunDomain.Persistence.NEventStore.NesGateway
+namespace FunDomain.Persistence.NEventStore
 
 open FunDomain.Persistence.Serialization
 
@@ -24,7 +24,7 @@ type EncodedEventBatch (events) =
         cached |> Seq.choose (fun (ee:EncodedEvent) -> ee.deserializeUnionByCaseItemType<'a> ())
 
 /// Wrapper yielded by create* functions with create/append functions matching FunDomain.CommandHandler requirements
-type Streamer private (inner') = 
+type Store private (inner') = 
     // Hoop jumping a la C++ pimpl pattern - if we don't do this, we're foisting an NEventStore package reference on all downstream users
     let inner : IPersistStreams = unbox inner'
     let defaultBucket bucketId = defaultArg bucketId "default"
@@ -89,7 +89,7 @@ type Streamer private (inner') =
             Some checkpoint
         batch |> Seq.fold dispatchElements None
 
-    static member internal wrap persister = Streamer( box persister)
+    static member internal wrap persister = Store( box persister)
 
     member this.read<'a> stream = 
         let commits,version,_ = 
@@ -110,24 +110,25 @@ type Streamer private (inner') =
         appendToStream stream metadata token encodedEvents
         |> Async.RunSynchronously
 
-let createFromStore (inner:IStoreEvents) = 
-    inner.Advanced |> Streamer.wrap
-
-let createInMemory () = 
-    Wireup.Init()
-        .LogToOutputWindow()
-        .UsingInMemoryPersistence()
-        .UsingJsonSerialization()
-        .Build()
-    |> createFromStore
-
-let createInMsSqlWithPerfCounters (connectionString:string) perfCounterSetName = 
-    Wireup.Init()
-        .UsingSqlPersistence(connectionString)
-        .WithDialect(new MsSqlDialect())
-        .InitializeStorageEngine()
-        .TrackPerformanceInstance(perfCounterSetName)
-        .UsingJsonSerialization()
-            .Compress()
-        .Build()
-    |> createFromStore
+module NesGateway =
+    let createFromStore (inner:IStoreEvents) = 
+        inner.Advanced |> Store.wrap
+
+    let createInMemory () = 
+        Wireup.Init()
+            .LogToOutputWindow()
+            .UsingInMemoryPersistence()
+            .UsingJsonSerialization()
+            .Build()
+        |> createFromStore
+
+    let createInMsSqlWithPerfCounters (connectionString:string) perfCounterSetName = 
+        Wireup.Init()
+            .UsingSqlPersistence(connectionString)
+            .WithDialect(new MsSqlDialect())
+            .InitializeStorageEngine()
+            .TrackPerformanceInstance(perfCounterSetName)
+            .UsingJsonSerialization()
+                .Compress()
+            .Build()
+        |> createFromStore

FunDomain.Persistence.NEventStore/NesProjector.fs

@@ -0,0 +1,23 @@
+namespace FunDomain.Persistence.NEventStore
+
+open System.Threading
+
+type Projector( store:Store, sleepMs, projection ) = 
+    let empty = new Event<unit>()
+    let wakeEvent = new AutoResetEvent false
+    let _ =
+        MailboxProcessor.Start <|
+            fun inbox ->
+                let rec loop token = async {
+                    match token |> store.project projection with
+                    | Some token -> 
+                        return! loop token
+                    | _ -> 
+                        empty.Trigger ()
+                        let! _ = Async.AwaitWaitHandle (wakeEvent,sleepMs)
+                        return! loop token }
+
+                async {
+                    return! loop CheckpointToken.initial }
+    member this.sleeping = empty.Publish
+    member this.Pulse = wakeEvent.Set >> ignore