Actually map batches

jet · Jun 7, 2019 · 1abf73c · 1abf73c
1 parent 621516d
commit 1abf73c
Show file tree

Hide file tree

Showing 4 changed files with 3 additions and 1 deletion.
diff --git a/src/Propulsion.Cosmos/CosmosSink.fs b/src/Propulsion.Cosmos/CosmosSink.fs
@@ -173,6 +173,7 @@ type CosmosSink =
         let dumpStreams (s : Scheduling.StreamStates<_>) l = s.Dump(l, Propulsion.Streams.Buffering.StreamState.eventsSize, categorize)
         let streamScheduler = Internal.CosmosSchedulingEngine.Create(log, cosmosContexts, dispatcher, projectionStats, dumpStreams)
         let mapBatch onCompletion (x : Submission.SubmissionBatch<StreamEvent<_>>) : Scheduling.StreamsBatch<_> =
+            let onCompletion () = x.onCompletion(); onCompletion()
             Scheduling.StreamsBatch.Create(onCompletion, x.messages) |> fst
         let submitBatch (x : Scheduling.StreamsBatch<_>) : int =
             streamScheduler.Submit x

diff --git a/src/Propulsion.EventStore/EventStoreSink.fs b/src/Propulsion.EventStore/EventStoreSink.fs
@@ -151,6 +151,7 @@ type EventStoreSink =
         let dumpStats (s : Scheduling.StreamStates<_>) l = s.Dump(l, Propulsion.Streams.Buffering.StreamState.eventsSize, categorize)
         let streamScheduler = Internal.EventStoreSchedulingEngine.Create(log, storeLog, conns, dispatcher, projectionStats, dumpStats)
         let mapBatch onCompletion (x : Submission.SubmissionBatch<StreamEvent<_>>) : Scheduling.StreamsBatch<_> =
+            let onCompletion () = x.onCompletion(); onCompletion()
             Scheduling.StreamsBatch.Create(onCompletion, x.messages) |> fst
         let submitBatch (x : Scheduling.StreamsBatch<_>) : int =
             streamScheduler.Submit x

diff --git a/src/Propulsion.EventStore/EventStoreSource.fs b/src/Propulsion.EventStore/EventStoreSource.fs
@@ -95,6 +95,5 @@ type EventStoreSource =
                 let cp = pos.CommitPosition
                 striper.Submit <| Message.Batch(seriesId, cp, checkpoints.Commit cp, xs)
         let reader = Reader.EventStoreReader(conns, spec.batchSize, spec.minBatchSize, categorize, tryMapEvent, post, spec.tailInterval, dop)
-        log.Warning("Pumping reader")
         do! reader.Pump(initialSeriesId, startPos, maxPos) 
         log.Warning("Finished Pumping reader") }
diff --git a/src/Propulsion/Streams.fs b/src/Propulsion/Streams.fs
@@ -602,6 +602,7 @@ module Projector =
         static member Start(log : Serilog.ILogger, pumpDispatcher, pumpScheduler, maxReadAhead, submitStreamsBatch, statsInterval, ?ingesterStatsInterval, ?maxSubmissionsPerPartition) =
             let maxSubmissionsPerPartition = defaultArg maxSubmissionsPerPartition 5
             let mapBatch onCompletion (x : Submission.SubmissionBatch<StreamEvent<_>>) : Scheduling.StreamsBatch<_> =
+                let onCompletion () = x.onCompletion(); onCompletion()
                 Scheduling.StreamsBatch.Create(onCompletion, x.messages) |> fst
             let submitBatch (x : Scheduling.StreamsBatch<_>) : int =
                 submitStreamsBatch x