fix(js)!: client streamFlow returns properties, not functions (#1690)

Reconciles the server & client SDK to return a promise and async iterable rather than a function that returns these types. Docs already suggested we did this.

Author: inlined

js/core/src/async.ts

@@ -14,6 +14,9 @@
  * limitations under the License.
  */
 
+// NOTE: This file is pulled into client code and cannot have any Node-only
+// dependencies.
+
 /**
  * A handle to a promise and its resolvers.
  */

js/genkit/src/client/client.ts

@@ -14,6 +14,8 @@
  * limitations under the License.
  */
 
+import { Channel } from '@genkit-ai/core/async';
+
 const __flowStreamDelimiter = '\n\n';
 
 /**
@@ -43,49 +45,25 @@ export function streamFlow<O = any, S = any>({
   input?: any;
   headers?: Record<string, string>;
 }): {
-  output(): Promise<O>;
-  stream(): AsyncIterable<S>;
+  readonly output: Promise<O>;
+  readonly stream: AsyncIterable<S>;
 } {
-  let chunkStreamController: ReadableStreamDefaultController | undefined =
-    undefined;
-  const chunkStream = new ReadableStream({
-    start(controller) {
-      chunkStreamController = controller;
-    },
-    pull() {},
-    cancel() {},
-  });
+  const channel = new Channel<S>();
 
   const operationPromise = __flowRunEnvelope({
     url,
     input,
-    streamingCallback: (c) => {
-      chunkStreamController?.enqueue(c);
-    },
+    streamingCallback: (c) => channel.send(c),
     headers,
   });
-  operationPromise.then((o) => {
-    chunkStreamController?.close();
-    return o;
-  });
+  operationPromise.then(
+    () => channel.close(),
+    (err) => channel.error(err)
+  );
 
   return {
-    output() {
-      return operationPromise;
-    },
-    async *stream() {
-      const reader = chunkStream.getReader();
-      while (true) {
-        const chunk = await reader.read();
-        if (chunk?.value !== undefined) {
-          yield chunk.value;
-        }
-        if (chunk.done) {
-          break;
-        }
-      }
-      return await operationPromise;
-    },
+    output: operationPromise,
+    stream: channel,
   };
 }
 

js/plugins/express/README.md

@@ -80,10 +80,10 @@ const result = streamFlow({
   url: `http://localhost:${port}/simpleFlow`,
   input: 'say hello',
 });
-for await (const chunk of result.stream()) {
+for await (const chunk of result.stream) {
   console.log(chunk);
 }
-console.log(await result.output());
+console.log(await result.output);
 ```
 
 The sources for this package are in the main [Genkit](https://github.com/firebase/genkit) repo. Please file issues and pull requests against that repo.

js/plugins/express/tests/express_test.ts

@@ -283,7 +283,7 @@ describe('expressHandler', async () => {
       });
 
       const gotChunks: GenerateResponseChunkData[] = [];
-      for await (const chunk of result.stream()) {
+      for await (const chunk of result.stream) {
         gotChunks.push(chunk);
       }
 
@@ -293,7 +293,7 @@ describe('expressHandler', async () => {
         { index: 0, role: 'model', content: [{ text: '1' }] },
       ]);
 
-      assert.strictEqual(await result.output(), 'Echo: olleh');
+      assert.strictEqual(await result.output, 'Echo: olleh');
     });
 
     it('stream a model', async () => {
@@ -310,11 +310,11 @@ describe('expressHandler', async () => {
       });
 
       const gotChunks: any[] = [];
-      for await (const chunk of result.stream()) {
+      for await (const chunk of result.stream) {
         gotChunks.push(chunk);
       }
 
-      const output = await result.output();
+      const output = await result.output;
       assert.strictEqual(output.finishReason, 'stop');
       assert.deepStrictEqual(output.message, {
         role: 'model',
@@ -502,7 +502,7 @@ describe('startFlowServer', async () => {
       });
 
       const gotChunks: GenerateResponseChunkData[] = [];
-      for await (const chunk of result.stream()) {
+      for await (const chunk of result.stream) {
         gotChunks.push(chunk);
       }
 
@@ -512,7 +512,7 @@ describe('startFlowServer', async () => {
         { index: 0, role: 'model', content: [{ text: '1' }] },
       ]);
 
-      assert.strictEqual(await result.output(), 'Echo: olleh');
+      assert.strictEqual(await result.output, 'Echo: olleh');
     });
   });
 });

js/plugins/firebase/tests/functions_test.ts

@@ -166,11 +166,11 @@ describe('function', () => {
     });
 
     const chunks: any[] = [];
-    for await (const chunk of result.stream()) {
+    for await (const chunk of result.stream) {
       chunks.push(chunk);
     }
 
-    expect(await result.output()).toBe('hi Pavel - {"user":"Ali Baba"}');
+    expect(await result.output).toBe('hi Pavel - {"user":"Ali Baba"}');
     expect(chunks).toStrictEqual([{ chubk: 1 }, { chubk: 2 }, { chubk: 3 }]);
   });
 });

samples/chatbot/genkit-app/src/app/samples/chatbot/chatbot.component.ts

@@ -117,7 +117,7 @@ export class ChatbotComponent {
       });
 
       let textBlock: OutputSchema | undefined = undefined;
-      for await (const chunk of response.stream()) {
+      for await (const chunk of response.stream) {
         for (const content of chunk.content) {
           if (content.text) {
             if (!textBlock) {
@@ -133,7 +133,7 @@ export class ChatbotComponent {
       this.loading = false;
       this.chatFormControl.enable();
 
-      await response.output();
+      await response.output;
     } catch (e) {
       this.loading = false;
       this.chatFormControl.enable();

samples/js-angular/genkit-app/src/app/samples/chatbot/chatbot.component.ts

@@ -109,7 +109,7 @@ export class ChatbotComponent {
       });
 
       let textBlock: OutputSchema | undefined = undefined;
-      for await (const chunk of response.stream()) {
+      for await (const chunk of response.stream) {
         for (const content of chunk.content) {
           if (content.text) {
             if (!textBlock) {

samples/js-angular/genkit-app/src/app/samples/streaming-json/streaming-json.component.ts

@@ -44,10 +44,10 @@ export class StreamingJSONComponent {
         url,
         input: parseInt(this.count),
       });
-      for await (const chunk of response.stream()) {
+      for await (const chunk of response.stream) {
         this.characters = chunk;
       }
-      console.log('streamConsumer done', await response.output());
+      console.log('streamConsumer done', await response.output);
       this.loading = false;
     } catch (e) {
       this.loading = false;

tests/src/flow_server_test.ts

@@ -68,7 +68,7 @@ async function testFlowServer() {
         input: test.post.data,
       });
 
-      for await (const chunk of response.stream()) {
+      for await (const chunk of response.stream) {
         expected = want.message.replace('{count}', chunkCount.toString());
         let chunkJSON = JSON.stringify(await chunk);
         if (chunkJSON != expected) {
@@ -83,7 +83,7 @@ async function testFlowServer() {
           `unexpected number of stream chunks received: got ${chunkCount}, want: ${test.post.data}`
         );
       }
-      let out = await response.output();
+      let out = await response.output;
       want.result = want.result.replace(/\{count\}/g, chunkCount.toString());
       if (out != want.result) {
         throw new Error(