client side impl begin

client/src/AdriftClient.ts

@@ -79,7 +79,7 @@ export class AdriftBareClient extends Client {
     const ws = new webSocketImpl("ws:null", protocols);
     // this will error. that's okay
 
-    let send = this.connection.wsconnect(
+    let { send, close } = this.connection.wsconnect(
       remote,
       () => {
         onReadyState(WebSocketFields.OPEN);
@@ -88,8 +88,6 @@ export class AdriftBareClient extends Client {
       () => {
         onReadyState(WebSocketFields.CLOSED);
         ws.dispatchEvent(new Event("close"));
-
-        // what do i do for WebSocketFields.closing?
       },
       (data) => {
         ws.dispatchEvent(
@@ -100,11 +98,16 @@ export class AdriftBareClient extends Client {
       }
     );
 
-
     (ws as any).__defineGetter__("send", () => (data: any) => {
       send(data);
     });
-    (ws as any).__defineSetter__("send", () => { });
+    // uv wraps it and we don't want that
+    // i can probably fix later but this is fine for now -CE
+    (ws as any).__defineSetter__("send", () => {});
+
+    (ws as any).__defineGetter__("close", (code?: number, reason?: string) => {
+      close(code, reason);
+    });
 
     return ws;
   }

client/src/Connection.ts

@@ -9,8 +9,13 @@ import {
 } from "protocol";
 
 export class Connection {
-  callbacks: Record<number, Function> = {};
-  openStreams: Record<number, ReadableStreamDefaultController<any>> = {};
+  requestCallbacks: Record<number, Function> = {};
+  openRequestStreams: Record<number, ReadableStreamDefaultController<any>> = {};
+  openingSockets: Record<number, () => void>;
+  openSockets: Record<
+    number,
+    { onclose: () => void; onmessage: (data: any) => void }
+  >;
 
   counter: number = 0;
 
@@ -18,6 +23,10 @@ export class Connection {
     transport.ondata = this.ondata.bind(this);
   }
 
+  nextSeq() {
+    return ++this.counter;
+  }
+
   ondata(data: ArrayBuffer) {
     let cursor = 0;
     const view = new DataView(data);
@@ -35,7 +44,7 @@ export class Connection {
         const payload = JSON.parse(decoder.decode(data.slice(cursor)));
         const stream = new ReadableStream({
           start: (controller) => {
-            this.openStreams[requestID] = controller;
+            this.openRequestStreams[requestID] = controller;
           },
           pull: (controller) => {
             // not needed
@@ -44,17 +53,19 @@ export class Connection {
             // TODO
           },
         });
-        this.callbacks[requestID]({ payload, body: stream });
+        this.requestCallbacks[requestID]({ payload, body: stream });
+        delete this.requestCallbacks[requestID];
         break;
 
       case S2CRequestTypes.HTTPResponseChunk:
-        this.openStreams[requestID]?.enqueue(
+        this.openRequestStreams[requestID]?.enqueue(
           new Uint8Array(data.slice(cursor))
         );
         break;
 
       case S2CRequestTypes.HTTPResponseEnd:
-        this.openStreams[requestID]?.close();
+        this.openRequestStreams[requestID]?.close();
+        delete this.openRequestStreams[requestID];
         break;
     }
   }
@@ -86,23 +97,82 @@ export class Connection {
     let json = JSON.stringify(data);
 
     return new Promise(async (resolve) => {
-      let id = ++this.counter;
-      this.callbacks[id] = resolve;
-      await this.send(id, new Blob([json]), C2SRequestTypes.HTTPRequest);
+      let seq = this.nextSeq();
+      this.requestCallbacks[seq] = resolve;
+      await this.send(seq, new Blob([json]), C2SRequestTypes.HTTPRequest);
     });
   }
-  // idk the type of data, figure it out ig
-  wsconnect(url: URL, onopen: () => void, onclose: () => void, onmessage: (data: any) => void): (data: any) => void {
 
-    // do the connection shit here
+  wsconnect(
+    url: URL,
+    onopen: () => void,
+    onclose: () => void,
+    onmessage: (data: any) => void
+  ): {
+    send: (data: any) => void;
+    close: (code?: number, reason?: string) => void;
+  } {
+    const payload = JSON.stringify({ url });
+    let seq = this.nextSeq();
+    this.send(
+      seq,
+      new TextEncoder().encode(payload),
+      C2SRequestTypes.WSOpen
+    ).catch((e) => {
+      console.error(e);
+      onclose();
+    });
 
-    onopen();
     // this can't be async, just call onopen when opened
+    this.openingSockets[seq] = onopen;
 
-    return (data) => {
-
-      // send "data" to the server
+    return {
+      send: (data) => {
+        if (!this.openSockets[seq]) {
+          throw new Error("send on closed socket");
+        }
+        const cleanup = (e: any) => {
+          console.error(e);
+          delete this.openSockets[seq];
+        };
+        if (typeof data === "string") {
+          this.send(
+            seq,
+            new TextEncoder().encode(data),
+            C2SRequestTypes.WSSendText
+          ).catch(cleanup);
+          return;
+        }
+        if (data instanceof ArrayBuffer) {
+          this.send(seq, data, C2SRequestTypes.WSSendBinary).catch(cleanup);
+          return;
+        }
+        if (ArrayBuffer.isView(data)) {
+          this.send(
+            seq,
+            data.buffer.slice(
+              data.byteOffset,
+              data.byteOffset + data.byteLength
+            ),
+            C2SRequestTypes.WSSendBinary
+          ).catch(cleanup);
+          return;
+        }
+        console.error({ data });
+        throw new Error("Unexpected type passed to send");
+      },
+      close: (code?: number, reason?: string) => {
+        const payload = JSON.stringify({ code, reason });
+        this.send(
+          seq,
+          new TextEncoder().encode(payload),
+          C2SRequestTypes.WSClose
+        ).catch((e) => {
+          // At this point there is nothing left to clean up
+          console.error(e);
+        });
+        delete this.openSockets[seq];
+      },
     };
-
   }
 }