Add NewStreamMessage wrapper and functions (#18)

Author: jamil7

src/SqlStreamStore.FSharp/Append.fs

@@ -2,7 +2,7 @@ namespace SqlStreamStore.FSharp
 
 open FSharp.Prelude
 open System.Threading
-open SqlStreamStore.Streams
+open SqlStreamStore
 
 [<RequireQualifiedAccess>]
 type AppendOption =
@@ -13,9 +13,10 @@ module Append =
     let streamMessages'
         (messages: NewStreamMessage list)
         (appendOptions: AppendOption list)
-        : Stream -> AsyncResult<AppendResult, exn> =
+        (Stream stream: Stream)
+        : AsyncResult<Streams.AppendResult, exn> =
 
-        let mutable expectedVersion = ExpectedVersion.Any
+        let mutable expectedVersion = Streams.ExpectedVersion.Any
         let mutable cancellationToken = Unchecked.defaultof<CancellationToken>
 
         appendOptions
@@ -24,8 +25,10 @@ module Append =
             | AppendOption.CancellationToken token -> cancellationToken <- token
             | AppendOption.ExpectedVersion version -> expectedVersion <- version)
 
-        fun (Stream stream) ->
-            stream.store.AppendToStream(stream.streamId, expectedVersion, List.toArray messages, cancellationToken)
+        let messages' =
+            List.map NewStreamMessage.toOriginalNewStreamMessage messages
 
-    let streamMessages (messages: NewStreamMessage list) : Stream -> AsyncResult<AppendResult, exn> =
+        stream.store.AppendToStream(stream.streamId, expectedVersion, List.toArray messages', cancellationToken)
+
+    let streamMessages (messages: NewStreamMessage list) : Stream -> AsyncResult<Streams.AppendResult, exn> =
         streamMessages' messages []

…Store.FSharp/SqlStreamStoreExtensions.fs src/SqlStreamStore.FSharp/Extensions.fssrc/SqlStreamStore.FSharp/SqlStreamStoreExtensions.fs renamed to src/SqlStreamStore.FSharp/Extensions.fs src/SqlStreamStore.FSharp/SqlStreamStoreExtensions.fs renamed to src/SqlStreamStore.FSharp/Extensions.fs

@@ -0,0 +1,44 @@
+namespace SqlStreamStore.FSharp
+
+type NewStreamMessageInternal =
+    {
+        messageId: System.Guid option
+        messageType: string
+        jsonData: string
+        jsonMetadata: string option
+    }
+
+type NewStreamMessage = private NewStreamMessage of NewStreamMessageInternal
+
+module NewStreamMessage =
+
+    let create (messageType: string) (jsonData: string) : NewStreamMessage =
+        NewStreamMessage
+            {
+                messageId = None
+                messageType = messageType
+                jsonData = jsonData
+                jsonMetadata = None
+            }
+
+    let withMessageId (messageId: System.Guid) (NewStreamMessage msg: NewStreamMessage) : NewStreamMessage =
+        NewStreamMessage { msg with messageId = Some messageId }
+
+    let withJsonMetadata (jsonMetadata: string) (NewStreamMessage msg: NewStreamMessage) : NewStreamMessage =
+        NewStreamMessage
+            { msg with
+                jsonMetadata = Some jsonMetadata
+            }
+
+    let internal toOriginalNewStreamMessage
+        (NewStreamMessage msg: NewStreamMessage)
+        : SqlStreamStore.Streams.NewStreamMessage =
+        let id =
+            match msg.messageId with
+            | Some id -> id
+            | None -> System.Guid.NewGuid()
+
+        match msg.jsonMetadata with
+        | Some metadata -> SqlStreamStore.Streams.NewStreamMessage(id, msg.messageType, msg.jsonData, metadata)
+        | None -> SqlStreamStore.Streams.NewStreamMessage(id, msg.messageType, msg.jsonData)
+

src/SqlStreamStore.FSharp/NewStreamMessage.fs

@@ -33,7 +33,7 @@ type ReadAllOption =
 
 module Read =
 
-    let partial' (readOptions: ReadPartialOption list) : Stream -> AsyncResult<ReadStreamPage, exn> =
+    let partial' (readOptions: ReadPartialOption list) (Stream stream: Stream) : AsyncResult<ReadStreamPage, exn> =
 
         let mutable cancellationToken = Unchecked.defaultof<CancellationToken>
         let mutable fromVersionInclusive : int option = None
@@ -59,25 +59,25 @@ module Read =
             | ReadDirection.Forward, Some index -> index
             | _ -> failwith "Illegal ReadDirection enum."
 
-        fun (Stream stream) ->
-            match readDirection with
-            | ReadDirection.Forward ->
-                stream.store.ReadStreamForwards(
-                    stream.streamId,
-                    fromVersionInclusive',
-                    messageCount,
-                    prefetch,
-                    cancellationToken
-                )
-            | ReadDirection.Backward ->
-                stream.store.ReadStreamBackwards(
-                    stream.streamId,
-                    fromVersionInclusive',
-                    messageCount,
-                    prefetch,
-                    cancellationToken
-                )
-            | _ -> failwith "Illegal ReadDirection enum."
+
+        match readDirection with
+        | ReadDirection.Forward ->
+            stream.store.ReadStreamForwards(
+                stream.streamId,
+                fromVersionInclusive',
+                messageCount,
+                prefetch,
+                cancellationToken
+            )
+        | ReadDirection.Backward ->
+            stream.store.ReadStreamBackwards(
+                stream.streamId,
+                fromVersionInclusive',
+                messageCount,
+                prefetch,
+                cancellationToken
+            )
+        | _ -> failwith "Illegal ReadDirection enum."
 
     let partial : Stream -> AsyncResult<ReadStreamPage, exn> = partial' []
 

src/SqlStreamStore.FSharp/Read.fs

@@ -12,21 +12,22 @@
     </PropertyGroup>
 
     <ItemGroup>
-        <PackageReference Update="FSharp.Core" Version="5.0.0"/>
-        <PackageReference Include="FSharp.Core" Version="5.0.0"/>
-        <PackageReference Include="FSharp.Prelude" Version="4.0.3"/>
-        <PackageReference Include="Npgsql" Version="5.0.4"/>
-        <PackageReference Include="SqlStreamStore" Version="1.2.0-beta.8"/>
+        <PackageReference Update="FSharp.Core" Version="5.0.0" />
+        <PackageReference Include="FSharp.Core" Version="5.0.0" />
+        <PackageReference Include="FSharp.Prelude" Version="4.0.3" />
+        <PackageReference Include="Npgsql" Version="5.0.4" />
+        <PackageReference Include="SqlStreamStore" Version="1.2.0-beta.8" />
     </ItemGroup>
 
     <ItemGroup>
-        <Compile Include="SqlStreamStoreExtensions.fs"/>
-        <Compile Include="Connect.fs"/>
-        <Compile Include="Read.fs"/>
-        <Compile Include="Get.fs"/>
-        <Compile Include="Subscribe.fs"/>
-        <Folder Include="SqlStreamStore.FSharp"/>
-        <Compile Include="Append.fs"/>
+        <Compile Include="Extensions.fs" />
+        <Compile Include="NewStreamMessage.fs" />
+        <Compile Include="Connect.fs" />
+        <Compile Include="Read.fs" />
+        <Compile Include="Get.fs" />
+        <Compile Include="Subscribe.fs" />
+        <Folder Include="SqlStreamStore.FSharp" />
+        <Compile Include="Append.fs" />
     </ItemGroup>
 
 </Project>

src/SqlStreamStore.FSharp/SqlStreamStore.FSharp.fsproj

@@ -26,8 +26,9 @@ module Subscribe =
         (subscriptionName: string)
         (continueAfterVersion: int)
         (streamMessageReceived: IStreamSubscription -> StreamMessage -> CancellationToken -> AsyncResult<_, _>)
-        (streamSubOption: StreamSubOption list)
-        : Stream -> IStreamSubscription =
+        (streamSubOptions: StreamSubOption list)
+        (Stream stream: Stream)
+        : IStreamSubscription =
 
         let mutable hasCaughtUp = null
         let mutable maxCountPerRead = 10
@@ -43,28 +44,27 @@ module Subscribe =
 
             StreamMessageReceived subs
 
-        streamSubOption
+        streamSubOptions
         |> List.iter
             (function
             | StreamSubOption.HasCaughtUp f -> hasCaughtUp <- HasCaughtUp f
             | StreamSubOption.MaxCountPerRead n -> maxCountPerRead <- n
             | StreamSubOption.NoPrefetch -> prefetch <- false
             | StreamSubOption.SubscriptionDropped f -> subscriptionDropped <- SubscriptionDropped f)
 
-        fun (Stream stream) ->
-            let sub =
-                stream.store.SubscribeToStream(
-                    streamId = StreamId stream.streamId,
-                    continueAfterVersion = System.Nullable continueAfterVersion,
-                    streamMessageReceived = streamMessageReceived',
-                    subscriptionDropped = subscriptionDropped,
-                    hasCaughtUp = hasCaughtUp,
-                    prefetchJsonData = prefetch,
-                    name = subscriptionName
-                )
+        let sub =
+            stream.store.SubscribeToStream(
+                streamId = StreamId stream.streamId,
+                continueAfterVersion = System.Nullable continueAfterVersion,
+                streamMessageReceived = streamMessageReceived',
+                subscriptionDropped = subscriptionDropped,
+                hasCaughtUp = hasCaughtUp,
+                prefetchJsonData = prefetch,
+                name = subscriptionName
+            )
 
-            sub.MaxCountPerRead <- maxCountPerRead
-            sub
+        sub.MaxCountPerRead <- maxCountPerRead
+        sub
 
     let toStreamMessages
         (subscriptionName: string)
@@ -77,7 +77,7 @@ module Subscribe =
         (subscriptionName: string)
         (continueAfterPosition: int64)
         (streamMessageReceived: IAllStreamSubscription -> StreamMessage -> CancellationToken -> AsyncResult<_, _>)
-        (streamSubOption: AllStreamSubOption list)
+        (streamSubOptions: AllStreamSubOption list)
         : IStreamStore -> IAllStreamSubscription =
 
         let mutable hasCaughtUp = null
@@ -94,7 +94,7 @@ module Subscribe =
 
             AllStreamMessageReceived subs
 
-        streamSubOption
+        streamSubOptions
         |> List.iter
             (function
             | AllStreamSubOption.HasCaughtUp f -> hasCaughtUp <- HasCaughtUp f