@std/json

import { JsonStringifyStream } from "@std/json";
import { assertEquals } from "@std/assert";

const stream = ReadableStream.from([{ foo: "bar" }, { baz: 100 }])
  .pipeThrough(new JsonStringifyStream());

assertEquals(await Array.fromAsync(stream), [
  `{"foo":"bar"}\n`,
  `{"baz":100}\n`
]);

deno add jsr:@std/json

import {
  ConcatenatedJsonParseStream,
} from "@std/json/concatenated-json-parse-stream";

// 流中包含两个不带分隔符的连续 JSON 文档。
const input = ReadableStream.from([
  '{"a":1}{',
  '"b":2}',
]);

const parsed = input.pipeThrough(new ConcatenatedJsonParseStream());
console.log(await Array.fromAsync(parsed)); // [{ a: 1 }, { b: 2 }]

import { JsonStringifyStream } from "@std/json/stringify-stream";

const data = [{ id: 1 }, { id: 2 }, { id: 3 }];

// 在每个 JSON 值后添加换行符，形成 NDJSON
const ndjson = ReadableStream
  .from(data)
  .pipeThrough(new JsonStringifyStream({ suffix: "\n" }));

// 发送到接受 application/x-ndjson 的服务器
await fetch("/ingest", {
  method: "POST",
  headers: { "content-type": "application/x-ndjson" },
  body: ndjson,
});

import { JsonParseStream } from "@std/json/parse-stream";

// 按换行符拆分，即使一行跨越多个块
function lineSplitter() {
  let buffer = "";
  return new TransformStream<string, string>({
    transform(chunk, controller) {
      buffer += chunk;
      const lines = buffer.split(/\r?\n/);
      buffer = lines.pop() ?? ""; // 保留最后不完整的行
      for (const line of lines) if (line) controller.enqueue(line);
    },
    flush(controller) {
      if (buffer) controller.enqueue(buffer);
    },
  });
}

const res = await fetch("/stream.ndjson");
const values = res.body!
  .pipeThrough(new TextDecoderStream())
  .pipeThrough(lineSplitter())
  .pipeThrough(new JsonParseStream());

for await (const obj of values) {
  // 按到达顺序处理每个 JSON 对象
  console.log(obj);
}

import { JsonParseStream } from "@std/json/parse-stream";
import { JsonStringifyStream } from "@std/json/stringify-stream";

// 输入对象 -> 映射 -> 输出 NDJSON
function mapStream<T, U>(map: (t: T) => U) {
  return new TransformStream<T, U>({ transform: (t, c) => c.enqueue(map(t)) });
}

const response = await fetch("/objects.jsonl");
const uppercased = response.body!
  .pipeThrough(new TextDecoderStream())
  .pipeThrough(lineSplitter()) // 来自前面示例
  .pipeThrough(new JsonParseStream<{ name: string }>())
  .pipeThrough(mapStream((o) => ({ name: o.name.toUpperCase() })))
  .pipeThrough(new JsonStringifyStream({ suffix: "\n" }));

// 管道输出到另一个请求、文件或进一步处理
await fetch("/store", { method: "POST", body: uppercased });