First draft of IMessageSystem, wrapRun utility for IMessageSystem imp…

…lementations (#29)

Emulsion.Tests/Emulsion.Tests.fsproj

@@ -5,6 +5,7 @@
   </PropertyGroup>
   <ItemGroup>
     <Compile Include="Settings.fs" />
+    <Compile Include="MessageSystem.fs" />
     <Compile Include="Actors/Core.fs" />
     <Compile Include="Actors/SyncTaskWatcher.fs" />
     <Compile Include="Actors/Telegram.fs" />

Emulsion.Tests/MessageSystem.fs

@@ -0,0 +1,38 @@
+module Emulsion.Tests.MessageSystem
+
+open Xunit
+
+open System
+open System.Threading
+open Emulsion
+
+let private performTest expectedStage runBody =
+    use cts = new CancellationTokenSource()
+    let mutable stage = 0
+    let run ct =
+        stage <- stage + 1
+        runBody cts ct stage
+    MessageSystem.wrapRun cts.Token run ignore
+    Assert.Equal(expectedStage, stage)
+
+[<Fact>]
+let ``wrapRun should restart the activity on error``() =
+    performTest 2 (fun cts _ stage ->
+        match stage with
+        | 1 -> raise <| Exception()
+        | 2 -> cts.Cancel()
+    )
+
+[<Fact>]
+let ``wrapRun should not restart on OperationCanceledException``() =
+    performTest 1 (fun cts ct _ ->
+        cts.Cancel()
+        ct.ThrowIfCancellationRequested()
+    )
+
+[<Fact>]
+let ``wrapRun should not restart on token.Cancel()``() =
+    performTest 4 (fun cts _ stage ->
+        if stage > 3 then
+            cts.Cancel()
+    )

Emulsion/Emulsion.fsproj

@@ -7,6 +7,7 @@
     <Compile Include="AssemblyInfo.fs" />
     <Compile Include="EmulsionSettings.fs" />
     <Compile Include="Message.fs" />
+    <Compile Include="MessageSystem.fs" />
     <Compile Include="Telegram\Html.fs" />
     <Compile Include="Telegram\Funogram.fs" />
     <Compile Include="Telegram\Client.fs" />

Emulsion/MessageSystem.fs

@@ -0,0 +1,24 @@
+module Emulsion.MessageSystem
+
+open System
+open System.Threading
+
+type IncomingMessageReceiver = IncomingMessage -> unit
+
+/// The IM message queue. Manages the underlying connection, reconnects when necessary, stores the outgoing messages in
+/// a queue and sends them when possible. Redirects the incoming messages to a function passed when starting the queue.
+type IMessageSystem =
+    /// Starts the IM connection, manages reconnects. On cancellation could either throw OperationCanceledException or
+    /// return a unit.
+    abstract member Run : IncomingMessageReceiver -> CancellationToken -> unit
+
+    /// Queues the message to be sent to the IM system when possible.
+    abstract member PutMessage : OutgoingMessage -> unit
+
+let internal wrapRun (token: CancellationToken) (run: CancellationToken -> unit) (log: Exception -> unit) : unit =
+    while not token.IsCancellationRequested do
+        try
+            run token
+        with
+        | :? OperationCanceledException -> ()
+        | ex -> log ex