///|
/// A chunked BSON frame decoder for transports that split documents arbitrarily.
pub struct BsonStreamDecoder {
  mut pending : Bytes
} derive(Debug)

///|
pub fn BsonStreamDecoder::new() -> BsonStreamDecoder {
  { pending: b"" }
}

///|
/// Feed a chunk and return every complete document now available.
pub fn BsonStreamDecoder::push(
  self : BsonStreamDecoder,
  chunk : BytesView,
) -> Array[Document] raise BsonError {
  let buffer = self.pending + chunk.to_owned()
  let documents : Array[Document] = []
  let options = DecodeOptions::new()
  let mut position = 0
  while position < buffer.length() {
    match bson_stream_frame_length(buffer[:], position, options.max_size()) {
      None => break
      Some(length) => {
        let (document, consumed) = decode_prefix_with_options(
          buffer[position:position + length],
          options,
        )
        documents.push(document)
        position += consumed
      }
    }
  }
  self.pending = bson_stream_pending(buffer, position)
  documents
}

///|
/// Finish the stream, rejecting an incomplete trailing frame.
pub fn BsonStreamDecoder::finish(
  self : BsonStreamDecoder,
) -> Unit raise BsonError {
  if !self.pending.is_empty() {
    raise bson_error(
      UnexpectedEnd,
      self.pending.length(),
      "$",
      "BSON stream ended with an incomplete document",
    )
  }
}

///|
/// A chunked BSON frame decoder that returns borrowed raw document views.
///
/// Complete frames are sliced from the decoder's immutable pending buffer; no
/// `Document` or per-frame byte copy is created. A frame split across chunks is
/// assembled in that buffer, and returned views remain valid after later
/// `push` calls.
pub struct BsonStreamRawDecoder {
  mut pending : Bytes
  max_size : Int
} derive(Debug)

///|
/// Create a raw decoder with a per-frame size limit (16 MiB by default).
pub fn BsonStreamRawDecoder::new(
  max_size? : Int = 16 * 1024 * 1024,
) -> BsonStreamRawDecoder {
  { pending: b"", max_size }
}

///|
/// Feed a chunk and return every complete raw view now available.
pub fn BsonStreamRawDecoder::push(
  self : BsonStreamRawDecoder,
  chunk : BytesView,
) -> Array[RawDocumentView] raise BsonError {
  let buffer = self.pending + chunk.to_owned()
  let views : Array[RawDocumentView] = []
  let mut position = 0
  while position < buffer.length() {
    match bson_stream_frame_length(buffer[:], position, self.max_size) {
      None => break
      Some(length) => {
        views.push(
          RawDocumentView::from_bytes(buffer[position:position + length]),
        )
        position += length
      }
    }
  }
  self.pending = bson_stream_pending(buffer, position)
  views
}

///|
/// Finish the stream, rejecting an incomplete trailing raw frame.
pub fn BsonStreamRawDecoder::finish(
  self : BsonStreamRawDecoder,
) -> Unit raise BsonError {
  if !self.pending.is_empty() {
    raise bson_error(
      UnexpectedEnd,
      self.pending.length(),
      "$",
      "BSON raw stream ended with an incomplete document",
    )
  }
}

///|
fn bson_stream_frame_length(
  buffer : BytesView,
  position : Int,
  max_size : Int,
) -> Int? raise BsonError {
  let remaining = buffer.length() - position
  if remaining < 4 {
    return None
  }
  let length = raw_view_i32(buffer, position, "$")
  if length < 5 {
    raise bson_error(
      InvalidLength,
      position,
      "$.length",
      "BSON stream frame length must be at least 5",
    )
  }
  if length > max_size {
    raise bson_error(
      SizeLimit,
      position,
      "$.length",
      "BSON stream frame exceeds max_size",
    )
  }
  if length > remaining {
    None
  } else {
    Some(length)
  }
}

///|
fn bson_stream_pending(buffer : Bytes, position : Int) -> Bytes {
  if position == 0 {
    buffer
  } else {
    // Returned views retain `buffer`; only the incomplete suffix needs a new
    // owner before the decoder accepts another chunk.
    buffer[position:].to_owned()
  }
}

///|
/// Append-only frame encoder for transports that batch complete BSON frames.
pub struct BsonStreamEncoder {
  buffer : @buffer.Buffer
}

///|
pub fn BsonStreamEncoder::new(size_hint? : Int = 256) -> BsonStreamEncoder {
  { buffer: Buffer(size_hint~) }
}

///|
/// Encode one complete document and append it as one frame.
pub fn BsonStreamEncoder::push(
  self : BsonStreamEncoder,
  document : Document,
) -> Unit raise BsonError {
  self.buffer.write_bytes(encode(document))
}

///|
pub fn BsonStreamEncoder::bytes(self : BsonStreamEncoder) -> Bytes {
  self.buffer.to_bytes()
}