diff --git a/packages/deno/src/utils/streaming.ts b/packages/deno/src/utils/streaming.ts index 045ad0735fb2..491131e6e093 100644 --- a/packages/deno/src/utils/streaming.ts +++ b/packages/deno/src/utils/streaming.ts @@ -103,5 +103,8 @@ function monitorStream( controller.close(); reader.releaseLock(); }, + cancel(reason) { + return reader.cancel(reason); + }, }); } diff --git a/packages/deno/test/streaming.test.ts b/packages/deno/test/streaming.test.ts index d9849ece3c76..cfa040ac01cf 100644 --- a/packages/deno/test/streaming.test.ts +++ b/packages/deno/test/streaming.test.ts @@ -1,6 +1,34 @@ // -import { assertEquals } from 'https://deno.land/std@0.212.0/assert/mod.ts'; +import { assertEquals, assertStrictEquals } from 'https://deno.land/std@0.212.0/assert/mod.ts'; + +import { streamResponse } from '../src/utils/streaming.ts'; + +Deno.test('cancels the source when the response reader is cancelled', async () => { + let sourceCancelReason: unknown; + + const source = new ReadableStream({ + start(controller) { + controller.enqueue(new TextEncoder().encode('first event')); + }, + cancel(reason) { + sourceCancelReason = reason; + }, + }); + + const span = { end() {} } as Parameters[0]; + const response = await streamResponse( + span, + new Response(source, { headers: { 'content-type': 'text/event-stream' } }), + ); + const reader = response.body!.getReader(); + const reason = new Error('client disconnected'); + + await reader.read(); + await reader.cancel(reason); + + assertStrictEquals(sourceCancelReason, reason); +}); Deno.test('reader.closed.then(f, f) suppresses rejection when releaseLock is called on an open stream', async () => { // Reproduces the bug from GitHub issue #20177: