From f5425aa1e0485ace2e0b7352eddf58ae8710ca2c Mon Sep 17 00:00:00 2001
From: Okiki Ojo <hey@okikio.dev>
Date: Sat, 21 Mar 2026 06:03:55 -0400
Subject: [PATCH] feat: enhance observableInputToStream with subscription
 management and cancellation support

Signed-off-by: Okiki Ojo <hey@okikio.dev>
---
 helpers/utils.ts | 8 +++++++-
 1 file changed, 7 insertions(+), 1 deletion(-)

diff --git a/helpers/utils.ts b/helpers/utils.ts
index 2514df9..6bb6654 100644
--- a/helpers/utils.ts
+++ b/helpers/utils.ts
@@ -250,11 +250,13 @@ export function observableInputToStream<T>(
   input: ObservableInteropInputLike<T>,
   errorContext: string,
 ): ReadableStream<T | ObservableError> {
+  let subscription: { unsubscribe?(): void } | void;
+
   return new ReadableStream<T | ObservableError>({
     start(controller) {
       const source = hasSubscribe<T>(input) ? input : Observable.from(input);
 
-      return source.subscribe({
+      subscription = source.subscribe({
         next(value) {
           try {
             controller.enqueue(value);
@@ -279,6 +281,10 @@ export function observableInputToStream<T>(
         },
       });
     },
+    cancel() {
+      subscription?.unsubscribe?.();
+      subscription = undefined;
+    },
   });
 }
 
-- 
2.51.2