US 3463 - Implement heartbeat protocol for taiga-events

app-loader/app-loader.coffee

@@ -2,6 +2,8 @@ version = ___VERSION___
 window.taigaConfig = {
     "api": "http://localhost:8000/api/v1/",
     "eventsUrl": null,
+    "eventsMaxMissedHeartbeats": 5,
+    "eventsHeartbeatIntervalTime": 60000,
     "debug": true,
     "defaultLanguage": "en",
     "themes": ["taiga", "material-design", "high-contrast"],

app/coffee/modules/events.coffee

@@ -37,6 +37,9 @@ class EventsService
         @.error = false
         @.pendingMessages = []
 
+        @.missedHeartbeats = 0
+        @.heartbeatInterval = null
+
         if @win.WebSocket is undefined
             @log.info "WebSockets not supported on your browser"
 
@@ -70,10 +73,53 @@ class EventsService
         @.ws.removeEventListener("close", @.onClose)
         @.ws.removeEventListener("error", @.onError)
         @.ws.removeEventListener("message", @.onMessage)
+        @.stopHeartBeatMessages()
         @.ws.close()
 
         delete @.ws
 
+    ###########################################
+    # Heartbeat (Ping - Pong)
+    ###########################################
+    # See  RFC https://tools.ietf.org/html/rfc6455#section-5.5.2
+    #      RFC https://tools.ietf.org/html/rfc6455#section-5.5.3
+    startHeartBeatMessages: ->
+        return if @.heartbeatInterval
+
+        maxMissedHeartbeats =  @config.get("eventsMaxMissedHeartbeats", 5)
+        heartbeatIntervalTime = @config.get("eventsHeartbeatIntervalTime", 60000)
+
+        @.missedHeartbeats = 0
+        @.heartbeatInterval = setInterval(() =>
+            try
+                if @.missedHeartbeats >= maxMissedHeartbeats
+                    throw new Error("Too many missed heartbeats PINGs.")
+
+                @.missedHeartbeats++
+                @.sendMessage({cmd: "ping"})
+                @log.debug("HeartBeat send PING")
+            catch e
+                @log.error("HeartBeat error: " + e.message)
+                @.stopHeartBeatMessages()
+        , heartbeatIntervalTime)
+
+        @log.debug("HeartBeat enabled")
+
+    stopHeartBeatMessages: ->
+        return if not @.heartbeatInterval
+
+        clearInterval(@.heartbeatInterval)
+        @.heartbeatInterval = null
+
+        @log.debug("HeartBeat disabled")
+
+    processHeartBeatPongMessage: (data) ->
+        @.missedHeartbeats = 0
+        @log.debug("HeartBeat recived PONG")
+
+    ###########################################
+    # Messages
+    ###########################################
     serialize: (message) ->
         if _.isObject(message)
             return JSON.stringify(message)
@@ -91,6 +137,19 @@ class EventsService
         for msg in messages
             @.ws.send(msg)
 
+    processMesage: (data) =>
+        routingKey = data.routing_key
+
+        if not @.subscriptions[routingKey]?
+            return
+
+        subscription = @.subscriptions[routingKey]
+        subscription.scope.$apply ->
+            subscription.callback(data.data)
+
+    ###########################################
+    # Subscribe and Unsubscribe
+    ###########################################
     subscribe: (scope, routingKey, callback) ->
         if @.error
             return
@@ -124,8 +183,12 @@ class EventsService
 
         @.sendMessage(message)
 
+    ###########################################
+    # Event listeners
+    ###########################################
     onOpen: ->
         @.connected = true
+        @.startHeartBeatMessages()
 
         @log.debug("WebSocket connection opened")
         token = @auth.getToken()
@@ -141,14 +204,10 @@ class EventsService
         @.log.debug "WebSocket message received: #{event.data}"
 
         data = JSON.parse(event.data)
-        routingKey = data.routing_key
+        if data.cmd = "pong"
-
+            @.processHeartBeatPongMessage(data)
-        if not @.subscriptions[routingKey]?
+        else
-            return
+            @.processMessage(data)
-
-        subscription = @.subscriptions[routingKey]
-        subscription.scope.$apply ->
-            subscription.callback(data.data)
 
     onError: (error) ->
         @log.error("WebSocket error: #{error}")
@@ -157,6 +216,7 @@ class EventsService
     onClose: ->
         @log.debug("WebSocket closed.")
         @.connected = false
+        @.stopHeartBeatMessages()
 
 
 class EventsProvider

conf/conf.example.json

@@ -1,6 +1,8 @@
 {
     "api": "http://localhost:8000/api/v1/",
     "eventsUrl": null,
+    "eventsMaxMissedHeartbeats": 5,
+    "eventsHeartbeatIntervalTime": 60000,
     "debug": true,
     "debugInfo": false,
     "defaultLanguage": "en",