Auto-reconnect to event stream after 20secs timeout

server/events/sse.go

@@ -17,7 +17,7 @@ type Broker interface {
 
 type broker struct {
 	// Events are pushed to this channel by the main events-gathering routine
-	Notifier chan []byte
+	notifier chan []byte
 
 	// New client connections
 	newClients chan chan []byte
@@ -32,7 +32,7 @@ type broker struct {
 func NewBroker() Broker {
 	// Instantiate a broker
 	broker := &broker{
-		Notifier:       make(chan []byte, 1),
+		notifier:       make(chan []byte, 1),
 		newClients:     make(chan chan []byte),
 		closingClients: make(chan chan []byte),
 		clients:        make(map[chan []byte]bool),
@@ -52,7 +52,7 @@ func (broker *broker) SendMessage(event Event) {
 	pkg.Name = event.EventName()
 	pkg.Event = event
 	data, _ := json.Marshal(pkg)
-	broker.Notifier <- data
+	broker.notifier <- data
 }
 
 func (broker *broker) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
@@ -119,7 +119,7 @@ func (broker *broker) listen() {
 			// stop sending them messages.
 			delete(broker.clients, s)
 			log.Debug("Removed client", "numClients", len(broker.clients))
-		case event := <-broker.Notifier:
+		case event := <-broker.notifier:
 
 			// We got a new event from the outside!
 			// Send event to all connected clients

ui/src/App.js

@@ -27,7 +27,7 @@ import createAdminStore from './store/createAdminStore'
 import { i18nProvider } from './i18n'
 import config from './config'
 import { startEventStream } from './eventStream'
-import { updateScanStatus } from './actions'
+import { processEvent } from './actions'
 const history = createHashHistory()
 
 if (config.gaTrackingId) {
@@ -59,7 +59,7 @@ const App = () => (
 
 const Admin = (props) => {
   const dispatch = useDispatch()
-  startEventStream((data) => dispatch(updateScanStatus(data)))
+  startEventStream((data) => dispatch(processEvent(data)))
 
   return (
     <RAAdmin

ui/src/actions/activity.js

@@ -1,12 +0,0 @@
-export const ACTIVITY_SCAN_STATUS_UPD = 'ACTIVITY_SCAN_STATUS_UPD'
-
-const actionsMap = { scanStatus: ACTIVITY_SCAN_STATUS_UPD }
-
-export const updateScanStatus = (data) => {
-  let type = actionsMap[data.name]
-  if (!type) type = 'UNKNOWN'
-  return {
-    type,
-    data: data.data,
-  }
-}

ui/src/actions/index.js

@@ -2,4 +2,4 @@ export * from './audioplayer'
 export * from './themes'
 export * from './albumView'
 export * from './dialogs'
-export * from './activity'
+export * from './serverEvents'

ui/src/actions/serverEvents.js

@@ -0,0 +1,12 @@
+export const EVENT_SCAN_STATUS = 'ACTIVITY_SCAN_STATUS_UPD'
+
+const actionsMap = { scanStatus: EVENT_SCAN_STATUS }
+
+export const processEvent = (data) => {
+  let type = actionsMap[data.name]
+  if (!type) type = 'EVENT_UNKNOWN'
+  return {
+    type,
+    data: data.data,
+  }
+}

ui/src/eventStream.js

@@ -1,9 +1,9 @@
 import baseUrl from './utils/baseUrl'
 import throttle from 'lodash.throttle'
 
-// TODO https://stackoverflow.com/a/20060461
 let es = null
-let dispatchFunc = null
+let onMessageHandler = null
+let timeOut = null
 
 const getEventStream = () => {
   if (es === null) {
@@ -14,17 +14,31 @@ const getEventStream = () => {
   return es
 }
 
-export const startEventStream = (func) => {
+// Reestablish the event stream after 20 secs of inactivity
+const setTimeout = () => {
+  if (timeOut != null) {
+    window.clearTimeout(timeOut)
+  }
+  timeOut = window.setTimeout(() => {
+    es.close()
+    es = null
+    startEventStream(onMessageHandler)
+  }, 20000)
+}
+
+export const startEventStream = (messageHandler) => {
   const es = getEventStream()
-  dispatchFunc = func
+  onMessageHandler = messageHandler
   es.onmessage = throttle(
     (msg) => {
       const data = JSON.parse(msg.data)
       if (data.name !== 'keepAlive') {
-        dispatchFunc(data)
+        onMessageHandler(data)
       }
+      setTimeout() // Reset timeout on every received message
     },
     100,
     { trailing: true }
   )
+  setTimeout()
 }

ui/src/reducers/activityReducer.js

@@ -1,4 +1,4 @@
-import { ACTIVITY_SCAN_STATUS_UPD } from '../actions'
+import { EVENT_SCAN_STATUS } from '../actions'
 
 export const activityReducer = (
   previousState = {
@@ -8,7 +8,7 @@ export const activityReducer = (
 ) => {
   const { type, data } = payload
   switch (type) {
-    case ACTIVITY_SCAN_STATUS_UPD:
+    case EVENT_SCAN_STATUS:
       return { ...previousState, scanStatus: data }
     default:
       return previousState