Tweak default print behavior.

Author: rkistner

test-client/README.md

@@ -5,13 +5,17 @@ This is a minimal client demonstrating direct usage of the HTTP stream sync API.
 For a full implementation, see our client SDKs.
 
 ## Setup
+
 1. Install dependencies on the project root
+
 ```shell
 # In project root directory
 pnpm install
 pnpm build:packages
 ```
+
 2. Build the test-client in the `test-client` directory
+
 ```shell
 # In the test-client directory
 pnpm build
@@ -20,6 +24,7 @@ pnpm build
 ## Usage
 
 ### fetch-operations
+
 The `fetch-operations` command downloads data for a single checkpoint, and outputs a normalized form: one CLEAR operation, followed by the latest PUT operation for each row. This normalized form is still split per bucket. The output is not affected by compacting, but can be affected by replication order.
 
 To avoid normalizing the data, use the `--raw` option. This may include additional CLEAR, MOVE, REMOVE and duplicate PUT operations.
@@ -57,18 +62,19 @@ and other load-testing use cases. There are two modes available, `websocket` or
 
 ```shell
 # Send two concurrent requests to request a download of sync operations using -n to specify the number of connections
-node ./dist/bin.js concurrent-connections -n 2 -t <token> 
+node ./dist/bin.js concurrent-connections -n 2 -t <token>
 
 # Send two concurrent requests to request a download of sync operations using websocket mode
 node ./dist/bin.js concurrent-connections -n 2 -t <token> -m websocket
 ```
 
-Once the sync has completed for a connection the command will print the `op_id`, `ops`, `bytes`, `duration` and `data` for
-each connection. By default, the `data` value will be an array id's of the rows synced from the source database, but you can
-specify an additional argument if you want to print a specific field from the data being synced.
+Once the sync has completed for a connection the command will print the `op_id`, `ops`, `bytes` and `duration`
+each connection.
+
+To see what rows are being synced, specify `--print=id`, or another relevant field. This will be included
+in the output as a `data:` array with each checkpoint.
 
 ```shell
 # Send two concurrent requests and print the name field, as an example.
 node ./dist/bin.js concurrent-connections -n 2 -t <token> -p name
 ```
-

test-client/src/bin.ts

@@ -44,7 +44,7 @@ program
   .action(async (options) => {
     const credentials = await getCredentials(options);
 
-    await concurrentConnections(credentials, options['numClients'] ?? 10, options.mode ?? 'http', options.print ?? "id");
+    await concurrentConnections(credentials, options['numClients'] ?? 10, options.mode ?? 'http', options.print);
   });
 
 await program.parseAsync();

test-client/src/load-testing/http-worker.ts

@@ -14,14 +14,16 @@ let lastCheckpointStart = 0;
 let printData: string[] = [];
 
 const parseChunk = (chunk: any) => {
+  if (print == null) {
+    return;
+  }
   chunk.data.forEach((data: any) => {
-    if(data.op == "MOVE") {
-      return;
+    if (data.op == 'PUT') {
+      const payload = JSON.parse(data.data);
+      printData.push(payload[print]);
     }
-    const payload = JSON.parse(data.data);
-    printData.push(payload[print]);
-  })
-}
+  });
+};
 
 const response = await fetch(url + '/sync/stream', {
   method: 'POST',
@@ -43,11 +45,12 @@ for await (let chunk of ndjsonStream<any>(response.body)) {
   size += JSON.stringify(chunk).length;
   if (chunk?.checkpoint_complete) {
     const duration = performance.now() - lastCheckpointStart;
-    console.log(
-      new Date().toISOString(),
-      i,
-      `checkpoint_complete op_id: ${chunk.checkpoint_complete.last_op_id}, ops: ${numOperations}, bytes: ${size}, duration: ${duration.toFixed(0)}ms, data: [${printData}]`
-    );
+    let message = `checkpoint_complete op_id: ${chunk.checkpoint_complete.last_op_id}, ops: ${numOperations}, bytes: ${size}, duration: ${duration.toFixed(0)}ms`;
+    if (print) {
+      message += `, data: [${printData}]`;
+    }
+    printData = [];
+    console.log(new Date().toISOString(), i, message);
   } else if (chunk?.data) {
     parseChunk(chunk.data);
     numOperations += chunk.data.data.length;

test-client/src/load-testing/load-test.ts

@@ -3,23 +3,23 @@ import { Credentials } from '../auth.js';
 
 export type Mode = 'http' | 'websocket';
 
-export async function stream(i: number, credentials: Credentials, mode: Mode, print: string) {
+export async function stream(i: number, credentials: Credentials, mode: Mode, print: string | undefined) {
   const worker =
     mode == 'websocket'
       ? new Worker(new URL('./rsocket-worker.js', import.meta.url), {
-        workerData: { i, token: credentials.token, url: credentials.endpoint.replace(/^http/, 'ws'), print }
-      })
+          workerData: { i, token: credentials.token, url: credentials.endpoint.replace(/^http/, 'ws'), print }
+        })
       : new Worker(new URL('./http-worker.js', import.meta.url), {
-        workerData: { i, token: credentials.token, url: credentials.endpoint, print }
-      });
+          workerData: { i, token: credentials.token, url: credentials.endpoint, print }
+        });
   await new Promise((resolve, reject) => {
     worker.on('message', (event) => resolve(event));
     worker.on('error', (err) => reject(err));
   });
   worker.terminate();
 }
 
-export async function streamForever(i: number, credentials: Credentials, mode: Mode, print: string) {
+export async function streamForever(i: number, credentials: Credentials, mode: Mode, print: string | undefined) {
   while (true) {
     try {
       await stream(i, credentials, mode, print);
@@ -31,7 +31,12 @@ export async function streamForever(i: number, credentials: Credentials, mode: M
   }
 }
 
-export async function concurrentConnections(credentials: Credentials, numClients: number, mode: Mode, print: string) {
+export async function concurrentConnections(
+  credentials: Credentials,
+  numClients: number,
+  mode: Mode,
+  print: string | undefined
+) {
   for (let i = 0; i < numClients; i++) {
     streamForever(i, credentials, mode, print);
   }

test-client/src/load-testing/rsocket-worker.ts

@@ -73,11 +73,12 @@ const stream = rsocket.requestStream(
 
       if (chunk?.checkpoint_complete) {
         const duration = performance.now() - lastCheckpointStart;
-        console.log(
-          new Date().toISOString(),
-          i,
-          `checkpoint_complete op_id: ${chunk.checkpoint_complete.last_op_id}, ops: ${numOperations}, bytes: ${size}, duration: ${duration.toFixed(0)}ms, data: [${printData}]`
-        );
+        let message = `checkpoint_complete op_id: ${chunk.checkpoint_complete.last_op_id}, ops: ${numOperations}, bytes: ${size}, duration: ${duration.toFixed(0)}ms`;
+        if (print) {
+          message += `, data: [${printData}]`;
+        }
+        console.log(new Date().toISOString(), i, message);
+        printData = [];
       } else if (chunk?.data) {
         parseChunk(chunk.data);
         numOperations += chunk.data.data.length;
@@ -112,11 +113,13 @@ const stream = rsocket.requestStream(
 );
 
 const parseChunk = (chunk: any) => {
+  if (print == null) {
+    return;
+  }
   chunk.data.forEach((data: any) => {
-    if(data.op == "MOVE") {
-      return;
+    if (data.op == 'PUT') {
+      const payload = JSON.parse(data.data);
+      printData.push(payload[print]);
     }
-    const payload = JSON.parse(data.data);
-    printData.push(payload[print]);
-  })
-}
+  });
+};