Refactor async.Delay calls to more familiar async blocks. Await tasks with awaitTaskWithInnerException instead of normal AwaitTask to handle AggregateExceptions.

Author: Jamil Maqdis Anton

src/AppendRaw.fs

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

src/Postgres.fs

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

src/ReadRaw.fs

@@ -21,41 +21,45 @@ module ReadRaw =
                           (startPositionInclusive: StartPosition)
                           (msgCount: int)
                           : Async<ReadAllPage> =
-        async.Delay(fun () ->
-            match readingDirection with
-            | ReadingDirection.Forward ->
-                store.ReadAllForwards(fromStartPositionInclusive startPositionInclusive, msgCount)
-            | ReadingDirection.Backward ->
-                store.ReadAllBackwards(fromStartPositionInclusive startPositionInclusive, msgCount)
-            |> Async.AwaitTask)
+        async {
+            return! match readingDirection with
+                    | ReadingDirection.Forward ->
+                        store.ReadAllForwards(fromStartPositionInclusive startPositionInclusive, msgCount)
+                    | ReadingDirection.Backward ->
+                        store.ReadAllBackwards(fromStartPositionInclusive startPositionInclusive, msgCount)
+                    |> Async.awaitTaskWithInnerException
+        }
 
     let readFromStream (store: SqlStreamStore.IStreamStore)
                        (readingDirection: ReadingDirection)
                        (streamName: string)
                        (readVersion: ReadVersion)
                        (msgCount: int)
                        : Async<ReadStreamPage> =
-        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)
+        async {
+            return! match readingDirection with
+                    | ReadingDirection.Forward ->
+                        store.ReadStreamForwards(StreamId(streamName), fromReadVersion readVersion, msgCount)
+                    | ReadingDirection.Backward ->
+                        store.ReadStreamBackwards(StreamId(streamName), fromReadVersion readVersion, msgCount)
+                    |> Async.awaitTaskWithInnerException
+        }
 
     let readFromAllStream' (store: SqlStreamStore.IStreamStore)
                            (readingDirection: ReadingDirection)
                            (startPositionInclusive: StartPosition)
                            (msgCount: int)
                            (prefetchJson: bool)
                            : Async<ReadAllPage> =
-        async.Delay(fun () ->
-            match readingDirection with
-            | ReadingDirection.Forward ->
-                store.ReadAllForwards(fromStartPositionInclusive startPositionInclusive, msgCount, prefetchJson)
-            | ReadingDirection.Backward ->
-                store.ReadAllBackwards(fromStartPositionInclusive startPositionInclusive, msgCount, prefetchJson)
-            |> Async.AwaitTask)
+        async {
+            return! match readingDirection with
+                    | ReadingDirection.Forward ->
+                        store.ReadAllForwards(fromStartPositionInclusive startPositionInclusive, msgCount, prefetchJson)
+                    | ReadingDirection.Backward ->
+                        store.ReadAllBackwards
+                            (fromStartPositionInclusive startPositionInclusive, msgCount, prefetchJson)
+                    |> Async.awaitTaskWithInnerException
+        }
 
     let readFromStream' (store: SqlStreamStore.IStreamStore)
                         (readingDirection: ReadingDirection)
@@ -64,13 +68,16 @@ module ReadRaw =
                         (msgCount: int)
                         (prefetchJson: bool)
                         : Async<ReadStreamPage> =
-        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)
+        async {
+            return! match readingDirection with
+                    | ReadingDirection.Forward ->
+                        store.ReadStreamForwards
+                            (StreamId(streamName), fromReadVersion readVersion, msgCount, prefetchJson)
+                    | ReadingDirection.Backward ->
+                        store.ReadStreamBackwards
+                            (StreamId(streamName), fromReadVersion readVersion, msgCount, prefetchJson)
+                    |> Async.awaitTaskWithInnerException
+        }
 
     let readFromAllStream'' (store: SqlStreamStore.IStreamStore)
                             (readingDirection: ReadingDirection)
@@ -79,15 +86,22 @@ module ReadRaw =
                             (prefetchJson: bool)
                             (cancellationToken: CancellationToken)
                             : Async<ReadAllPage> =
-        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)
+        async {
+            return! match readingDirection with
+                    | ReadingDirection.Forward ->
+                        store.ReadAllForwards
+                            (fromStartPositionInclusive startPositionInclusive,
+                             msgCount,
+                             prefetchJson,
+                             cancellationToken)
+                    | ReadingDirection.Backward ->
+                        store.ReadAllBackwards
+                            (fromStartPositionInclusive startPositionInclusive,
+                             msgCount,
+                             prefetchJson,
+                             cancellationToken)
+                    |> Async.awaitTaskWithInnerException
+        }
 
     let readFromStream'' (store: SqlStreamStore.IStreamStore)
                          (readingDirection: ReadingDirection)
@@ -97,12 +111,21 @@ module ReadRaw =
                          (prefetchJson: bool)
                          (cancellationToken: CancellationToken)
                          : Async<ReadStreamPage> =
-        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)
+        async {
+            return! 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.awaitTaskWithInnerException
+        }