wip: BinaryCoStream and the sync issues

e2e/BinaryCoStream/src/lib/generateTestFile.ts

@@ -6,12 +6,18 @@ export async function generateTestFile(me: Account, bytes: number) {
   group.addMember("everyone", "writer");
 
   const ownership = { owner: group };
+
+  const file = await BinaryCoStream.createFromBlob(
+    new Blob(['1'.repeat(bytes)], { type: 'image/png' }),
+    {
+      ...ownership,
+      nonBlocking: true,
+    }
+  )
+
   const testFile = UploadedFile.create(
     {
-      file: await BinaryCoStream.createFromBlob(
-        new Blob(['1'.repeat(bytes)], { type: 'image/png' }),
-        ownership
-      ),
+      file,
       syncCompleted: false,
     },
     ownership

packages/jazz-run/src/startSync.ts

@@ -89,7 +89,9 @@ export const startSync = Command.make(
                         ?.split(",")[0]
                         ?.trim() || req.socket.remoteAddress;
 
-                const clientId = clientAddress + "@" + new Date().toISOString();
+                const url = req.url ? new URL(`http://${process.env.HOST ?? 'localhost'}${req.url}`) : undefined;
+
+                const clientId = clientAddress + "@" + new Date().toISOString() + url?.search;
 
                 localNode.syncManager.addPeer(
                     createWebSocketPeer({

packages/jazz-tools/src/coValues/coStream.ts

@@ -577,6 +577,7 @@ export class BinaryCoStream extends CoValueBase implements CoValue {
         options: {
             owner: Group | Account;
             onProgress?: (progress: number) => void;
+            nonBlocking?: boolean;
         },
     ): Promise<BinaryCoStream> {
         const stream = this.create({ owner: options.owner });
@@ -593,26 +594,34 @@ export class BinaryCoStream extends CoValueBase implements CoValue {
 
         let lastProgressUpdate = Date.now();
 
-        for (let idx = 0; idx < data.length; idx += chunkSize) {
-            stream.push(data.slice(idx, idx + chunkSize));
-
-            if (Date.now() - lastProgressUpdate > 100) {
-                options.onProgress?.(idx / data.length);
-                lastProgressUpdate = Date.now();
+        async function loadChunks() {
+            for (let idx = 0; idx < data.length; idx += chunkSize) {
+                stream.push(data.slice(idx, idx + chunkSize));
+    
+                if (Date.now() - lastProgressUpdate > 100) {
+                    options.onProgress?.(idx / data.length);
+                    lastProgressUpdate = Date.now();
+                }
+    
+                await new Promise((resolve) => setTimeout(resolve, 0));
             }
-
-            await new Promise((resolve) => setTimeout(resolve, 0));
+            stream.end();
+            const end = Date.now();
+    
+            console.debug(
+                "Finished creating binary stream in",
+                (end - start) / 1000,
+                "s - Throughput in MB/s",
+                (1000 * (blob.size / (end - start))) / (1024 * 1024),
+            );
+            options.onProgress?.(1);
         }
-        stream.end();
-        const end = Date.now();
 
-        console.debug(
-            "Finished creating binary stream in",
-            (end - start) / 1000,
-            "s - Throughput in MB/s",
-            (1000 * (blob.size / (end - start))) / (1024 * 1024),
-        );
-        options.onProgress?.(1);
+        if (options.nonBlocking) {
+            void loadChunks();
+        } else {
+            await loadChunks();
+        }
 
         return stream;
     }