Add basic chunk-by-chunk payload transformation

earthstar-project · May 1, 2024 · fc29124 · fc29124
1 parent f62e9e0
commit fc29124
Show file tree

Hide file tree

Showing 5 changed files with 54 additions and 4 deletions.
diff --git a/.nova/Configuration.json b/.nova/Configuration.json
@@ -1,5 +1,4 @@
 {
-  "co.gwil.deno.config.documentPreloadLimit" : 2000,
   "co.gwil.deno.config.enabledPaths" : [
     "\/Users\/gwil\/Projects\/willow-js\/src",
     "\/Users\/gwil\/Projects\/willow-js\/scripts",
@@ -10,5 +9,8 @@
   "deno.disablePaths" : [
     "dist"
   ],
+  "deno.enablePaths" : [
+    "src"
+  ],
   "deno.unstable" : true
 }
diff --git a/src/wgps/data/data_sender.ts b/src/wgps/data/data_sender.ts
@@ -54,6 +54,7 @@ export class DataSender<
         SubspaceId,
         PayloadDigest
       >;
+      transformPayload: (chunk: Uint8Array) => Uint8Array;
     },
   ) {
   }
@@ -138,10 +139,12 @@ export class DataSender<
       const payloadIterator = await pack.payload.stream(pack.offset);
 
       for await (const chunk of payloadIterator) {
+        const transformed = this.opts.transformPayload(chunk);
+
         yield {
           kind: MsgKind.DataSendPayload,
-          amount: BigInt(chunk.byteLength),
-          bytes: chunk,
+          amount: BigInt(transformed.byteLength),
+          bytes: transformed,
         };
       }
 

diff --git a/src/wgps/data/payload_ingester.ts b/src/wgps/data/payload_ingester.ts
@@ -20,6 +20,7 @@ export class PayloadIngester<
       entry: Entry<NamespaceId, SubspaceId, PayloadDigest>;
     } | typeof CANCELLATION
   >();
+  private processReceivedPayload: (bytes: Uint8Array) => Uint8Array;
 
   constructor(opts: {
     getStore: GetStoreFn<
@@ -31,7 +32,10 @@ export class PayloadIngester<
       SubspaceId,
       PayloadDigest
     >;
+    processReceivedPayload: (bytes: Uint8Array) => Uint8Array;
   }) {
+    this.processReceivedPayload = opts.processReceivedPayload;
+
     onAsyncIterate(this.events, async (event) => {
       if (event === CANCELLATION) {
         this.currentIngestion.push(CANCELLATION);
@@ -65,7 +69,9 @@ export class PayloadIngester<
   }
 
   push(bytes: Uint8Array, end: boolean) {
-    this.events.push(bytes);
+    const processed = this.processReceivedPayload(bytes);
+
+    this.events.push(processed);
 
     if (end) {
       this.events.push(CANCELLATION);

diff --git a/src/wgps/wgps_messenger.test.ts b/src/wgps/wgps_messenger.test.ts
@@ -225,6 +225,8 @@ function testWgpsMessenger(scenario: WgpsScenario) {
             maxSize: BigInt(0),
           }],
         ]]),
+        transformPayload: (bytes) => bytes,
+        processReceivedPayload: (bytes) => bytes,
       });
 
       const storeMapBetty = new StoreMap(
@@ -319,6 +321,8 @@ function testWgpsMessenger(scenario: WgpsScenario) {
             maxSize: BigInt(0),
           }],
         ]]),
+        transformPayload: (bytes) => bytes,
+        processReceivedPayload: (bytes) => bytes,
       });
 
       await delay(20 * scenario.timeMultiplier);
@@ -543,6 +547,8 @@ function testWgpsMessenger(scenario: WgpsScenario) {
             maxSize: BigInt(0),
           }],
         ]]),
+        transformPayload: (bytes) => bytes,
+        processReceivedPayload: (bytes) => bytes,
       });
 
       const storeMapBetty = new StoreMap(
@@ -632,6 +638,8 @@ function testWgpsMessenger(scenario: WgpsScenario) {
             maxSize: BigInt(0),
           }],
         ]]),
+        transformPayload: (bytes) => bytes,
+        processReceivedPayload: (bytes) => bytes,
       });
 
       await delay(20 * scenario.timeMultiplier);
@@ -818,6 +826,8 @@ function testWgpsMessenger(scenario: WgpsScenario) {
             maxSize: BigInt(0),
           }],
         ]]),
+        transformPayload: (bytes) => bytes,
+        processReceivedPayload: (bytes) => bytes,
       });
 
       const storeMapBetty = new StoreMap(
@@ -909,6 +919,8 @@ function testWgpsMessenger(scenario: WgpsScenario) {
             maxSize: BigInt(0),
           }],
         ]]),
+        transformPayload: (bytes) => bytes,
+        processReceivedPayload: (bytes) => bytes,
       });
 
       await delay(20 * scenario.timeMultiplier);
@@ -1091,6 +1103,8 @@ function testWgpsMessenger(scenario: WgpsScenario) {
             maxSize: BigInt(0),
           }],
         ]]),
+        transformPayload: (bytes) => bytes,
+        processReceivedPayload: (bytes) => bytes,
       });
 
       const storeMapBetty = new StoreMap(
@@ -1181,6 +1195,8 @@ function testWgpsMessenger(scenario: WgpsScenario) {
             maxSize: BigInt(0),
           }],
         ]]),
+        transformPayload: (bytes) => bytes,
+        processReceivedPayload: (bytes) => bytes,
       });
 
       await delay(20 * scenario.timeMultiplier);
@@ -1355,6 +1371,8 @@ function testWgpsMessenger(scenario: WgpsScenario) {
               maxSize: BigInt(0),
             }],
           ]]),
+          transformPayload: (bytes) => bytes,
+          processReceivedPayload: (bytes) => bytes,
         });
 
         const storeMapBetty = new StoreMap(
@@ -1454,6 +1472,8 @@ function testWgpsMessenger(scenario: WgpsScenario) {
               maxSize: BigInt(0),
             }],
           ]]),
+          transformPayload: (bytes) => bytes,
+          processReceivedPayload: (bytes) => bytes,
         });
 
         await delay(20 * scenario.timeMultiplier);
@@ -1641,6 +1661,8 @@ function testWgpsMessenger(scenario: WgpsScenario) {
               maxSize: BigInt(0),
             }],
           ]]),
+          transformPayload: (bytes) => bytes,
+          processReceivedPayload: (bytes) => bytes,
         });
 
         const storeMapBetty = new StoreMap(
@@ -1732,6 +1754,8 @@ function testWgpsMessenger(scenario: WgpsScenario) {
               maxSize: BigInt(0),
             }],
           ]]),
+          transformPayload: (bytes) => bytes,
+          processReceivedPayload: (bytes) => bytes,
         });
 
         await delay(20 * scenario.timeMultiplier);
@@ -1906,6 +1930,8 @@ function testWgpsMessenger(scenario: WgpsScenario) {
               maxSize: BigInt(0),
             }],
           ]]),
+          transformPayload: (bytes) => bytes,
+          processReceivedPayload: (bytes) => bytes,
         });
 
         const storeMapBetty = new StoreMap(
@@ -1997,6 +2023,8 @@ function testWgpsMessenger(scenario: WgpsScenario) {
               maxSize: BigInt(0),
             }],
           ]]),
+          transformPayload: (bytes) => bytes,
+          processReceivedPayload: (bytes) => bytes,
         });
 
         await delay(20 * scenario.timeMultiplier);
@@ -2211,6 +2239,8 @@ function testWgpsMessenger(scenario: WgpsScenario) {
               maxSize: BigInt(0),
             }],
           ]]),
+          transformPayload: (bytes) => bytes,
+          processReceivedPayload: (bytes) => bytes,
         });
 
         const storeMapBetty = new StoreMap(
@@ -2342,6 +2372,8 @@ function testWgpsMessenger(scenario: WgpsScenario) {
               maxSize: BigInt(0),
             }],
           ]]),
+          transformPayload: (bytes) => bytes,
+          processReceivedPayload: (bytes) => bytes,
         });
 
         await delay(20 * scenario.timeMultiplier);

diff --git a/src/wgps/wgps_messenger.ts b/src/wgps/wgps_messenger.ts
@@ -143,6 +143,11 @@ export type WgpsMessengerOpts<
     SubspaceId,
     PayloadDigest
   >;
+
+  /** A (not necessarily deterministic) algorithm that converts a chunk of a payload into another bytestring. */
+  transformPayload: (chunk: Uint8Array) => Uint8Array;
+  /** Process transformed payload chunks. */
+  processReceivedPayload: (chunk: Uint8Array) => Uint8Array;
 };
 
 /** Coordinates a complete WGPS synchronisation session. */
@@ -496,10 +501,12 @@ export class WgpsMessenger<
     this.dataSender = new DataSender({
       handlesPayloadRequestsTheirs: this.handlesPayloadRequestsTheirs,
       getStore: this.getStore,
+      transformPayload: opts.transformPayload,
     });
 
     this.payloadIngester = new PayloadIngester({
       getStore: this.getStore,
+      processReceivedPayload: opts.processReceivedPayload,
     });
 
     // Send encoded messages