properly end connection span

package.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.203.1",
+  "version": "0.203.2",
   "type": "module",
   "exports": {
     ".": {

transport/connection.ts

@@ -47,6 +47,26 @@ export abstract class Connection {
     return [...this._errorListeners];
   }
 
+  onData(msg: Uint8Array) {
+    for (const cb of this.dataListeners) {
+      cb(msg);
+    }
+  }
+
+  onError(err: Error) {
+    for (const cb of this.errorListeners) {
+      cb(err);
+    }
+  }
+
+  onClose() {
+    for (const cb of this.closeListeners) {
+      cb();
+    }
+
+    this.telemetry?.span.end();
+  }
+
   /**
    * Handle adding a callback for when a message is received.
    * @param msg The message that was received.

transport/impls/ws/connection.ts

@@ -40,20 +40,14 @@ export class WebSocketConnection extends Connection {
           `websocket closed with code and reason: ${code} - ${reason}`,
         );
 
-        for (const cb of this.errorListeners) {
-          cb(err);
-        }
+        this.onError(err);
       }
 
-      for (const cb of this.closeListeners) {
-        cb();
-      }
+      this.onClose();
     };
 
     this.ws.onmessage = (msg) => {
-      for (const cb of this.dataListeners) {
-        cb(msg.data as Uint8Array);
-      }
+      this.onData(msg.data as Uint8Array);
     };
   }