Commit

Author: qwtel

async-queue.ts

@@ -49,7 +49,7 @@ export class AsyncQueue<T = any> implements AsyncIterableIterator<T> {
       this.#error = err;
     }
 
-    this.return(); 
+    this.return();
   }
 
   #abortListener = () => {
@@ -59,17 +59,17 @@ export class AsyncQueue<T = any> implements AsyncIterableIterator<T> {
   push(el: T) {
     const promise = this.#unconsumedPromises.shift();
     if (promise) {
-      promise.resolve({ value: el as T, done: false }); // FIXME
+      promise.resolve({ value: el, done: false });
     } else {
-      this.#unconsumedValues.push(el as T); // FIXME
+      this.#unconsumedValues.push(el);
     }
   }
 
   // TODO: does it make sense/is it possible to add `shift` / `pop`??
 
-  /** Alias for `next` */
-  shift(): Promise<IteratorResult<T, void>> {
-    return this.next();
+  async shift(): Promise<T | undefined> {
+    const { done, value } = await this.next();
+    return done ? undefined : value;
   }
 
   next(): Promise<IteratorResult<T, void>> {
@@ -99,7 +99,7 @@ export class AsyncQueue<T = any> implements AsyncIterableIterator<T> {
       this.#unconsumedPromises.push({ resolve, reject });
     });
   }
-  
+
   return(): Promise<IteratorResult<T, void>> {
     if (this.#signal) {
       this.#signal.removeEventListener('abort', this.#abortListener);

index.ts

@@ -1,5 +1,6 @@
 export * from './json-stringify.ts'
 export * from './json-parse-stream.ts'
 export * from './json-stringify-stream.ts'
+export * from './json-fetch.ts'
 export * from './ndjson-parse-stream.ts'
 export * from './ndjson-stringify-stream.ts'

json-fetch.ts

@@ -0,0 +1,58 @@
+// deno-lint-ignore-file no-explicit-any
+import { StreamResponse, StreamRequest } from "../stream-response/index.ts";
+import { JSONStringifyReadable } from './json-stringify.ts';
+
+export type JSONStreamBodyInit = any 
+export type JSONStreamRequestInit = Omit<RequestInit, 'body'> & { body?: JSONStreamBodyInit }
+
+export class JSONStreamRequest extends StreamRequest {
+  static contentType = 'application/json;charset=UTF-8';
+  static accept = 'application/json, text/plain, */*';
+
+  constructor(
+    input: RequestInfo | URL,
+    init?: JSONStreamRequestInit,
+    // replacer?: Parameters<typeof JSON.stringify>[1],
+    // space?: Parameters<typeof JSON.stringify>[2],
+  ) {
+    const { headers: _headers, body: _body, ...rest } = init || {};
+
+    const body = new JSONStringifyReadable(_body)
+
+    const headers = new Headers(_headers);
+    if (!headers.has('Content-Type') && _body != null)
+      headers.set('Content-Type', JSONStreamRequest.contentType);
+
+    if (!headers.has('Accept'))
+      headers.set('Accept', JSONStreamRequest.accept);
+
+    super(input instanceof URL ? input.href : input, { headers, body, ...rest });
+  }
+}
+
+export class JSONStreamResponse extends StreamResponse {
+  static contentType = 'application/json;charset=UTF-8';
+
+  constructor(
+    body?: JSONStreamBodyInit | null,
+    init?: ResponseInit,
+    // replacer?: Parameters<typeof JSON.stringify>[1],
+    // space?: Parameters<typeof JSON.stringify>[2],
+  ) {
+    const { headers: _headers, ...rest } = init || {};
+
+    const _body = new JSONStringifyReadable(body)
+
+    const headers = new Headers(_headers);
+
+    if (!headers.has('Content-Type') && body != null)
+      headers.set('Content-Type', JSONStreamResponse.contentType);
+
+    super(_body, { headers, ...rest });
+  }
+}
+
+export type {
+  JSONStreamRequest as JSONRequest,
+  JSONStreamResponse as JSONResponse,
+}

json-stringify.ts

@@ -1,16 +1,16 @@
 // deno-lint-ignore-file no-explicit-any no-empty
-import { asyncIterableToStream } from 'https://ghuc.cc/qwtel/whatwg-stream-to-async-iter/index.ts'
+import { asyncIterToStream } from 'https://ghuc.cc/qwtel/whatwg-stream-to-async-iter/index.ts'
 
 type SeenWeakSet = WeakSet<any>;
 
 type Primitive = undefined | boolean | number | string | bigint | symbol;
 
 export type ToJSON = { toJSON: (key?: any) => string }
 
-const _isIterable = <T>(x: unknown): x is Iterable<T> => 
+export const isIterable = <T>(x: unknown): x is Iterable<T> =>
   x != null && typeof x === 'object' && Symbol.iterator in x
 
-const isAsyncIterable = <T>(x: unknown): x is AsyncIterable<T> => 
+export const isAsyncIterable = <T>(x: unknown): x is AsyncIterable<T> =>
   x != null && typeof x === 'object' && Symbol.asyncIterator in x
 
 const isPromiseLike = <T>(x: unknown): x is PromiseLike<T> =>
@@ -103,7 +103,7 @@ export async function* jsonStringifyGenerator(
 export function jsonStringifyStream(
   value: null | Primitive | ToJSON | any[] | Record<string, any> | PromiseLike<any> | AsyncIterable<any> | ReadableStream,
 ): ReadableStream<string> {
-  return asyncIterableToStream(jsonStringifyGenerator(value))
+  return asyncIterToStream(jsonStringifyGenerator(value))
 }
 
 export class JSONStringifyReadable extends ReadableStream<string> {
@@ -118,7 +118,7 @@ export class JSONStringifyReadable extends ReadableStream<string> {
         if (!done) controller.enqueue(value); else controller.close();
       },
       async cancel(reason) {
-        try { await iterator.throw?.(reason) } catch {  }
+        try { await iterator.throw?.(reason) } catch { }
       },
     })
   }

test/json-fetch.test.ts

@@ -0,0 +1,76 @@
+// deno-lint-ignore-file no-unused-vars no-explicit-any
+import 'https://gist.githubusercontent.com/qwtel/b14f0f81e3a96189f7771f83ee113f64/raw/TestRequest.ts'
+import {
+  assert,
+  assertExists,
+  assertEquals,
+  assertStrictEquals,
+  assertStringIncludes,
+  assertThrows,
+  assertRejects,
+  assertArrayIncludes,
+} from 'https://deno.land/std@0.133.0/testing/asserts.ts'
+const { test } = Deno;
+
+import { JSONStreamResponse, JSONStreamRequest } from '../json-fetch.ts'
+import { JSONParseStream } from '../index.ts'
+
+test('exists', () =>{
+  assertExists(JSONStreamRequest)
+  assertExists(JSONStreamResponse)
+})
+
+test('simple response', async () => {
+  const actual = await new JSONStreamResponse({ a: 3, b: { nested: 4 }, c: [1, 2, 3], __x: undefined }).json()
+  assertEquals(actual, { a: 3, b: { nested: 4 }, c: [1, 2, 3] })
+})
+
+test('simple request', async () => {
+  const actual = await new JSONStreamRequest('/', { method: 'PUT', body: { a: 3, b: { nested: 4 }, c: [1, 2, 3], __x: undefined } }).json()
+  assertEquals(actual, { a: 3, b: { nested: 4 }, c: [1, 2, 3] })
+})
+
+test('with promise response', async () => {
+  const actual = await new JSONStreamResponse(({ a: 3, b: Promise.resolve(4) })).json()
+  assertEquals(actual, { a: 3, b: 4 })
+})
+
+test('with promise request', async () => {
+  const actual = await new JSONStreamRequest('/', { method: 'PUT', body: { a: 3, b: Promise.resolve(4) } }).json()
+  assertEquals(actual, { a: 3, b: 4 })
+})
+
+const timeout = (n?: number) => new Promise(r => setTimeout(r, n))
+async function* asyncGen<T>(xs: T[]) {
+  for (const x of xs) { await timeout(); yield x }
+}
+
+test('with generator response', async () => {
+  const actual = await new JSONStreamResponse(({ a: 3, b: Promise.resolve(4), c: asyncGen([1, 2, 3]) })).text()
+  assertEquals(actual, JSON.stringify({ a: 3, b: 4, c: [1, 2, 3] }))
+})
+
+test('with generator request', async () => {
+  const actual = await new JSONStreamRequest('/', { method: 'PUT', body: { a: 3, b: Promise.resolve(4), c: asyncGen([1, 2, 3]) } }).json()
+  assertEquals(actual, { a: 3, b: 4, c: [1, 2, 3] })
+})
+
+test('circular throws', () => {
+  const a: any = { a: 3, foo: { b: 4 } }
+  a.foo.a = a;
+  assertRejects(() => new JSONStreamResponse((a)).json(), TypeError)
+})
+
+test('GET with body throws', () => {
+  const a: any = { a: 3, foo: { b: 4 } }
+  assertRejects(() => new JSONStreamRequest('/', { body: a }).json(), TypeError)
+})
+
+test('stream', async () => {
+  const actual = new JSONStreamResponse(({ a: 3, b: Promise.resolve(4), c: asyncGen([1, 2, 3]) }))
+  const reader = actual.body!.pipeThrough(new JSONParseStream('$.c.*')).getReader()
+  assertEquals((await reader.read()).value, 1)
+  assertEquals((await reader.read()).value, 2)
+  assertEquals((await reader.read()).value, 3)
+  assertEquals((await reader.read()).done, true)
+})