Improved tests

src/cloudflare/internal/pipeline-transform.ts

@@ -37,7 +37,7 @@ async function* readLines(
 type Batch = {
   id: string; // unique identifier for the batch
   shard: string; // assigned shard
-  ts: number; // timestamp of the event
+  ts: number; // creation timestamp of the batch
 
   format: Format;
   size: {
@@ -118,6 +118,10 @@ export class PipelineTransformImpl extends entrypoints.WorkerEntrypoint {
   }
 
   #sendJson(data: object[]): JsonStream {
+    if (!(data instanceof Array)) {
+      throw new Error('transformJson must return an array of objects');
+    }
+
     let written = 0;
     const encoder = new TextEncoder();
     const readable = new ReadableStream<Uint8Array>({

src/cloudflare/internal/test/pipeline-transform/transform-test.js

@@ -15,6 +15,8 @@ const customTransform = class MyEntrypoint extends PipelineTransform {
   async transformJson(batch) {
     for (const obj of batch) {
       obj.dispatcher = 'was here!';
+      await new Promise((resolve) => setTimeout(resolve, 50));
+      obj.wait = 'happened!';
     }
 
     return batch;
@@ -96,11 +98,14 @@ export const tests = {
       for (const line of resultLines) {
         objects.push(JSON.parse(line));
       }
+      assert.equal(objects.length, 3);
 
       let index = 0;
       for (const obj of objects) {
         assert.equal(obj.dispatcher, 'was here!');
         delete obj.dispatcher;
+        assert.equal(obj.wait, 'happened!');
+        delete obj.wait;
 
         assert.equal(`${JSON.stringify(obj)}\n`, lines[index]);
         index++;