WIP

Author: Jamil Maqdis Anton

src/Read.fs

@@ -10,15 +10,16 @@ type ReadingDirection =
     | Backward
 
 module Read =
-    type private StartPositionInclusive = int64
-    let readFromAllStreamAsync: IStreamStore -> ReadingDirection -> StartPositionInclusive -> int -> Async<ReadAllPage> =
+    type StartPositionInclusive = int64
+
+    let readFromAllStreamAsync: IStreamStore -> ReadingDirection -> StartPositionInclusive -> MessageCount -> Async<ReadAllPage> =
         fun store readingDirection startPositionInclusive msgCount ->
             match readingDirection with
             | Forward -> store.ReadAllForwards(startPositionInclusive, msgCount)
             | Backward -> store.ReadAllBackwards(startPositionInclusive, msgCount)
             |> Async.AwaitTask
 
-    let readFromStreamAsync: IStreamStore -> ReadingDirection -> StreamDetails -> int -> Async<ReadStreamPage> =
+    let readFromStreamAsync: IStreamStore -> ReadingDirection -> StreamDetails -> MessageCount -> Async<ReadStreamPage> =
         fun store readingDirection streamDetails msgCount ->
             match readingDirection with
             | Forward ->
@@ -27,7 +28,7 @@ module Read =
                 store.ReadStreamBackwards(streamDetails.streamName, Helpers.toVersion streamDetails.version, msgCount)
             |> Async.AwaitTask
 
-    let readFromStreamAsync': IStreamStore -> ReadingDirection -> StreamDetails -> int -> CancellationToken -> Async<ReadStreamPage> =
+    let readFromStreamAsync': IStreamStore -> ReadingDirection -> StreamDetails -> MessageCount -> CancellationToken -> Async<ReadStreamPage> =
         fun store readingDirection streamDetails msgCount cancellationToken ->
             match readingDirection with
             | Forward ->
@@ -39,7 +40,21 @@ module Read =
             |> Async.AwaitTask
 
 module ReadExtras =
-    let readStreamMessages: IStreamStore -> ReadingDirection -> StreamDetails -> int -> AsyncResult<List<StreamMessage>, string> =
+    let readAllStreamMessages: IStreamStore -> ReadingDirection -> StartPositionInclusive -> MessageCount -> AsyncResult<List<StreamMessage>, string> =
+        fun store readingDirection startPositionInclusive msgCount ->
+            Read.readFromAllStreamAsync store readingDirection startPositionInclusive msgCount
+            |> Async.bind (fun readAllPage ->
+                readAllPage.Messages
+                |> Seq.toList
+                |> fun messageList ->
+                    if messageList.Length = msgCount then
+                        Ok messageList
+                    else
+                        Error
+                            (sprintf "Failed to retrieve all messages. Retrieved messages count: %d" messageList.Length)
+                |> AsyncResult.fromResult)
+
+    let readStreamMessages: IStreamStore -> ReadingDirection -> StreamDetails -> MessageCount -> AsyncResult<List<StreamMessage>, string> =
         fun store readingDirection streamDetails msgCount ->
             Read.readFromStreamAsync store readingDirection streamDetails msgCount
             |> Async.bind (fun readStreamPage ->

src/Types.fs

@@ -13,6 +13,9 @@ type StreamDetails =
     { streamName: string
       version: Version }
 
+type StartPositionInclusive = int64
+type MessageCount = int
+
 module Helpers =
     let toVersion: Version -> int =
         function