feat: finish stream

Author: 0xtsukino

demo/web-to-web-p2p/src/app.tsx

@@ -1,24 +1,16 @@
-import React, {
-  ReactElement,
-  ReactNode,
-  useCallback,
-  useEffect,
-  useState,
-} from 'react';
+import React, { ReactElement, useCallback, useEffect, useState } from 'react';
 import { createRoot } from 'react-dom/client';
 import * as Comlink from 'comlink';
 import {
   Prover as TProver,
   Verifier as TVerifier,
-  Presentation as TPresentation,
   Commit,
-  NotaryServer,
   Transcript,
 } from 'tlsn-js';
-import { PresentationJSON } from 'tlsn-js/build/types';
 import './app.scss';
+import WebscoketStream from './stream';
 
-const { init, Prover, Presentation, Verifier }: any = Comlink.wrap(
+const { init, Prover, Verifier }: any = Comlink.wrap(
   new Worker(new URL('./worker.ts', import.meta.url)),
 );
 
@@ -43,6 +35,37 @@ function App(): ReactElement {
     })();
   }, []);
 
+  useEffect(() => {
+    (async () => {
+      // Set up stream for prover
+      (async () => {
+        const proverStream = new WebscoketStream(
+          'ws://localhost:3001?id=prover',
+        );
+        const reader = await proverStream.reader();
+        while (true) {
+          const { done, value } = await reader.read();
+          if (done) {
+            console.log('stream finished');
+            break;
+          }
+          console.log(`Received data from stream:`, await value.text());
+        }
+      })();
+
+      // Set up stream for verifier
+      (async () => {
+        const verifierStream = new WebscoketStream(
+          'ws://localhost:3001?id=verifier',
+        );
+        const writer = await verifierStream.writer();
+        writer.write('Hello');
+        writer.write('World!');
+        writer.close();
+      })();
+    })();
+  }, []);
+
   const addProverLog = useCallback((log: string) => {
     pLogs = pLogs.concat(`${new Date().toLocaleTimeString()} - ${log}`);
     setProverMessages(pLogs);
@@ -102,7 +125,7 @@ function App(): ReactElement {
     });
     addProverLog('request sent');
     const transcript = await prover.transcript();
-    console.log(transcript);
+
     addProverLog('response received');
     addProverLog('transcript.sent');
     addProverLog(transcript.sent);

demo/web-to-web-p2p/src/stream.ts

@@ -0,0 +1,62 @@
+export default class WebscoketStream {
+  client: WebSocket;
+  readable: Promise<ReadableStream>;
+  writable: Promise<WritableStream>;
+  constructor(url: string) {
+    const client = new WebSocket(url);
+
+    const deferredReadable = defer<ReadableStream>();
+    const deferredWritable = defer<WritableStream>();
+    this.client = client;
+    this.readable = deferredReadable.promise;
+    this.writable = deferredWritable.promise;
+
+    client.onopen = () => {
+      const readable = new ReadableStream({
+        start(controller) {
+          client.onmessage = async (event) => {
+            controller.enqueue(event.data);
+          };
+        },
+        cancel() {
+          client.close();
+        },
+      });
+
+      const writable = new WritableStream({
+        write(chunk) {
+          client.send(chunk);
+        },
+        close() {
+          client.close();
+        },
+      });
+
+      deferredReadable.resolve(readable);
+      deferredWritable.resolve(writable);
+    };
+  }
+
+  async reader() {
+    return this.readable.then((stream) => stream.getReader());
+  }
+
+  async writer() {
+    return this.writable.then((stream) => stream.getWriter());
+  }
+}
+
+function defer<value = any>(): {
+  promise: Promise<value>;
+  resolve: (val: value | Promise<value>) => void;
+  reject: (err: any) => void;
+} {
+  let resolve: (val: value | Promise<value>) => void,
+    reject: (err: any) => void;
+  const promise: Promise<value> = new Promise((res, rej) => {
+    resolve = res;
+    reject = rej;
+  });
+
+  return { promise, resolve: resolve!, reject: reject! };
+}