var wait = async (ms) => new Promise((r) => setTimeout(r, ms));
var encoder = new TextEncoder();
var decoder = new TextDecoder();
var { writable, readable } = new TransformStream();
var abortable = new AbortController();
var { signal } = abortable;
var writer = writable.getWriter();
var settings = { url: "https://comfortable-deer-52.deno.dev", method: "query" };
fetch(settings.url, {
  duplex: "half",
  method: settings.method,
  body: readable.pipeThrough(
    new TransformStream({
      transform(value, c) {
        c.enqueue(encoder.encode(value));
      },
    }),
  ),
  signal,
})
  .then((r) =>
    r.body.pipeTo(
      new WritableStream({
        async start() {
          this.now = performance.now();
          console.log(this.now);
          return;
        },
        async write(value) {
          console.log(decoder.decode(value));
        },
        async close() {
          console.log("Stream closed");
        },
        async abort(reason) {
          const now = ((performance.now() - this.now) / 1000) / 60;
          console.log({ reason });
        },
      }),
    )
  ).catch(async (e) => {
    console.log(e);
  });
await wait(1000);
await writer.write("test");
await wait(1500);
await writer.write("test, again");
await writer.close();

user@debian:~/bin$ node full_duplex_fetch_test.js
141976.214995
TEST
TEST, AGAIN
Stream closed

user@debian:~/bin$ deno -A full_duplex_fetch_test.js
A new canary release of Deno is available. Run `deno upgrade canary` to install it.
39487.778984
TESTTEST, AGAIN
Stream closed
user@debian:~/bin$ bun full_duplex_fetch_test.js
357.140365
TEST
TEST, AGAIN
Stream closed

user@debian:~/bin$ tjs run full_duplex_fetch_test.js
Error: Unsupported payload type
    at send (polyfills.js:10:8288)
    at <anonymous> (polyfills.js:10:15684)
    at Promise (native)
    at fetch (polyfills.js:10:14397)
    at <anonymous> (full_duplex_fetch_test.js:13:9)
    at evalFile (native)
    at <anonymous> (run-main.js:27:1459)

var abortable = new AbortController();

var { readable, writable } = new TransformStream({
  async transform(v, c) {
    for (let i = 0; i < v.length; i += 8192) {
      c.enqueue(v.subarray(i, i + 8192));
      await scheduler.postTask(() => {}, { delay: 30 });
    }
  },
  flush() {
    console.log("flush");
    // abortable.abort("Stream over");
  },
});
var writer = writable.getWriter();
var response = fetch("http://127.0.0.1:44818", {
  method: "post",
  duplex: "half",
  body: readable,
  signal: abortable.signal,
  allowHTTP1ForStreamingUpload: true,
}).then((r) => {
  console.log(r);
  return r.body.pipeThrough(new TextDecoderStream()).pipeTo(
    new WritableStream({
      write(v) {
        console.log(v);
        //abortable.abort("a reason");
      },
      close() {
        console.log("close");
      },
      abort(reason) {
        console.log(reason);
      },
    }),
  );
})
  .catch((e) => {
    console.log(e);
  })
  .then(() => {
    console.log("Done streaming");
  })
  .catch(console.log);
await scheduler.postTask(() => {}, { delay: 45 });
await writer.write(new Uint8Array(1024**2*7).fill(1));
await writer.ready;
await writer.close();