improve presence listener interface

packages/collab-client/src/index.ts

@@ -27,7 +27,6 @@ export interface CollabClientConfig {
 
 export class CollabClient {
   private sending: null | string = null;
-  private controller = new AbortController();
 
   private sendCommit: CollabClientConfig["sendCommit"];
   private listener: CollabClientConfig["listener"];
@@ -64,12 +63,9 @@ export class CollabClient {
   }
 
   async listen(editorState: EditorState, signal?: AbortSignal) {
-    const getCommitsSignal = AbortSignal.any([...(signal ? [signal] : []), this.controller.signal]);
-
-    for await (const newCommits of this.listener.listen(editorState, {
-      signal: getCommitsSignal,
-    })) {
-      if (getCommitsSignal.aborted) break;
+    let options = signal ? { signal } : {};
+    for await (const newCommits of this.listener.listen(editorState, options)) {
+      if (signal && signal.aborted) break;
       this.receiveCommits(newCommits);
     }
   }

packages/comments-client/src/config.ts

@@ -3,10 +3,6 @@ import { PresenceIndicator } from "./PresenceIndicator";
 export interface PresenceClientConfig {
   userId: string;
   sendIndicator: (clientId: string, indicator: PresenceIndicator) => Promise<void>;
-  getIndicators: (
-    clientId: string,
-    refs?: Record<string, string>,
-    options?: { signal?: AbortSignal },
-  ) => Promise<Record<string, PresenceIndicator>>;
+  listener: PresenceListener;
   receiveIndicators: (indicators: Record<string, PresenceIndicator>) => void;
 }

packages/demo/src/editor/Editor.tsx

@@ -8,7 +8,7 @@ import { EditorState, Transaction } from "prosemirror-state";
 import { useCallback, useEffect, useMemo, useState } from "react";
 
 import {
-  LongPollListener,
+  LongPollListener as CollabLongPollListner,
   CollabClient,
   CollabClientConfig,
   receiveCommitTransaction,
@@ -18,7 +18,7 @@ import {
   presence,
   PresenceClient,
   receivePresenceTransaction,
-  LongPollListener as PresenceListener,
+  LongPollListener as PresenceLongPollListener,
   PresenceClientConfig,
 } from "@pitter-patter/presence-client";
 import {
@@ -62,7 +62,7 @@ export function Editor({ doc }: Props) {
 
   const [listener] = useState(
     () =>
-      new LongPollListener(
+      new CollabLongPollListner(
         new URL(
           `/api/docs/${doc.id}/commits`,
           typeof window !== "undefined" ? window.location.href : "http://localhost:3000",
@@ -91,7 +91,7 @@ export function Editor({ doc }: Props) {
 
   const [presenceListener] = useState(
     () =>
-      new PresenceListener(
+      new PresenceLongPollListener(
         new URL(
           `/api/docs/${doc.id}/presence`,
           typeof window !== "undefined" ? window.location.href : "http://localhost:3000",
@@ -109,10 +109,10 @@ export function Editor({ doc }: Props) {
           body: JSON.stringify(indicator),
         });
       },
-      getIndicators: presenceListener.getIndicators.bind(presenceListener),
       receiveIndicators: (indicators) => {
         setState((prev) => prev.apply(receivePresenceTransaction(prev, indicators)));
       },
+      listener: presenceListener,
     }),
     [presenceListener, doc.id],
   );

packages/presence-client/src/config.ts

@@ -1,12 +1,10 @@
 import { type PresenceIndicator } from "./PresenceIndicator";
 
+import { IndicatorListener } from ".";
+
 export interface PresenceClientConfig {
   userId: string;
   sendIndicator: (clientId: string, indicator: PresenceIndicator) => Promise<void>;
-  getIndicators: (
-    clientId: string,
-    refs?: Record<string, string>,
-    options?: { signal?: AbortSignal },
-  ) => Promise<Record<string, PresenceIndicator>>;
   receiveIndicators: (indicators: Record<string, PresenceIndicator>) => void;
+  listener: IndicatorListener;
 }

packages/presence-client/src/index.ts

@@ -10,12 +10,18 @@ export { presence, presenceKey, receivePresenceTransaction } from "./plugin";
 
 export { type PresenceIndicator, type PresenceClientConfig };
 
+export interface IndicatorListener {
+  listen: (
+    clientId: string,
+    options?: { signal?: AbortSignal },
+  ) => AsyncIterableIterator<Record<string, PresenceIndicator>>;
+}
+
 export class PresenceClient {
   private userId: string;
   private clientId: string;
-  private refs: Record<string, string> = {};
   private sendIndicator: PresenceClientConfig["sendIndicator"];
-  private getIndicators: PresenceClientConfig["getIndicators"];
+  private listener: PresenceClientConfig["listener"];
   private receiveIndicators: PresenceClientConfig["receiveIndicators"];
 
   private lastSent: PresenceIndicator | null = null;
@@ -24,7 +30,7 @@ export class PresenceClient {
     this.clientId = randomRef();
     this.userId = config.userId;
     this.sendIndicator = config.sendIndicator;
-    this.getIndicators = config.getIndicators;
+    this.listener = config.listener;
     this.receiveIndicators = config.receiveIndicators;
   }
 
@@ -65,30 +71,16 @@ export class PresenceClient {
     } catch {}
   }
 
-  async listen(signal: AbortSignal) {
-    while (!signal.aborted) {
-      try {
-        const indicators = await this.getIndicators(this.clientId, this.refs, {
-          signal,
-        });
-
-        const newRefs = Object.fromEntries(
-          Object.entries(indicators).map(([clientId, indicator]) => [clientId, indicator.ref]),
-        );
-
-        if (Object.entries(newRefs).every(([clientId, ref]) => this.refs[clientId] === ref)) {
-          continue;
-        }
-
-        this.refs = newRefs;
+  update(config: Partial<Omit<PresenceClientConfig, "listener">>) {
+    if (config.sendIndicator) this.sendIndicator = config.sendIndicator;
+    if (config.receiveIndicators) this.receiveIndicators = config.receiveIndicators;
+  }
 
-        this.receiveIndicators(indicators);
-      } catch (e) {
-        console.error(e);
-        await new Promise<void>((resolve) => {
-          setTimeout(() => resolve(), 3_000);
-        });
-      }
+  async listen(signal?: AbortSignal) {
+    const options = signal ? { signal } : {};
+    for await (const indicators of this.listener.listen(this.clientId, options)) {
+      if (signal && signal.aborted) break;
+      this.receiveIndicators(indicators);
     }
   }
 }
@@ -109,24 +101,55 @@ export class LongPollListener {
   ) {
     this.headers = options.headers ?? {};
     this.fetch = options.fetch ?? globalThis.fetch.bind(globalThis);
+  }
 
-    this.getIndicators = this.getIndicators.bind(this);
+  update(headers: Record<string, string>) {
+    this.headers = headers;
   }
 
-  async getIndicators(clientId: string, refs?: Record<string, string>) {
-    const response = await this.fetch(this.url, {
-      headers: { ...this.headers, "Content-Type": "application/json" },
-      method: "POST",
-      body: JSON.stringify({
-        clientId,
-        refs,
-      }),
-    });
-
-    if (!response.ok) {
-      throw new Error(`Failed to get commits. ${response.status}: ${response.statusText}`);
-    }
+  async *listen(clientId: string, options: { signal?: AbortSignal } = {}) {
+    let refs: Record<string, string> = {};
+
+    while (!options?.signal || !options.signal.aborted) {
+      try {
+        const response = await this.fetch(this.url, {
+          headers: { ...this.headers, "Content-Type": "application/json" },
+          method: "POST",
+          body: JSON.stringify({
+            clientId,
+            refs,
+          }),
+        });
+
+        if (!response.ok) {
+          throw new Error(
+            `Failed to get presence indicators. ${response.status}: ${response.statusText}`,
+          );
+        }
+
+        const indicators = (await response.json()) as Record<string, PresenceIndicator>;
+
+        const newRefs = Object.fromEntries(
+          Object.entries(indicators).map(([clientId, indicator]) => [clientId, indicator.ref]),
+        );
+
+        if (Object.entries(newRefs).every(([clientId, ref]) => refs[clientId] === ref)) {
+          continue;
+        }
+
+        refs = newRefs;
+
+        yield indicators;
+      } catch (e) {
+        console.error(e);
+
+        if (options.signal?.aborted) return;
 
-    return (await response.json()) as Record<string, PresenceIndicator>;
+        // TODO: Implement a backoff strategy
+        await new Promise<void>((resolve) => {
+          setTimeout(() => resolve(), 3_000);
+        });
+      }
+    }
   }
 }