dont need multiple listeners on connection (#304)

## Why

the connection class used to support a set of listeners for each type,
YAGNI! and can conceal potential failures where we are dispatching
events to multiple sessions

## What changed

make listeners `cb | undefined` instead of `Set<cb>`

## Versioning

- [ ] Breaking protocol change
- [x] Breaking ts/js API change

<!-- Kind reminder to add tests and updated documentation if needed -->

Author: jackyzha0

package-lock.json

@@ -1,7 +1,7 @@
 {
   "name": "@replit/river",
   "description": "It's like tRPC but... with JSON Schema Support, duplex streaming and support for service multiplexing. Transport agnostic!",
-  "version": "0.207.3",
+  "version": "0.207.4",
   "type": "module",
   "exports": {
     ".": {

package.json

@@ -18,21 +18,15 @@ export class InMemoryConnection extends Connection {
     this.conn.allowHalfOpen = false;
 
     this.conn.on('data', (data: Uint8Array) => {
-      for (const cb of this.dataListeners) {
-        cb(data);
-      }
+      this.dataListener?.(data);
     });
 
     this.conn.on('close', () => {
-      for (const cb of this.closeListeners) {
-        cb();
-      }
+      this.closeListener?.();
     });
 
     this.conn.on('error', (err) => {
-      for (const cb of this.errorListeners) {
-        cb(err);
-      }
+      this.errorListener?.(err);
     });
   }
 

testUtil/fixtures/mockTransport.ts

@@ -4,9 +4,9 @@ import { generateId } from './id';
 
 /**
  * A connection is the actual raw underlying transport connection.
- * It’s responsible for dispatching to/from the actual connection itself
+ * It's responsible for dispatching to/from the actual connection itself
  * This should be instantiated as soon as the client/server has a connection
- * It’s tied to the lifecycle of the underlying transport connection (i.e. if the WS drops, this connection should be deleted)
+ * It's tied to the lifecycle of the underlying transport connection (i.e. if the WS drops, this connection should be deleted)
  */
 export abstract class Connection {
   id: string;
@@ -30,85 +30,65 @@ export abstract class Connection {
     return metadata;
   }
 
-  // can't use event emitter because we need this to work in both node + browser
-  private _dataListeners = new Set<(msg: Uint8Array) => void>();
-  private _closeListeners = new Set<() => void>();
-  private _errorListeners = new Set<(err: Error) => void>();
-
-  get dataListeners() {
-    return [...this._dataListeners];
-  }
-
-  get closeListeners() {
-    return [...this._closeListeners];
-  }
-
-  get errorListeners() {
-    return [...this._errorListeners];
-  }
+  dataListener?: (msg: Uint8Array) => void;
+  closeListener?: () => void;
+  errorListener?: (err: Error) => void;
 
   onData(msg: Uint8Array) {
-    for (const cb of this.dataListeners) {
-      cb(msg);
-    }
+    this.dataListener?.(msg);
   }
 
   onError(err: Error) {
-    for (const cb of this.errorListeners) {
-      cb(err);
-    }
+    this.errorListener?.(err);
   }
 
   onClose() {
-    for (const cb of this.closeListeners) {
-      cb();
-    }
-
+    this.closeListener?.();
     this.telemetry?.span.end();
   }
 
   /**
-   * Handle adding a callback for when a message is received.
-   * @param msg The message that was received.
+   * Set the callback for when a message is received.
+   * @param cb The message handler callback.
    */
-  addDataListener(cb: (msg: Uint8Array) => void) {
-    this._dataListeners.add(cb);
+  setDataListener(cb: (msg: Uint8Array) => void) {
+    this.dataListener = cb;
   }
 
-  removeDataListener(cb: (msg: Uint8Array) => void): void {
-    this._dataListeners.delete(cb);
+  removeDataListener() {
+    this.dataListener = undefined;
   }
 
   /**
-   * Handle adding a callback for when the connection is closed.
-   * This should also be called if an error happens and after notifying all the error listeners.
+   * Set the callback for when the connection is closed.
+   * This should also be called if an error happens and after notifying the error listener.
    * @param cb The callback to call when the connection is closed.
    */
-  addCloseListener(cb: () => void): void {
-    this._closeListeners.add(cb);
+  setCloseListener(cb: () => void): void {
+    this.closeListener = cb;
   }
 
-  removeCloseListener(cb: () => void): void {
-    this._closeListeners.delete(cb);
+  removeCloseListener(): void {
+    this.closeListener = undefined;
   }
 
   /**
-   * Handle adding a callback for when an error is received.
-   * This should only be used for this.logging errors, all cleanup
-   * should be delegated to addCloseListener.
+   * Set the callback for when an error is received.
+   * This should only be used for logging errors, all cleanup
+   * should be delegated to setCloseListener.
    *
    * The implementer should take care such that the implemented
    * connection will call both the close and error callbacks
    * on an error.
    *
    * @param cb The callback to call when an error is received.
    */
-  addErrorListener(cb: (err: Error) => void): void {
-    this._errorListeners.add(cb);
+  setErrorListener(cb: (err: Error) => void): void {
+    this.errorListener = cb;
   }
 
-  removeErrorListener(cb: (err: Error) => void): void {
-    this._errorListeners.delete(cb);
+  removeErrorListener(): void {
+    this.errorListener = undefined;
   }
 
   /**

transport/connection.ts

@@ -91,9 +91,9 @@ export class SessionConnected<
     this.conn = props.conn;
     this.listeners = props.listeners;
 
-    this.conn.addDataListener(this.onMessageData);
-    this.conn.addCloseListener(this.listeners.onConnectionClosed);
-    this.conn.addErrorListener(this.listeners.onConnectionErrored);
+    this.conn.setDataListener(this.onMessageData);
+    this.conn.setCloseListener(this.listeners.onConnectionClosed);
+    this.conn.setErrorListener(this.listeners.onConnectionErrored);
   }
 
   sendBufferedMessages(): SendBufferResult {
@@ -240,9 +240,9 @@ export class SessionConnected<
 
   _handleStateExit(): void {
     super._handleStateExit();
-    this.conn.removeDataListener(this.onMessageData);
-    this.conn.removeCloseListener(this.listeners.onConnectionClosed);
-    this.conn.removeErrorListener(this.listeners.onConnectionErrored);
+    this.conn.removeDataListener();
+    this.conn.removeCloseListener();
+    this.conn.removeErrorListener();
 
     if (this.heartbeatHandle) {
       clearInterval(this.heartbeatHandle);

transport/sessionStateMachine/SessionConnected.ts

@@ -56,9 +56,9 @@ export class SessionHandshaking<
       this.listeners.onHandshakeTimeout();
     }, this.options.handshakeTimeoutMs);
 
-    this.conn.addDataListener(this.onHandshakeData);
-    this.conn.addErrorListener(this.listeners.onConnectionErrored);
-    this.conn.addCloseListener(this.listeners.onConnectionClosed);
+    this.conn.setDataListener(this.onHandshakeData);
+    this.conn.setErrorListener(this.listeners.onConnectionErrored);
+    this.conn.setCloseListener(this.listeners.onConnectionClosed);
   }
 
   get loggingMetadata() {
@@ -88,9 +88,9 @@ export class SessionHandshaking<
 
   _handleStateExit(): void {
     super._handleStateExit();
-    this.conn.removeDataListener(this.onHandshakeData);
-    this.conn.removeErrorListener(this.listeners.onConnectionErrored);
-    this.conn.removeCloseListener(this.listeners.onConnectionClosed);
+    this.conn.removeDataListener();
+    this.conn.removeErrorListener();
+    this.conn.removeCloseListener();
 
     if (this.handshakeTimeout) {
       clearTimeout(this.handshakeTimeout);

transport/sessionStateMachine/SessionHandshaking.ts

@@ -54,9 +54,9 @@ export class SessionWaitingForHandshake<
       this.listeners.onHandshakeTimeout();
     }, this.options.handshakeTimeoutMs);
 
-    this.conn.addDataListener(this.onHandshakeData);
-    this.conn.addErrorListener(this.listeners.onConnectionErrored);
-    this.conn.addCloseListener(this.listeners.onConnectionClosed);
+    this.conn.setDataListener(this.onHandshakeData);
+    this.conn.setErrorListener(this.listeners.onConnectionErrored);
+    this.conn.setCloseListener(this.listeners.onConnectionClosed);
   }
 
   get loggingMetadata() {
@@ -88,9 +88,9 @@ export class SessionWaitingForHandshake<
   }
 
   _handleStateExit(): void {
-    this.conn.removeDataListener(this.onHandshakeData);
-    this.conn.removeErrorListener(this.listeners.onConnectionErrored);
-    this.conn.removeCloseListener(this.listeners.onConnectionClosed);
+    this.conn.removeDataListener();
+    this.conn.removeErrorListener();
+    this.conn.removeCloseListener();
     clearTimeout(this.handshakeTimeout);
     this.handshakeTimeout = undefined;
   }