Add XMPP connection timeout

CHANGELOG.md

@@ -8,6 +8,9 @@ Versioning v2.0.0](https://semver.org/spec/v2.0.0.html).
 ### Changed
 - Additional logging for mailbox processor errors
 
+### Added
+- XMPP: connection timeout support ([#141](https://github.com/codingteam/emulsion/issues/141))
+
 ## [1.7.0] - 2021-06-29
 ### Changed
 - Runtime: upgrade to .NET 5

Emulsion.Tests/SettingsTests.fs

@@ -36,6 +36,7 @@ let private testConfiguration = {
         Room = "room"
         RoomPassword = None
         Nickname = "nickname"
+        ConnectionTimeout = TimeSpan.FromMinutes 5.0
         MessageTimeout = TimeSpan.FromSeconds 30.0
         PingInterval = None
         PingTimeout = TimeSpan.FromSeconds 30.0

Emulsion.Tests/Xmpp/EmulsionXmppTests.fs

@@ -1,6 +1,8 @@
 module Emulsion.Tests.Xmpp.EmulsionXmppTests
 
 open System
+open System.Diagnostics
+open System.Threading
 open System.Threading.Tasks
 
 open JetBrains.Lifetimes
@@ -22,6 +24,7 @@ let private settings = {
     Room = "room@conference.example.org"
     RoomPassword = None
     Nickname = "nickname"
+    ConnectionTimeout = TimeSpan.FromSeconds 30.0
     MessageTimeout = TimeSpan.FromSeconds 30.0
     PingInterval = None
     PingTimeout = defaultPingTimeout
@@ -70,6 +73,43 @@ type RunTests(outputHelper: ITestOutputHelper) =
         |> ignore
         Assert.Equal((settings.Room, settings.Nickname), joinRoomArgs)
 
+    [<Fact>]
+    member _.``EmulsionXmpp cancels the connection after timeout``(): unit =
+        let timeout = TimeSpan.FromSeconds 1.0
+        let settings = {
+            settings with
+                ConnectionTimeout = timeout
+        }
+        let client =
+            XmppClientFactory.create(
+                connect = fun () -> async {
+                    do! Async.Sleep(timeout * 10.0)
+                }
+            )
+        let sw = Stopwatch.StartNew()
+        Assert.Throws<TimeoutException>(fun () -> runClientSynchronously settings logger client ignore)
+        |> ignore
+        Assert.True(sw.Elapsed < timeout * 2.0)
+
+    [<Fact>]
+    member _.``EmulsionXmpp forcibly terminates the connection after timeout * 3``(): unit =
+        let timeout = TimeSpan.FromSeconds 1.0
+        let settings = {
+            settings with
+                ConnectionTimeout = timeout
+        }
+        let client =
+            XmppClientFactory.create(
+                connect = fun () -> async {
+                    Thread.Sleep(timeout * 10.0) // non-cancellable
+                }
+            )
+        let sw = Stopwatch.StartNew()
+        Assert.Throws<OperationCanceledException>(fun () -> runClientSynchronously settings logger client ignore)
+        |> ignore
+        Assert.True(sw.Elapsed > timeout)
+        Assert.True(sw.Elapsed < timeout * 6.0)
+
 type ReceiveMessageTests(outputHelper: ITestOutputHelper) =
     let logger = Logging.xunitLogger outputHelper
 

Emulsion/Emulsion.fsproj

@@ -41,6 +41,6 @@
     <PackageReference Include="Serilog" Version="2.8.0" />
     <PackageReference Include="Serilog.Sinks.Console" Version="3.1.1" />
     <PackageReference Include="Serilog.Sinks.RollingFile" Version="3.3.0" />
-    <PackageReference Include="SharpXMPP" Version="0.2.0" />
+    <PackageReference Include="SharpXMPP" Version="0.3.0" />
   </ItemGroup>
 </Project>

Emulsion/Settings.fs

@@ -11,6 +11,7 @@ type XmppSettings = {
     Room: string
     RoomPassword: string option
     Nickname: string
+    ConnectionTimeout: TimeSpan
     MessageTimeout: TimeSpan
     PingInterval: TimeSpan option
     PingTimeout: TimeSpan
@@ -31,6 +32,7 @@ type EmulsionSettings = {
     Log: LogSettings
 }
 
+let defaultConnectionTimeout = TimeSpan.FromMinutes 5.0
 let defaultMessageTimeout = TimeSpan.FromMinutes 5.0
 let defaultPingTimeout = TimeSpan.FromSeconds 30.0
 
@@ -50,6 +52,7 @@ let read (config : IConfiguration) : EmulsionSettings =
         Room = section.["room"]
         RoomPassword = Option.ofObj section.["roomPassword"]
         Nickname = section.["nickname"]
+        ConnectionTimeout = readTimeSpan defaultConnectionTimeout "connectionTimeout" section
         MessageTimeout = readTimeSpan defaultMessageTimeout  "messageTimeout" section
         PingInterval = readTimeSpanOpt "pingInterval" section
         PingTimeout = readTimeSpan defaultPingTimeout "pingTimeout" section

Emulsion/Xmpp/EmulsionXmpp.fs

@@ -1,6 +1,8 @@
 /// Main business logic for an XMPP part of the Emulsion application.
 module Emulsion.Xmpp.EmulsionXmpp
 
+open System
+
 open JetBrains.Lifetimes
 open Serilog
 open SharpXMPP.XMPP
@@ -36,14 +38,49 @@ let initializeLogging (logger: ILogger) (client: IXmppClient): IXmppClient =
     )
     client
 
+let private withTimeout title (logger: ILogger) workflow (timeout: TimeSpan) = async {
+    logger.Information("Starting \"{Title}\" with timeout {Timeout}.", title, timeout)
+    let! child = Async.StartChild(workflow, int timeout.TotalMilliseconds)
+
+    let! childWaiter = Async.StartChild(async {
+        let! result = child
+        return Some(ValueSome result)
+    })
+
+    let waitTime = timeout * 1.5
+    let timeoutWaiter = async {
+        do! Async.Sleep waitTime
+        return Some ValueNone
+    }
+
+    let! completedInTime = Async.Choice [| childWaiter; timeoutWaiter |]
+    match completedInTime with
+    | Some(ValueSome r) -> return r
+    | _ ->
+        logger.Information(
+            "Task {Title} neither complete nor cancelled in {Timeout}. Entering extended wait mode.",
+            title,
+            waitTime
+        )
+        let! completedInTime = Async.Choice [| childWaiter; timeoutWaiter |]
+        match completedInTime with
+        | Some(ValueSome r) -> return r
+        | _ ->
+            logger.Warning(
+                "Task {Title} neither complete nor cancelled in another {Timeout}. Trying to cancel forcibly by terminating the client.",
+                title,
+                waitTime
+            )
+            return raise <| OperationCanceledException($"Operation \"%s{title}\" forcibly cancelled")
+}
+
 /// Outer async will establish a connection and enter the room, inner async will await for the room session
 /// termination.
 let run (settings: XmppSettings)
         (logger: ILogger)
         (client: IXmppClient)
         (messageReceiver: IncomingMessageReceiver): Async<Async<unit>> = async {
-    logger.Information "Connecting to the server"
-    let! sessionLifetime = connect client
+    let! sessionLifetime = withTimeout "server connection" logger (connect client) settings.ConnectionTimeout
     sessionLifetime.ThrowIfNotAlive()
     logger.Information "Connection succeeded"
 

README.md

@@ -23,6 +23,7 @@ settings, there're defaults:
 {
     "xmpp": {
         "roomPassword": null,
+        "connectionTimeout": "00:05:00",
         "messageTimeout": "00:05:00",
         "pingInterval": null,
         "pingTimeout": "00:00:30"

emulsion.example.json

@@ -5,6 +5,7 @@
         "room": "xxxxx@conference.example.org",
         "roomPassword": "// optional",
         "nickname": "хортолёт",
+        "connectionTimeout": "00:05:00",
         "messageTimeout": "00:05:00",
         "pingInterval": "00:01:00",
         "pingTimeout": "00:00:05"