devplacepy - Molodetz Institute

 // retoor <retoor@molodetz.nl>
 export class PubSubClient {
     constructor() {
         this.socket = null;
         this.ready = false;
         this.handlers = new Map();
         this.pending = [];
         this.backoff = 200;
         this.maxBackoff = 5000;
     }
     _url() {
         const protocol = location.protocol === "https:" ? "wss:" : "ws:";
         return `${protocol}//${location.host}/pubsub/ws`;
     }
     _connect() {
         if (this.socket) return;
         const socket = new WebSocket(this._url());
         this.socket = socket;
         socket.addEventListener("open", () => {
             this.ready = true;
             this.backoff = 200;
             for (const topic of this.handlers.keys()) {
                 socket.send(JSON.stringify({ type: "subscribe", topic }));
             }
             const queued = this.pending;
             this.pending = [];
             queued.forEach((frame) => socket.send(JSON.stringify(frame)));
         });
         socket.addEventListener("message", (event) => {
             let frame;
             try {
                 frame = JSON.parse(event.data);
             } catch (error) {
                 return;
             }
             if (frame.type === "message") this._dispatch(frame);
         });
         socket.addEventListener("close", (event) => {
             this.ready = false;
             this.socket = null;
             const delay = event.code === 4013 ? 200 : this.backoff;
             this.backoff = Math.min(this.backoff * 2, this.maxBackoff);
             if (this.handlers.size || this.pending.length) {
                 setTimeout(() => this._connect(), delay);
             }
         });
     }
     _matches(pattern, topic) {
         if (pattern === topic || pattern === "*") return true;
         if (pattern.endsWith(".*")) {
             return topic === pattern.slice(0, -2) || topic.startsWith(pattern.slice(0, -1));
         }
         return false;
     }
     _dispatch(frame) {
         for (const [pattern, callbacks] of this.handlers.entries()) {
             if (this._matches(pattern, frame.topic)) {
                 callbacks.forEach((callback) => callback(frame.data, frame.topic));
             }
         }
     }
     _send(frame) {
         if (this.ready && this.socket) {
             this.socket.send(JSON.stringify(frame));
         } else {
             this.pending.push(frame);
             this._connect();
         }
     }
     subscribe(topic, callback) {
         let callbacks = this.handlers.get(topic);
         if (!callbacks) {
             callbacks = new Set();
             this.handlers.set(topic, callbacks);
             this._send({ type: "subscribe", topic });
         }
         callbacks.add(callback);
         this._connect();
         return () => this.unsubscribe(topic, callback);
     }
     unsubscribe(topic, callback) {
         const callbacks = this.handlers.get(topic);
         if (!callbacks) return;
         callbacks.delete(callback);
         if (!callbacks.size) {
             this.handlers.delete(topic);
             this._send({ type: "unsubscribe", topic });
         }
     }
     publish(topic, data) {
         this._send({ type: "publish", topic, data });
     }
 }