fetch-stream-json-parser.ts basic code example

fetch-stream-json-parser.ts

// https://github.com/dominictarr/JSONStream
// @ts-ignore
import JSONStream from "JSONStream";

import { Observable } from "rxjs";

class Bla {
  public creativesOfChannelV4StreamEndpoint<T=any>(
    channel: T,
    brandId: string,
    period: number,
    endTime: Date,
    offset: number,
    limit: number,
  ): Observable<any> {
    const startTime = new Date(endTime);
    startTime.setDate(startTime.getDate() - period);

    return new Observable<any>(subscriber => {
      const ctrl = new AbortController();

      const json = {
        brandId,
        channel,
        start_time: startTime,
        end_time: endTime,
        limit,
        offset,
      };


      (async () => {
        const r = await fetch(`https://sheker.sheker/creatives`, {
          headers: [["Content-Type", "application/json"]],
          method: "post",
          body: JSON.stringify(json),
          credentials: "include",
          cache: "no-cache",
          mode: "cors",
          signal: ctrl.signal,
        });

        if (!r.body) {
          throw new Error("No body");
        }
        const reader = r.body.getReader();
        // https://developer.mozilla.org/en-US/docs/Web/API/TextDecoder/decode
        const decoder = new TextDecoder("utf8");
        const jsonStreamParser = JSONStream.parse("*", dateJSONFieldsInPlaceReviver);

        jsonStreamParser.on("data", (row: any) => {
          subscriber.next(row);
        });

        jsonStreamParser.on("end", () => {
          subscriber.complete();
        });

        let chunk = await reader.read();
        while (!chunk.done && !ctrl.signal.aborted) {
          jsonStreamParser.write(decoder.decode(chunk.value, { stream: true }));
          chunk = await reader.read();
        }
        jsonStreamParser.write(decoder.decode(new ArrayBuffer(0), { stream: false }));
        jsonStreamParser.end();
      })();

      return function cleanup() {
        ctrl.abort();
      };
    });
  }
}

export function dateJSONFieldsInPlaceReviver(o: any) {
  for (const f of Object.keys(o)) {
    o[f] = dateReviver(f, o[f]);
  }

  return o;
}

export function dateReviver(key: string, value: any) {
  if (typeof value === "string" && key) {
    const a = /^(\d{4})-(\d{2})-(\d{2})T(\d{2}):(\d{2}):(\d{2}(?:\.\d*)?)Z$/.exec(value);
    if (a) {
      return new Date(Date.UTC(+a[1], +a[2] - 1, +a[3], +a[4], +a[5], +a[6]));
    }
  }

  return value;
}