From eda71b02505d4081a2a4feb32b2f68e3679c2a53 Mon Sep 17 00:00:00 2001
From: Thomas Randolph <trandolph@gitlab.com>
Date: Wed, 11 May 2022 20:46:41 -0600
Subject: [PATCH] Simple multi-cast, multi-channel communication foundation
 layer

---
 .../javascripts/lib/utils/communication.js    |  65 ++++++++++
 spec/frontend/lib/utils/communication_spec.js | 116 ++++++++++++++++++
 2 files changed, 181 insertions(+)
 create mode 100644 app/assets/javascripts/lib/utils/communication.js
 create mode 100644 spec/frontend/lib/utils/communication_spec.js

diff --git a/app/assets/javascripts/lib/utils/communication.js b/app/assets/javascripts/lib/utils/communication.js
new file mode 100644
index 00000000000000..ac5716951dece4
--- /dev/null
+++ b/app/assets/javascripts/lib/utils/communication.js
@@ -0,0 +1,65 @@
+import { uuids } from './uuids';
+
+function createSimpleSubject() {
+  var errored = false;
+  var completed = false;
+  var err = null;
+  var subscribers = {};
+
+  return {
+    subscribe: (subscriber) => {
+      const id = uuids()[0];
+      let internalSubscriber;
+
+      if (typeof subscriber === 'function') {
+        internalSubscriber = { next: subscriber };
+      } else {
+        internalSubscriber = subscriber;
+      }
+
+      subscribers[id] = internalSubscriber;
+
+      return () => {
+        delete subscribers[id];
+      };
+    },
+    next: (value) => {
+      if (!completed && !errored) {
+        Object.values(subscribers).forEach((sub) => sub.next?.(value));
+      }
+    },
+    error: (e) => {
+      errored = true;
+      err = e;
+
+      Object.values(subscribers).forEach((sub) => sub.error?.(e));
+    },
+    complete: () => {
+      completed = true;
+
+      Object.values(subscribers).forEach((sub) => sub.complete?.());
+    },
+    get observed() {
+      return Object.values(subscribers).length > 0;
+    },
+    get lastError() {
+      return err;
+    },
+  };
+}
+
+const channels = {};
+
+export function openChannel({ name }) {
+  var channel = channels[name];
+
+  if (!channel) {
+    channel = channels[name] = createSimpleSubject();
+  }
+
+  return channel;
+}
+
+export function flushChannel({ name }) {
+  channels[name] = null;
+}
diff --git a/spec/frontend/lib/utils/communication_spec.js b/spec/frontend/lib/utils/communication_spec.js
new file mode 100644
index 00000000000000..86d4e71c140b28
--- /dev/null
+++ b/spec/frontend/lib/utils/communication_spec.js
@@ -0,0 +1,116 @@
+import { openChannel, flushChannel } from '~/lib/utils/communication';
+
+describe('communication', () => {
+  describe('openChannel', () => {
+    let channel;
+
+    beforeEach(() => {
+      channel = openChannel({ name: 'channel' });
+    });
+
+    afterEach(() => {
+      flushChannel({ name: 'channel' });
+    });
+
+    it("should return a 'nextable' object", () => {
+      expect(channel.next).toBeInstanceOf(Function);
+      expect(channel.error).toBeInstanceOf(Function);
+      expect(channel.complete).toBeInstanceOf(Function);
+    });
+
+    it('should return an observable Subject', () => {
+      expect(channel.subscribe).toBeInstanceOf(Function);
+      expect(channel.lastError).toBeNull();
+      expect(channel.observed).toBe(false);
+    });
+
+    it('should always return the same channel for a given name', () => {
+      expect(channel).toBe(openChannel({ name: 'channel' }));
+    });
+  });
+
+  describe('flushChannel', () => {
+    it('should not return the same channel if the channel is flushed between opens', () => {
+      const channel = openChannel({ name: 'channel' });
+
+      flushChannel({ name: 'channel' });
+
+      expect(channel).not.toBe(openChannel({ name: 'channel' }));
+    });
+  });
+
+  describe('channel', () => {
+    let channel;
+
+    beforeEach(() => {
+      channel = openChannel({ name: 'channel' });
+    });
+
+    afterEach(() => {
+      flushChannel({ name: 'channel' });
+    });
+
+    it('should send events to observers', () => {
+      const events = [];
+
+      channel.subscribe((event) => {
+        events.push(event.message);
+      });
+
+      channel.next({ message: 'sent' });
+
+      expect(events).toEqual(['sent']);
+    });
+
+    it('should pass on a completion to observers', () => {
+      channel.subscribe({
+        complete: () => expect(true).toBe(true), // We just want to test that this runs
+      });
+
+      channel.complete();
+    });
+
+    it('should not send events after completing', () => {
+      const events = [];
+
+      channel.subscribe({ next: (event) => events.push(event) });
+      channel.complete();
+
+      channel.next({ name: 'sent' });
+
+      expect(events).toEqual([]);
+    });
+
+    it('should pass on an error to observers', async () => {
+      let caught;
+
+      channel.subscribe({
+        error: (e) => {
+          caught = e;
+        },
+      });
+
+      channel.error('broke');
+
+      expect(caught).toBe('broke');
+    });
+
+    it('should not send events after erroring', () => {
+      const events = [];
+
+      channel.subscribe({
+        next: (e) => events.push(e),
+        error: () => {
+          /* swallow errors */
+        },
+      });
+
+      channel.next('a');
+      channel.next('b');
+      channel.error('error');
+      channel.next('c');
+
+      expect(events).toStrictEqual(['a', 'b']);
+    });
+  });
+});
-- 
GitLab