Stream request response replies from queues

ts/packages/base/src/__tests__/messaging-runtime.test.ts

@@ -215,6 +215,53 @@ describe("Effect-native messaging runtime", () => {
     }),
   );
 
+  it.effect(
+    "waits until the request recipient accepts a response",
+    Effect.fnUntraced(function* () {
+      const responseConsumer = new ScriptedConsumer<unknown>();
+      const backend = new RuntimeBackend(
+        responseConsumer,
+        (_message, properties) => {
+          const id = properties?.id ?? "";
+          responseConsumer.push(createMessage("partial", { id }));
+          responseConsumer.push(createMessage("final", { id }));
+        },
+      );
+      const seen: Array<string> = [];
+
+      const response = yield* Effect.scoped(
+        Effect.gen(function* () {
+          const requestor = yield* makeEffectRequestResponseFromPubSub<string, string>(
+            PubSub.fromBackend(backend),
+            {
+              ...defaultMessagingRuntimeConfig,
+              consumerReceiveTimeoutMs: 1,
+            },
+            {
+              requestTopic: "tg.test.request",
+              responseTopic: "tg.test.response",
+              subscription: "sub",
+            },
+          );
+          const fiber = yield* requestor.request("request", {
+            timeoutMs: 250,
+            recipient: (candidate) =>
+              Effect.sync(() => {
+                seen.push(candidate);
+                return candidate === "final";
+              }),
+          }).pipe(Effect.forkChild);
+          yield* TestClock.adjust(Duration.millis(5));
+          return yield* Fiber.join(fiber);
+        }),
+      );
+
+      expect(response).toBe("final");
+      expect(seen).toEqual(["partial", "final"]);
+      expect(responseConsumer.acknowledged.length).toBe(2);
+    }),
+  );
+
   it.effect(
     "fails request-response calls with a typed timeout",
     Effect.fnUntraced(function* () {

ts/packages/base/src/messaging/runtime.ts

@@ -3,7 +3,7 @@
  */
 
 import { randomUUID } from "node:crypto";
-import { Context, Duration, Effect, Fiber, Layer, Queue, Scope } from "effect";
+import { Context, Duration, Effect, Fiber, Layer, Queue, Result, Scope, Stream } from "effect";
 import * as O from "effect/Option";
 import * as S from "effect/Schema";
 import type {
@@ -403,17 +403,27 @@ const waitForResponse = Effect.fn("waitForResponse")(function* <TRes, E, R>(
   queue: Queue.Queue<TRes>,
   options: EffectRequestOptions<TRes, E, R> | undefined,
 ) {
-  while (true) {
-    const response = yield* Queue.take(queue);
-    if (options?.recipient === undefined) {
-      return response;
-    }
+  const response = yield* Stream.fromQueue(queue).pipe(
+    Stream.filterMapEffect((candidate) => {
+      if (options?.recipient === undefined) {
+        return Effect.succeed(Result.succeed(candidate));
+      }
 
-    const complete = yield* options.recipient(response);
-    if (complete) {
-      return response;
-    }
-  }
+      return options.recipient(candidate).pipe(
+        Effect.map((complete) =>
+          complete
+            ? Result.succeed(candidate)
+            : Result.fail(undefined)
+        ),
+      );
+    }),
+    Stream.runHead,
+  );
+
+  return yield* O.match(response, {
+    onNone: () => Effect.never,
+    onSome: Effect.succeed,
+  });
 });
 
 export const makeEffectRequestResponseFromPubSub = Effect.fn("makeEffectRequestResponseFromPubSub")(function* <