Wrap all code that uses Tasks with async.Delay, so that it behaves properly and doesn't execute until the Async computation is run. (#5)

Co-authored-by: Simon Herteby <simon.herteby@insurello.se>

Author: Herteby

src/AppendRaw.fs

@@ -26,21 +26,23 @@ module AppendRaw =
                          (appendVersion: AppendVersion)
                          (messageDetails: MessageDetails)
                          : Async<AppendResult> =
-        store.AppendToStream
-            (StreamId(streamName),
-             fromAppendVersion appendVersion,
-             [| newStreamMessageFromMessageDetails messageDetails |])
-        |> Async.AwaitTask
+        async.Delay(fun () ->
+            store.AppendToStream
+                (StreamId(streamName),
+                 fromAppendVersion appendVersion,
+                 [| newStreamMessageFromMessageDetails messageDetails |])
+            |> Async.AwaitTask)
 
     let appendNewMessages (store: SqlStreamStore.IStreamStore)
                           (streamName: string)
                           (appendVersion: AppendVersion)
                           (messages: MessageDetails list)
                           : Async<AppendResult> =
-        store.AppendToStream
-            (StreamId(streamName),
-             fromAppendVersion appendVersion,
-             messages
-             |> List.map newStreamMessageFromMessageDetails
-             |> List.toArray)
-        |> Async.AwaitTask
+        async.Delay(fun () ->
+            store.AppendToStream
+                (StreamId(streamName),
+                 fromAppendVersion appendVersion,
+                 messages
+                 |> List.map newStreamMessageFromMessageDetails
+                 |> List.toArray)
+            |> Async.AwaitTask)

src/Postgres.fs

@@ -61,7 +61,7 @@ module Postgres =
         new SqlStreamStore.PostgresStreamStore(SqlStreamStore.PostgresStreamStoreSettings(config))
 
     let createSchemaRaw (store: SqlStreamStore.PostgresStreamStore): Async<unit> =
-        store.CreateSchemaIfNotExists() |> Async.AwaitTask
+        async.Delay(fun () -> store.CreateSchemaIfNotExists() |> Async.AwaitTask)
 
     let createSchema (store: SqlStreamStore.PostgresStreamStore): Async<Result<unit, string>> =
         createSchemaRaw store

src/ReadRaw.fs

@@ -21,37 +21,41 @@ module ReadRaw =
                           (startPositionInclusive: StartPosition)
                           (msgCount: int)
                           : Async<ReadAllPage> =
-        match readingDirection with
-        | ReadingDirection.Forward -> store.ReadAllForwards(fromStartPositionInclusive startPositionInclusive, msgCount)
-        | ReadingDirection.Backward ->
-            store.ReadAllBackwards(fromStartPositionInclusive startPositionInclusive, msgCount)
-        |> Async.AwaitTask
+        async.Delay(fun () ->
+            match readingDirection with
+            | ReadingDirection.Forward ->
+                store.ReadAllForwards(fromStartPositionInclusive startPositionInclusive, msgCount)
+            | ReadingDirection.Backward ->
+                store.ReadAllBackwards(fromStartPositionInclusive startPositionInclusive, msgCount)
+            |> Async.AwaitTask)
 
     let readFromStream (store: SqlStreamStore.IStreamStore)
                        (readingDirection: ReadingDirection)
                        (streamName: string)
                        (readVersion: ReadVersion)
                        (msgCount: int)
                        : Async<ReadStreamPage> =
-        match readingDirection with
-        | ReadingDirection.Forward ->
-            store.ReadStreamForwards(StreamId(streamName), fromReadVersion readVersion, msgCount)
-        | ReadingDirection.Backward ->
-            store.ReadStreamBackwards(StreamId(streamName), fromReadVersion readVersion, msgCount)
-        |> Async.AwaitTask
+        async.Delay(fun () ->
+            match readingDirection with
+            | ReadingDirection.Forward ->
+                store.ReadStreamForwards(StreamId(streamName), fromReadVersion readVersion, msgCount)
+            | ReadingDirection.Backward ->
+                store.ReadStreamBackwards(StreamId(streamName), fromReadVersion readVersion, msgCount)
+            |> Async.AwaitTask)
 
     let readFromAllStream' (store: SqlStreamStore.IStreamStore)
                            (readingDirection: ReadingDirection)
                            (startPositionInclusive: StartPosition)
                            (msgCount: int)
                            (prefetchJson: bool)
                            : Async<ReadAllPage> =
-        match readingDirection with
-        | ReadingDirection.Forward ->
-            store.ReadAllForwards(fromStartPositionInclusive startPositionInclusive, msgCount, prefetchJson)
-        | ReadingDirection.Backward ->
-            store.ReadAllBackwards(fromStartPositionInclusive startPositionInclusive, msgCount, prefetchJson)
-        |> Async.AwaitTask
+        async.Delay(fun () ->
+            match readingDirection with
+            | ReadingDirection.Forward ->
+                store.ReadAllForwards(fromStartPositionInclusive startPositionInclusive, msgCount, prefetchJson)
+            | ReadingDirection.Backward ->
+                store.ReadAllBackwards(fromStartPositionInclusive startPositionInclusive, msgCount, prefetchJson)
+            |> Async.AwaitTask)
 
     let readFromStream' (store: SqlStreamStore.IStreamStore)
                         (readingDirection: ReadingDirection)
@@ -60,12 +64,13 @@ module ReadRaw =
                         (msgCount: int)
                         (prefetchJson: bool)
                         : Async<ReadStreamPage> =
-        match readingDirection with
-        | ReadingDirection.Forward ->
-            store.ReadStreamForwards(StreamId(streamName), fromReadVersion readVersion, msgCount, prefetchJson)
-        | ReadingDirection.Backward ->
-            store.ReadStreamBackwards(StreamId(streamName), fromReadVersion readVersion, msgCount, prefetchJson)
-        |> Async.AwaitTask
+        async.Delay(fun () ->
+            match readingDirection with
+            | ReadingDirection.Forward ->
+                store.ReadStreamForwards(StreamId(streamName), fromReadVersion readVersion, msgCount, prefetchJson)
+            | ReadingDirection.Backward ->
+                store.ReadStreamBackwards(StreamId(streamName), fromReadVersion readVersion, msgCount, prefetchJson)
+            |> Async.AwaitTask)
 
     let readFromAllStream'' (store: SqlStreamStore.IStreamStore)
                             (readingDirection: ReadingDirection)
@@ -74,14 +79,15 @@ module ReadRaw =
                             (prefetchJson: bool)
                             (cancellationToken: CancellationToken)
                             : Async<ReadAllPage> =
-        match readingDirection with
-        | ReadingDirection.Forward ->
-            store.ReadAllForwards
-                (fromStartPositionInclusive startPositionInclusive, msgCount, prefetchJson, cancellationToken)
-        | ReadingDirection.Backward ->
-            store.ReadAllBackwards
-                (fromStartPositionInclusive startPositionInclusive, msgCount, prefetchJson, cancellationToken)
-        |> Async.AwaitTask
+        async.Delay(fun () ->
+            match readingDirection with
+            | ReadingDirection.Forward ->
+                store.ReadAllForwards
+                    (fromStartPositionInclusive startPositionInclusive, msgCount, prefetchJson, cancellationToken)
+            | ReadingDirection.Backward ->
+                store.ReadAllBackwards
+                    (fromStartPositionInclusive startPositionInclusive, msgCount, prefetchJson, cancellationToken)
+            |> Async.AwaitTask)
 
     let readFromStream'' (store: SqlStreamStore.IStreamStore)
                          (readingDirection: ReadingDirection)
@@ -91,11 +97,12 @@ module ReadRaw =
                          (prefetchJson: bool)
                          (cancellationToken: CancellationToken)
                          : Async<ReadStreamPage> =
-        match readingDirection with
-        | ReadingDirection.Forward ->
-            store.ReadStreamForwards
-                (StreamId(streamName), fromReadVersion readVersion, msgCount, prefetchJson, cancellationToken)
-        | ReadingDirection.Backward ->
-            store.ReadStreamBackwards
-                (StreamId(streamName), fromReadVersion readVersion, msgCount, prefetchJson, cancellationToken)
-        |> Async.AwaitTask
+        async.Delay(fun () ->
+            match readingDirection with
+            | ReadingDirection.Forward ->
+                store.ReadStreamForwards
+                    (StreamId(streamName), fromReadVersion readVersion, msgCount, prefetchJson, cancellationToken)
+            | ReadingDirection.Backward ->
+                store.ReadStreamBackwards
+                    (StreamId(streamName), fromReadVersion readVersion, msgCount, prefetchJson, cancellationToken)
+            |> Async.AwaitTask)