Merge branch 'master' of https://github.com/MercuryWorkshop/adrift

client/AdriftClient.ts

@@ -0,0 +1,25 @@
+
+import Connection from "./Connection";
+import { BareClient as BareClientCustom, registerRemoteListener, setBareClientImplementation, Client, GetRequestHeadersCallback, MetaCallback, ReadyStateCallback, WebSocketImpl, BareHeaders, BareResponse } from "bare-client-custom";
+
+
+// export class Adrift {
+//   bareclient:AdriftBareClient,
+//   constructor(connection:Connection){
+//
+//   }
+// }
+//
+export class AdriftBareClient extends Client {
+  constructor(private connection: Connection) {
+    super();
+  }
+  async request(method: string, requestHeaders: BareHeaders, body: BodyInit | null, remote: URL, cache: string | undefined, duplex: string | undefined, signal: AbortSignal | undefined): Promise<BareResponse> {
+    let rawResponse = await this.connection.httprequest({ a: "test data" });
+
+    return new Response(JSON.stringify(rawResponse)) as BareResponse;
+  }
+  async connect(remote: URL, protocols: string[], getRequestHeaders: GetRequestHeadersCallback, onMeta: MetaCallback, onReadyState: ReadyStateCallback, webSocketImpl: WebSocketImpl): WebSocket {
+
+  }
+}

client/App.tsx

@@ -1,82 +1,93 @@
-import { openWindow, deleteWindow } from "corium";
+import { Component, Fragment, h, render } from "preact";
-import { h, render, Component, Fragment } from 'preact';
+import { RTCTransport } from "./RTCTransport";
-import { RTCConnection } from "./rtc";
 
-
+const _ = [h, render, Component, Fragment];
-// import { setOffer, setCallback } from "./firebase";
+import "./firebase";
 import { BareClient as BareClientCustom, registerRemoteListener, setBareClientImplementation, Client, GetRequestHeadersCallback, MetaCallback, ReadyStateCallback, WebSocketImpl, BareHeaders, BareResponse } from "bare-client-custom";
 
 import { createBareClient } from "@tomphttp/bare-client";
+import { AdriftBareClient } from "./AdriftClient";
+import Connection from "./Connection";
+
+let rtc = new RTCTransport(
+  console.log,
+  () => {
+    // rtc.dataChannel.send("test message");
 
 
-class AdriftClient extends Client {
+    // let client = new AdriftBareClient;
-  
+    // setBareClientImplementation(client);
-  async request(method: string, requestHeaders: BareHeaders, body: BodyInit | null, remote: URL, cache: string | undefined, duplex: string | undefined, signal: AbortSignal | undefined): Promise<BareResponse> {
+    //
-    return new Response("test") as BareResponse;
+
-  }
+  },
-  async connect(remote: URL, protocols: string[], getRequestHeaders: GetRequestHeadersCallback, onMeta: MetaCallback, onReadyState: ReadyStateCallback, webSocketImpl: WebSocketImpl): WebSocket {
+  console.log,
-    
+  console.log,
-  }
+  console.log
-}
+);
+let connection = new Connection(rtc);
+window["co"] = connection;
+// connection.httprequest({ a: 1, b: 2 });
+
+let bare = new AdriftBareClient(connection);
+setBareClientImplementation(bare);
+registerRemoteListener();
 
 
 export default class App extends Component {
-  rtc = new RTCConnection({
+  rtc = rtc;
-    onmessage: console.log,
-    onopen: () => {
-      this.rtc.dataChannel.send("test message");
-
-      let client = new AdriftClient;
-      setBareClientImplementation(client);
-
-
-    }
-  });
-
-  state = {
 
+  state = {};
 
+  onInput = (e) => {
+    const { value } = e.target;
+    this.setState((prev) => ({ ...prev, answer: value }));
   };
 
-  onInput = e => {
-    const { value } = e.target;
-    this.setState(prev => ({ ...prev, answer: value }));
-  }
-
   render(props, state) {
-
-
-
     // setCallback(this.rtc.answer.bind(this.rtc));
-    return <>
+    return (
-      <div>
+      <>
-      </div>
+        <div></div>
+        <button
+          onClick={async () => {
+            let offer = await this.rtc.createOffer();
+            console.log("offer created", offer);
+            console.log(JSON.stringify(offer));
 
-      <button onClick={async () => {
+            const r = await fetch("http://localhost:3000/connect", {
-        let offer = await this.rtc.createOffer();
+              method: "POST",
-        console.log("offer created", offer);
+              headers: { "Content-Type": "application/json" },
+              body: JSON.stringify(offer),
+            });
+            if (r.status != 200) {
+              throw new Error("connect: " + r.status + " " + r.statusText);
+            }
+            const { answer, candidates } = await r.json();
+            await this.rtc.answer(answer, candidates);
+            alert("connected");
 
-        // setOffer(JSON.stringify(offer));
+            // setOffer(JSON.stringify(offer));
-
-        // this.setState(prev => ({ ...prev, offer: JSON.stringify(offer) }));
-
-
-      }}>create offer</button>
-      <p>
-        offer: <code>{state.offer}</code>
-      </p>
-
-      paste answer: <input type="text" value={state.answer} onInput={this.onInput} />
-
-      <button onClick={async () => {
-        let { answer, candidates } = JSON.parse(state.answer);
-        await this.rtc.answer(answer, candidates);
-        alert("connected");
-      }}>
-        connect
-      </button>
-
-    </>;
 
+            // this.setState(prev => ({ ...prev, offer: JSON.stringify(offer) }));
+          }}
+        >
+          create offer
+        </button>
+        <p>
+          offer: <code>{state.offer}</code>
+        </p>
+        paste answer:{" "}
+        <input type="text" value={state.answer} onInput={this.onInput} />
+        <button
+          onClick={async () => {
+            let { answer, candidates } = JSON.parse(state.answer);
+            await this.rtc.answer(answer, candidates);
+            alert("connected");
+          }}
+        >
+          connect
+        </button>
+      </>
+    );
   }
 }

client/Connection.ts

@@ -0,0 +1,96 @@
+import Transport from "./Transport";
+
+
+type ObjectValues<T> = T[keyof T];
+const C2SRequestTypes = {
+  HTTPRequest: 0,
+  WSOpen: 1,
+  WSClose: 2,
+  WSSendText: 3,
+  WSSendBinary: 4,
+} as const;
+type C2SRequestType = ObjectValues<typeof C2SRequestTypes>
+
+const S2CRequestTypes = {
+  HTTPResponse: 0,
+  WSOpen: 1,
+  WSDataText: 2,
+  WSDataBinary: 3,
+} as const;
+type S2CRequestType = ObjectValues<typeof S2CRequestTypes>
+
+
+
+
+export default class Connection {
+
+
+  callbacks: Record<number, Function> = {};
+
+  counter: number = 0;
+
+  constructor(public transport: Transport) {
+
+    transport.ondata = this.ondata.bind(this);
+  }
+
+  ondata(data: ArrayBuffer) {
+    let cursor = 0;
+    const view = new DataView(data);
+
+    let requestID = view.getUint16(cursor);
+    cursor += 2;
+    let requestType = view.getUint8(cursor) as S2CRequestType;
+    cursor += 1;
+
+    console.log(requestID, requestType);
+
+    switch (requestType) {
+      case S2CRequestTypes.HTTPResponse: {
+        let decoder = new TextDecoder();
+        let text = decoder.decode(data.slice(cursor));
+        console.log(text);
+        let json = JSON.parse(text);
+
+        console.log(requestID);
+
+        this.callbacks[requestID](json);
+        break;
+      }
+    }
+  }
+
+  async send(data: ArrayBuffer | Blob, type: C2SRequestType): Promise<number> {
+
+    let requestID = this.counter++;
+
+    let header = new ArrayBuffer(2 + 1);
+    let view = new DataView(header);
+
+    let cursor = 0;
+
+    view.setUint16(cursor, requestID);
+    cursor += 2;
+    view.setUint8(cursor, type);
+    cursor += 1;
+
+    let buf = await new Blob([header, data]).arrayBuffer();
+
+    this.transport.send(buf);
+    console.log(buf);
+
+    return requestID;
+
+  }
+
+  httprequest(data: object): Promise<object> {
+    let json = JSON.stringify(data);
+
+    return new Promise(async (resolve) => {
+
+      let id = this.counter;
+      this.callbacks[id] = resolve;
+      await this.send(new Blob([json]), C2SRequestTypes.HTTPRequest);
+    });
+  }
+}

client/RTCTransport.ts

@@ -1,3 +1,6 @@
+import Transport from "./Transport";
+import Connection from "./Connection";
+
 const rtcConf = {
   iceServers: [
     {
@@ -23,56 +26,41 @@ interface RTCOptions {
   onmessage?
 }
 
-export class RTCConnection {
+export class RTCTransport extends Transport {
   peer: RTCPeerConnection;
 
   dataChannel: RTCDataChannel;
-
+  constructor(onopen, onclose,
-  constructor(options: RTCOptions) {
+    public onconnectionstatechange: () => void,
-    let {
+    public onsignalingstatechange: () => void,
-
+    public onicegatheringstatechange: () => void,
-      onconnectionstatechange,
+  ) {
-      onsignalingstatechange,
+    super(onopen, onclose);
-      oniceconnectionstatechange,
-      onicegatheringstatechange,
-      onopen,
-      onclose,
-      onmessage
-    } = options;
-
     this.peer = new RTCPeerConnection(rtcConf);
     this.peer.onconnectionstatechange = onconnectionstatechange;
-    //   (event) => {
-    //   console.log('Connection state:', this.peer.connectionState);
-    // };
-    this.peer.onsignalingstatechange = onsignalingstatechange;
-    //   (event) => {
-    //   console.log('Signaling state:', this.peer.signalingState);
-    // };
-    this.peer.oniceconnectionstatechange = oniceconnectionstatechange;
-    //   (event) => {
-    //   console.log('ICE connection state:', this.peer.iceConnectionState);
-    //   if (this.peer.iceConnectionState == "disconnected" || this.peer.iceConnectionState == "failed") {
-    //     console.log("disconnected");
-    //     // ondisconnect();
-    //   }
-    // };
-    this.peer.onicegatheringstatechange = onicegatheringstatechange;
-    //   (event) => {
-    //   console.log('ICE gathering state:', this.peer.iceGatheringState);
-    // };
 
+    this.peer.onsignalingstatechange = onsignalingstatechange;
+
+    this.peer.oniceconnectionstatechange =
+      (event) => {
+        console.log('ICE connection state:', this.peer.iceConnectionState);
+        if (this.peer.iceConnectionState == "disconnected" || this.peer.iceConnectionState == "failed") {
+          console.log("disconnected");
+          onclose();
+        }
+      };
+    this.peer.onicegatheringstatechange = onicegatheringstatechange;
     this.dataChannel = this.peer.createDataChannel('host-server');
     this.dataChannel.onopen = onopen;
-    //   () => {
+
-    //   console.log("READY!!!");
-    //
-    //   dataChannel.send("test data");
-    // };
     this.dataChannel.onclose = onclose;
-    this.dataChannel.onmessage = onmessage;
+    this.dataChannel.onmessage = (event) => {
-
+      this.ondata(event.data)
+    };
+  }
 
+  send(data: ArrayBuffer) {
+    this.dataChannel.send(data);
   }
 
 

client/Transport.ts

@@ -0,0 +1,10 @@
+export default abstract class Transport {
+  public ondata: (data: ArrayBuffer) => void = () => { }
+  constructor(
+    public onopen: () => void,
+    public onclose: () => void
+  ) { }
+
+  abstract send(data: ArrayBuffer): void;
+  abstract send(data: ArrayBuffer): void;
+}

package.json

@@ -16,8 +16,15 @@
     "corium": "file:corium",
     "dotenv": "^16.3.1",
     "esbuild": "0.19.0",
+    "express": "^4.18.2",
+    "express-ws": "^5.0.2",
     "firebase": "^10.1.0",
     "preact": "^10.16.0",
+    "ts-node": "^10.9.1",
     "wrtc": "^0.4.7"
+  },
+  "devDependencies": {
+    "@types/express": "^4.17.17",
+    "@types/express-ws": "^3.0.1"
   }
 }

server/main.ts

@@ -1,16 +1,8 @@
 import dotenv from "dotenv";
+import express from "express";
+import expressWs from "express-ws";
 import * as wrtc from "wrtc";
 
-
-import { app } from "../firebase-config";
-console.log(app);
-import { getDatabase, ref, onValue, set } from "firebase/database";
-
-const db = getDatabase();
-let reff = ref(db, "/peers/demo");
-
-
-
 const configuration = {
   iceServers: [
     {
@@ -20,7 +12,11 @@ const configuration = {
 };
 dotenv.config();
 
-async function connect(offer, candidates) {
+async function connect(
+  offer,
+  candidates,
+  onAnswer: (answer: Record<string, any>) => void
+) {
   const localCandidates: any[] = [];
   let dataChannel;
   const peer = new wrtc.RTCPeerConnection(configuration);
@@ -38,18 +34,19 @@ async function connect(offer, candidates) {
     };
   };
   peer.onconnectionstatechange = () => {
-    console.log('Connection state:', peer.connectionState);
+    console.log("Connection state:", peer.connectionState);
   };
   peer.onsignalingstatechange = () => {
-    console.log('Signaling state:', peer.signalingState);
+    console.log("Signaling state:", peer.signalingState);
   };
   peer.oniceconnectionstatechange = () => {
-    console.log('ICE connection state:', peer.iceConnectionState);
+    console.log("ICE connection state:", peer.iceConnectionState);
   };
   peer.onicegatheringstatechange = () => {
-    console.log('ICE gathering state:', peer.iceGatheringState);
+    console.log("ICE gathering state:", peer.iceGatheringState);
   };
   peer.onicecandidate = (event: any) => {
+    console.log("onicecandidate");
     if (event.candidate) {
       localCandidates.push(event.candidate);
       return;
@@ -58,26 +55,46 @@ async function connect(offer, candidates) {
       answer: peer.localDescription,
       candidates: localCandidates,
     };
-    let string = JSON.stringify(payload)
+    onAnswer(payload);
-
-    set(reff, string);
   };
   await peer.setRemoteDescription(offer);
   let answer = await peer.createAnswer();
   await peer.setLocalDescription(answer);
   for (let candidate of candidates) {
+    if (!candidate.candidate) continue;
+    console.log({ candidate });
     await peer.addIceCandidate(candidate);
   }
 }
-onValue(reff, (snapshot) => {
-  const rawdata = snapshot.val();
 
-  let data = JSON.parse(rawdata);
+const app = express() as unknown as expressWs.Application;
-  console.log(data);
+expressWs(app);
 
+app.use(express.json());
+app.use((req, res, next) => {
+  res.header("Access-Control-Allow-Origin", "*");
+  next();
+});
+
+app.post("/connect", (req, res) => {
+  const data = req.body;
   if (data && data.offer && data.localCandidates) {
     const { offer, localCandidates } = data;
-    connect(offer, localCandidates);
+    let didAnswer = false;
+    connect(offer, localCandidates, (answer) => {
+      if (!didAnswer) {
+        didAnswer = true;
+        res.json(answer);
+      }
+    });
   }
 });
 
+app.ws("/dev-ws", (ws, req) => {
+  console.log("ws connect");
+  ws.on("message", (msg) => {
+    console.log({ msg });
+  });
+});
+
+app.listen(3000, () => console.log("listening"));