#Asynchronous I/O (@moonbitlang/core/async/io)

#Quick start

///|
async test "binary pipe quick start" {
  @async.with_task_group() <| root => {
    let (reader, writer) = @io.pipe()
    defer reader.close()
    root.spawn_bg() <| () => {
      defer writer.close()
      writer.write(b"Hello, ")
      writer.write(b"MoonBit!")
    }
    inspect(@utf8.decode(reader.read_all()), content="Hello, MoonBit!")
  }
}

#Binary streams

#BinaryReader

• read_into(dst, offset?, max_len?) reads into caller-owned storage and returns the number of bytes read. A zero return value indicates EOF.
• read() returns an owned chunk as soon as data is available, or None at EOF.
• read_at_least(min_len) returns an owned chunk containing at least min_len bytes, and may return more.
• peek() returns a zero-copy view as soon as data is available, or None at EOF, without consuming it.
• peek_at_least(min_len) returns a zero-copy view containing at least min_len bytes without consuming it.

///|
async test "read into caller-owned storage" {
  @async.with_task_group() <| root => {
    let (reader, writer) = @io.pipe()
    defer reader.close()
    root.spawn_bg() <| () => {
      defer writer.close()
      writer.write(b"hello")
    }
    let dst = FixedArray::make(9, b'.')
    let count = reader.read_into(dst, offset=2, max_len=5)
    inspect(count, content="5")
    inspect(
      @utf8.decode(dst.unsafe_reinterpret_as_bytes()[2:7]),
      content="hello",
    )
  }
}

///|
async test "peek then consume an exact frame" {
  @async.with_task_group() <| root => {
    let (reader, writer) = @io.pipe()
    defer reader.close()
    root.spawn_bg() <| () => {
      defer writer.close()
      writer.write(b"HEADbody")
    }
    let preview = reader.peek_exactly(4).to_owned()
    inspect(@utf8.decode(preview), content="HEAD")
    inspect(@utf8.decode(reader.read_exactly(4)), content="HEAD")
    inspect(@utf8.decode(reader.read_all()), content="body")
  }
}

• read_exactly(len) and peek_exactly(len) wait for exactly len bytes and raise ReaderClosed if EOF arrives first.
• drop_exactly(len) consumes exactly len bytes without returning them.
• read_all() drains the stream into a Bytes value.
• drop_all() drains the stream and returns the number of discarded bytes.
• read_until(pattern, include_delimiter?) reads through the next delimiter.
• peek_until(pattern, include_delimiter?) finds the same segment without consuming it.

///|
async test "read delimiter-separated binary records" {
  @async.with_task_group() <| root => {
    let (reader, writer) = @io.pipe()
    defer reader.close()
    root.spawn_bg() <| () => {
      defer writer.close()
      writer.write(b"alpha|beta")
    }
    guard reader.read_until(b"|") is Some(first) else {
      fail("missing first record")
    }
    inspect(@utf8.decode(first), content="alpha")
    guard reader.read_until(b"|") is Some(second) else {
      fail("missing final record")
    }
    inspect(@utf8.decode(second), content="beta")
    assert_eq(reader.read_until(b"|"), None)
  }
}

#BinaryWriter

///|
async test "write a byte view" {
  @async.with_task_group() <| root => {
    let (reader, writer) = @io.pipe()
    defer reader.close()
    root.spawn_bg() <| () => {
      defer writer.close()
      let packet = b"xxpayloadyy"
      writer.write(packet[2:9])
    }
    inspect(@utf8.decode(reader.read_all()), content="payload")
  }
}

#Text streams and UTF-8 adapters

• TextReader::read_all() decodes all remaining bytes as UTF-8.
• TextReader::read_until(pattern, include_delimiter?) searches for a text delimiter and returns decoded text.
• TextWriter::write(text) UTF-8 encodes a StringView.
• TextWriter::write_json(json) writes the JSON text representation.

///|
async test "UTF-8 text adapters" {
  @async.with_task_group() <| root => {
    let (reader, writer) = @io.pipe()
    defer reader.close()
    root.spawn_bg() <| () => {
      defer writer.close()
      let text_writer = @io.Utf8Writer(writer)
      text_writer.write("你好|MoonBit")
    }
    let text_reader = @io.Utf8Reader(reader)
    guard text_reader.read_until("|") is Some(first) else {
      fail("missing first text record")
    }
    inspect(first, content="你好")
    inspect(text_reader.read_all(), content="MoonBit")
  }
}

#MemoryReader

///|
async test "callback-backed memory reader" {
  let reader = @io.MemoryReader() <| writer => {
    writer.write(b"generated ")
    @async.sleep(10)
    writer.write(b"content")
  }
  defer reader.close()
  inspect(@utf8.decode(reader.read_all()), content="generated content")
}

#Buffered writing

///|
async test "buffered writer" {
  @async.with_task_group() <| root => {
    let (reader, writer) = @io.pipe()
    defer reader.close()
    root.spawn_bg() <| () => {
      defer writer.close()
      let buffered = @io.BufferedWriter::new(writer, size=4)
      buffered.write(b"ab")
      buffered.write(b"cd")
      buffered.write(b"ef")
      buffered.flush()
    }
    inspect(@utf8.decode(reader.read_all()), content="abcdef")
  }
}

#Implementing an endpoint

///|
pub(open) trait BinaryReaderImpl {
  fn get_controller(Self) -> BinaryReaderController
  async fn raw_read_impl(
    Self,
    FixedArray[Byte],
    offset~ : Int32,
    max_len~ : Int32,
  ) -> Int32
}

///|
pub(open) trait BinaryWriterImpl {
  fn get_controller(Self) -> BinaryWriterController
  async fn raw_write_impl(Self, Bytes, offset~ : Int32, len~ : Int32) -> Int32
}

#Usage notes

• Run only one operation at a time on a given reader or writer. The interfaces do not serialize concurrent calls.
• Use peek for parser lookahead, but copy a returned view before another reader operation if it must be retained.
• Prefer read_into for caller-managed buffers, read for streaming chunks, and read_all only when the remaining input comfortably fits in memory.
• Use Utf8Reader and Utf8Writer at the text boundary instead of mixing text conversion into binary protocol code.
• Flush a BufferedWriter at protocol boundaries and before closing its underlying writer.