Move EventSource to SharedWorker (#12095)

Move EventSource to use a SharedWorker. This prevents issues with HTTP/1.1
open browser connections from preventing gitea from opening multiple tabs.

Also allow setting EVENT_SOURCE_UPDATE_TIME to disable EventSource updating

Fix #11978

Signed-off-by: Andrew Thornton <art27@cantab.net>

Co-authored-by: silverwind <me@silverwind.io>
Co-authored-by: techknowlogick <techknowlogick@gitea.io>

.eslintrc

@@ -28,7 +28,7 @@ globals:
   Tribute: false
 
 overrides:
-  - files: ["web_src/**/*.worker.js", "web_src/js/serviceworker.js"]
+  - files: ["web_src/**/*worker.js"]
     env:
       worker: true
 

custom/conf/app.example.ini

@@ -218,7 +218,7 @@ MIN_TIMEOUT = 10s
 MAX_TIMEOUT = 60s
 TIMEOUT_STEP = 10s
 ; This setting determines how often the db is queried to get the latest notification counts.
-; If the browser client supports EventSource, it will be used in preference to polling notification.
+; If the browser client supports EventSource and SharedWorker, a SharedWorker will be used in preference to polling notification. Set to -1 to disable the EventSource
 EVENT_SOURCE_UPDATE_TIME = 10s
 
 [markdown]

docs/content/doc/advanced/config-cheat-sheet.en-us.md

@@ -150,8 +150,7 @@ Values containing `#` or `;` must be quoted using `` ` `` or `"""`.
 - `MIN_TIMEOUT`: **10s**: These options control how often notification endpoint is polled to update the notification count. On page load the notification count will be checked after `MIN_TIMEOUT`. The timeout will increase to `MAX_TIMEOUT` by `TIMEOUT_STEP` if the notification count is unchanged. Set MIN_TIMEOUT to 0 to turn off.
 - `MAX_TIMEOUT`: **60s**.
 - `TIMEOUT_STEP`: **10s**.
-- `EVENT_SOURCE_UPDATE_TIME`: **10s**: This setting determines how often the database is queried to update notification counts. If the browser client supports `EventSource`, it will be used in preference to polling notification endpoint.
+- `EVENT_SOURCE_UPDATE_TIME`: **10s**: This setting determines how often the database is queried to update notification counts. If the browser client supports `EventSource` and `SharedWorker`, a `SharedWorker` will be used in preference to polling notification endpoint. Set to **-1** to disable the `EventSource`.
-
 
 ## Markdown (`markdown`)
 

modules/eventsource/manager_run.go

@@ -17,6 +17,9 @@ import (
 
 // Init starts this eventsource
 func (m *Manager) Init() {
+	if setting.UI.Notification.EventSourceUpdateTime <= 0 {
+		return
+	}
 	go graceful.GetManager().RunWithShutdownContext(m.Run)
 }
 

modules/templates/helper.go

@@ -289,8 +289,8 @@ func NewFuncMap() []template.FuncMap {
 				return ""
 			}
 		},
-		"NotificationSettings": func() map[string]int {
+		"NotificationSettings": func() map[string]interface{} {
-			return map[string]int{
+			return map[string]interface{}{
 				"MinTimeout":            int(setting.UI.Notification.MinTimeout / time.Millisecond),
 				"TimeoutStep":           int(setting.UI.Notification.TimeoutStep / time.Millisecond),
 				"MaxTimeout":            int(setting.UI.Notification.MaxTimeout / time.Millisecond),

web_src/js/features/eventsource.sharedworker.js

@@ -0,0 +1,140 @@
+self.name = 'eventsource.sharedworker.js';
+
+const sourcesByUrl = {};
+const sourcesByPort = {};
+
+class Source {
+  constructor(url) {
+    this.url = url;
+    this.eventSource = new EventSource(url);
+    this.listening = {};
+    this.clients = [];
+    this.listen('open');
+    this.listen('logout');
+    this.listen('notification-count');
+    this.listen('error');
+  }
+
+  register(port) {
+    if (!this.clients.includes(port)) return;
+
+    this.clients.push(port);
+
+    port.postMessage({
+      type: 'status',
+      message: `registered to ${this.url}`,
+    });
+  }
+
+  deregister(port) {
+    const portIdx = this.clients.indexOf(port);
+    if (portIdx < 0) {
+      return this.clients.length;
+    }
+    this.clients.splice(portIdx, 1);
+    return this.clients.length;
+  }
+
+  close() {
+    if (!this.eventSource) return;
+
+    this.eventSource.close();
+    this.eventSource = null;
+  }
+
+  listen(eventType) {
+    if (this.listening[eventType]) return;
+    this.listening[eventType] = true;
+    const self = this;
+    this.eventSource.addEventListener(eventType, (event) => {
+      self.notifyClients({
+        type: eventType,
+        data: event.data
+      });
+    });
+  }
+
+  notifyClients(event) {
+    for (const client of this.clients) {
+      client.postMessage(event);
+    }
+  }
+
+  status(port) {
+    port.postMessage({
+      type: 'status',
+      message: `url: ${this.url} readyState: ${this.eventSource.readyState}`,
+    });
+  }
+}
+
+self.onconnect = (e) => {
+  for (const port of e.ports) {
+    port.addEventListener('message', (event) => {
+      if (event.data.type === 'start') {
+        const url = event.data.url;
+        if (sourcesByUrl[url]) {
+          // we have a Source registered to this url
+          const source = sourcesByUrl[url];
+          source.register(port);
+          sourcesByPort[port] = source;
+          return;
+        }
+        let source = sourcesByPort[port];
+        if (source) {
+          if (source.eventSource && source.url === url) return;
+
+          // How this has happened I don't understand...
+          // deregister from that source
+          const count = source.deregister(port);
+            // Clean-up
+          if (count === 0) {
+            source.close();
+            sourcesByUrl[source.url] = null;
+          }
+        }
+        // Create a new Source
+        source = new Source(url);
+        source.register(port);
+        sourcesByUrl[url] = source;
+        sourcesByPort[port] = source;
+        return;
+      } else if (event.data.type === 'listen') {
+        const source = sourcesByPort[port];
+        source.listen(event.data.eventType);
+        return;
+      } else if (event.data.type === 'close') {
+        const source = sourcesByPort[port];
+
+        if (!source) return;
+
+        const count = source.deregister(port);
+        if (count === 0) {
+          source.close();
+          sourcesByUrl[source.url] = null;
+          sourcesByPort[port] = null;
+        }
+        return;
+      } else if (event.data.type === 'status') {
+        const source = sourcesByPort[port];
+        if (!source) {
+          port.postMessage({
+            type: 'status',
+            message: 'not connected',
+          });
+          return;
+        }
+        source.status(port);
+        return;
+      } else {
+        // just send it back
+        port.postMessage({
+          type: 'error',
+          message: `received but don't know how to handle: ${event.data}`,
+        });
+        return;
+      }
+    });
+    port.start();
+  }
+};

web_src/js/features/notification.js

@@ -18,45 +18,84 @@ export function initNotificationsTable() {
   });
 }
 
-export function initNotificationCount() {
+async function receiveUpdateCount(event) {
-  const notificationCount = $('.notification_count');
-
-  if (!notificationCount.length) {
-    return;
-  }
-
-  if (NotificationSettings.EventSourceUpdateTime > 0 && !!window.EventSource) {
-    // Try to connect to the event source first
-    const source = new EventSource(`${AppSubUrl}/user/events`);
-    source.addEventListener('notification-count', async (e) => {
   try {
-        const data = JSON.parse(e.data);
+    const data = JSON.parse(event.data);
 
-        const notificationCount = $('.notification_count');
+    const notificationCount = document.querySelector('.notification_count');
-        if (data.Count === 0) {
+    if (data.Count > 0) {
-          notificationCount.addClass('hidden');
+      notificationCount.classList.remove('hidden');
     } else {
-          notificationCount.removeClass('hidden');
+      notificationCount.classList.add('hidden');
     }
 
     notificationCount.text(`${data.Count}`);
     await updateNotificationTable();
   } catch (error) {
-        console.error(error);
+    console.error(error, event);
+  }
+}
+
+export async function initNotificationCount() {
+  const notificationCount = $('.notification_count');
+
+  if (!notificationCount.length) {
+    return;
   }
+
+  if (NotificationSettings.EventSourceUpdateTime > 0 && !!window.EventSource) {
+    // Try to connect to the event source via the shared worker first
+    if (window.SharedWorker) {
+      const worker = new SharedWorker(`${__webpack_public_path__}js/eventsource.sharedworker.js`, 'notification-worker');
+      worker.addEventListener('error', (event) => {
+        console.error(event);
       });
-    source.addEventListener('logout', async (e) => {
+      worker.port.onmessageerror = () => {
+        console.error('Unable to deserialize message');
+      };
+      worker.port.postMessage({
+        type: 'start',
+        url: `${window.location.origin}${AppSubUrl}/user/events`,
+      });
+      worker.port.addEventListener('message', (e) => {
+        if (!e.data || !e.data.type) {
+          console.error(e);
+          return;
+        }
+        if (event.data.type === 'notification-count') {
+          receiveUpdateCount(e.data);
+          return;
+        } else if (event.data.type === 'error') {
+          console.error(e.data);
+          return;
+        } else if (event.data.type === 'logout') {
           if (e.data !== 'here') {
             return;
           }
-      source.close();
+          worker.port.postMessage({
+            type: 'close',
+          });
+          worker.port.close();
           window.location.href = AppSubUrl;
+          return;
+        } else {
+          return;
+        }
       });
+      worker.port.addEventListener('error', (e) => {
+        console.error(e);
+      });
+      worker.port.start();
       window.addEventListener('beforeunload', () => {
-      source.close();
+        worker.port.postMessage({
+          type: 'close',
+        });
+        worker.port.close();
       });
+
       return;
     }
+  }
 
   if (NotificationSettings.MinTimeout <= 0) {
     return;

web_src/js/index.js

@@ -2432,7 +2432,6 @@ $(document).ready(async () => {
   initContextPopups();
   initTableSort();
   initNotificationsTable();
-  initNotificationCount();
 
   // Repo clone url.
   if ($('#repo-clone-url').length > 0) {
@@ -2477,6 +2476,7 @@ $(document).ready(async () => {
     initClipboard(),
     initUserHeatmap(),
     initServiceWorker(),
+    initNotificationCount(),
   ]);
 });
 

webpack.config.js

@@ -52,6 +52,9 @@ module.exports = {
     serviceworker: [
       resolve(__dirname, 'web_src/js/serviceworker.js'),
     ],
+    'eventsource.sharedworker': [
+      resolve(__dirname, 'web_src/js/features/eventsource.sharedworker.js'),
+    ],
     icons: [
       ...glob('node_modules/@primer/octicons/build/svg/**/*.svg'),
       ...glob('assets/svg/*.svg'),