Skip to content

Commit

Permalink
Replace gix-packetline-blocking/src with generated files
Browse files Browse the repository at this point in the history
These are copies, except for their headers, and generated by the
etc/copy-packetline.sh script.

This should make the new CI checks, which use the script, pass.
  • Loading branch information
EliahKagan committed Apr 9, 2024
1 parent a975339 commit a74573b
Show file tree
Hide file tree
Showing 18 changed files with 2,095 additions and 1 deletion.
1 change: 0 additions & 1 deletion gix-packetline-blocking/src

This file was deleted.

148 changes: 148 additions & 0 deletions gix-packetline-blocking/src/decode.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,148 @@
//! DO NOT EDIT - this is a copy of gix-packetline/src/decode.rs. Run `just copy-packetline` to update it.
use bstr::BString;

use crate::{PacketLineRef, DELIMITER_LINE, FLUSH_LINE, MAX_DATA_LEN, MAX_LINE_LEN, RESPONSE_END_LINE, U16_HEX_BYTES};

/// The error used in the [`decode`][mod@crate::decode] module
#[derive(Debug, thiserror::Error)]
#[allow(missing_docs)]
pub enum Error {
#[error("Failed to decode the first four hex bytes indicating the line length: {err}")]
HexDecode { err: String },
#[error("The data received claims to be larger than the maximum allowed size: got {length_in_bytes}, exceeds {MAX_DATA_LEN}")]
DataLengthLimitExceeded { length_in_bytes: usize },
#[error("Received an invalid empty line")]
DataIsEmpty,
#[error("Received an invalid line of length 3")]
InvalidLineLength,
#[error("{data:?} - consumed {bytes_consumed} bytes")]
Line { data: BString, bytes_consumed: usize },
#[error("Needing {bytes_needed} additional bytes to decode the line successfully")]
NotEnoughData { bytes_needed: usize },
}

///
#[allow(clippy::empty_docs)]
pub mod band {
/// The error used in [`PacketLineRef::decode_band()`][super::PacketLineRef::decode_band()].
#[derive(Debug, thiserror::Error)]
#[allow(missing_docs)]
pub enum Error {
#[error("attempt to decode a non-side channel line or input was malformed: {band_id}")]
InvalidSideBand { band_id: u8 },
#[error("attempt to decode a non-data line into a side-channel band")]
NonDataLine,
}
}

/// A utility return type to support incremental parsing of packet lines.
#[derive(Debug, Clone)]
pub enum Stream<'a> {
/// Indicate a single packet line was parsed completely
Complete {
/// The parsed packet line
line: PacketLineRef<'a>,
/// The amount of bytes consumed from input
bytes_consumed: usize,
},
/// A packet line could not yet be parsed due to missing bytes
Incomplete {
/// The amount of additional bytes needed for the parsing to complete
bytes_needed: usize,
},
}

/// The result of [`hex_prefix()`] indicating either a special packet line or the amount of wanted bytes
pub enum PacketLineOrWantedSize<'a> {
/// The special kind of packet line decoded from the hex prefix. It never contains actual data.
Line(PacketLineRef<'a>),
/// The amount of bytes indicated by the hex prefix of the packet line.
Wanted(u16),
}

/// Decode the `four_bytes` packet line prefix provided in hexadecimal form and check it for validity.
pub fn hex_prefix(four_bytes: &[u8]) -> Result<PacketLineOrWantedSize<'_>, Error> {
debug_assert_eq!(four_bytes.len(), 4, "need four hex bytes");
for (line_bytes, line_type) in &[
(FLUSH_LINE, PacketLineRef::Flush),
(DELIMITER_LINE, PacketLineRef::Delimiter),
(RESPONSE_END_LINE, PacketLineRef::ResponseEnd),
] {
if four_bytes == *line_bytes {
return Ok(PacketLineOrWantedSize::Line(*line_type));
}
}

let mut buf = [0u8; U16_HEX_BYTES / 2];
faster_hex::hex_decode(four_bytes, &mut buf).map_err(|err| Error::HexDecode { err: err.to_string() })?;
let wanted_bytes = u16::from_be_bytes(buf);

if wanted_bytes == 3 {
return Err(Error::InvalidLineLength);
}
if wanted_bytes == 4 {
return Err(Error::DataIsEmpty);
}
debug_assert!(
wanted_bytes as usize > U16_HEX_BYTES,
"by now there should be more wanted bytes than prefix bytes"
);
Ok(PacketLineOrWantedSize::Wanted(wanted_bytes - U16_HEX_BYTES as u16))
}

/// Obtain a `PacketLine` from `data` after assuring `data` is small enough to fit.
pub fn to_data_line(data: &[u8]) -> Result<PacketLineRef<'_>, Error> {
if data.len() > MAX_LINE_LEN {
return Err(Error::DataLengthLimitExceeded {
length_in_bytes: data.len(),
});
}

Ok(PacketLineRef::Data(data))
}

/// Decode `data` as packet line while reporting whether the data is complete or not using a [`Stream`].
pub fn streaming(data: &[u8]) -> Result<Stream<'_>, Error> {
let data_len = data.len();
if data_len < U16_HEX_BYTES {
return Ok(Stream::Incomplete {
bytes_needed: U16_HEX_BYTES - data_len,
});
}
let wanted_bytes = match hex_prefix(&data[..U16_HEX_BYTES])? {
PacketLineOrWantedSize::Wanted(s) => s as usize,
PacketLineOrWantedSize::Line(line) => {
return Ok(Stream::Complete {
line,
bytes_consumed: 4,
})
}
} + U16_HEX_BYTES;
if wanted_bytes > MAX_LINE_LEN {
return Err(Error::DataLengthLimitExceeded {
length_in_bytes: wanted_bytes,
});
}
if data_len < wanted_bytes {
return Ok(Stream::Incomplete {
bytes_needed: wanted_bytes - data_len,
});
}

Ok(Stream::Complete {
line: to_data_line(&data[U16_HEX_BYTES..wanted_bytes])?,
bytes_consumed: wanted_bytes,
})
}

/// Decode an entire packet line from data or fail.
///
/// Note that failure also happens if there is not enough data to parse a complete packet line, as opposed to [`streaming()`] decoding
/// succeeds in that case, stating how much more bytes are required.
pub fn all_at_once(data: &[u8]) -> Result<PacketLineRef<'_>, Error> {
match streaming(data)? {
Stream::Complete { line, .. } => Ok(line),
Stream::Incomplete { bytes_needed } => Err(Error::NotEnoughData { bytes_needed }),
}
}
215 changes: 215 additions & 0 deletions gix-packetline-blocking/src/encode/async_io.rs
Original file line number Diff line number Diff line change
@@ -0,0 +1,215 @@
//! DO NOT EDIT - this is a copy of gix-packetline/src/encode/async_io.rs. Run `just copy-packetline` to update it.
use std::{
io,
pin::Pin,
task::{Context, Poll},
};

use futures_io::AsyncWrite;
use futures_lite::AsyncWriteExt;

use super::u16_to_hex;
use crate::{encode::Error, Channel, DELIMITER_LINE, ERR_PREFIX, FLUSH_LINE, MAX_DATA_LEN, RESPONSE_END_LINE};

pin_project_lite::pin_project! {
/// A way of writing packet lines asynchronously.
pub struct LineWriter<'a, W> {
#[pin]
pub(crate) writer: W,
pub(crate) prefix: &'a [u8],
pub(crate) suffix: &'a [u8],
state: State<'a>,
}
}

enum State<'a> {
Idle,
WriteHexLen([u8; 4], usize),
WritePrefix(&'a [u8]),
WriteData(usize),
WriteSuffix(&'a [u8]),
}

impl<'a, W: AsyncWrite + Unpin> LineWriter<'a, W> {
/// Create a new line writer writing data with a `prefix` and `suffix`.
///
/// Keep the additional `prefix` or `suffix` buffers empty if no prefix or suffix should be written.
pub fn new(writer: W, prefix: &'a [u8], suffix: &'a [u8]) -> Self {
LineWriter {
writer,
prefix,
suffix,
state: State::Idle,
}
}

/// Consume self and reveal the inner writer.
pub fn into_inner(self) -> W {
self.writer
}
}

fn into_io_err(err: Error) -> io::Error {
io::Error::new(io::ErrorKind::Other, err)
}

impl<W: AsyncWrite + Unpin> AsyncWrite for LineWriter<'_, W> {
fn poll_write(self: Pin<&mut Self>, cx: &mut Context<'_>, data: &[u8]) -> Poll<io::Result<usize>> {
use futures_lite::ready;
let mut this = self.project();
loop {
match &mut this.state {
State::Idle => {
let data_len = this.prefix.len() + data.len() + this.suffix.len();
if data_len > MAX_DATA_LEN {
return Poll::Ready(Err(into_io_err(Error::DataLengthLimitExceeded {
length_in_bytes: data_len,
})));
}
if data.is_empty() {
return Poll::Ready(Err(into_io_err(Error::DataIsEmpty)));
}
let data_len = data_len + 4;
let len_buf = u16_to_hex(data_len as u16);
*this.state = State::WriteHexLen(len_buf, 0)
}
State::WriteHexLen(hex_len, written) => {
while *written != hex_len.len() {
let n = ready!(this.writer.as_mut().poll_write(cx, &hex_len[*written..]))?;
if n == 0 {
return Poll::Ready(Err(io::ErrorKind::WriteZero.into()));
}
*written += n;
}
if this.prefix.is_empty() {
*this.state = State::WriteData(0)
} else {
*this.state = State::WritePrefix(this.prefix)
}
}
State::WritePrefix(buf) => {
while !buf.is_empty() {
let n = ready!(this.writer.as_mut().poll_write(cx, buf))?;
if n == 0 {
return Poll::Ready(Err(io::ErrorKind::WriteZero.into()));
}
let (_, rest) = std::mem::take(buf).split_at(n);
*buf = rest;
}
*this.state = State::WriteData(0)
}
State::WriteData(written) => {
while *written != data.len() {
let n = ready!(this.writer.as_mut().poll_write(cx, &data[*written..]))?;
if n == 0 {
return Poll::Ready(Err(io::ErrorKind::WriteZero.into()));
}
*written += n;
}
if this.suffix.is_empty() {
let written = 4 + this.prefix.len() + *written;
*this.state = State::Idle;
return Poll::Ready(Ok(written));
} else {
*this.state = State::WriteSuffix(this.suffix)
}
}
State::WriteSuffix(buf) => {
while !buf.is_empty() {
let n = ready!(this.writer.as_mut().poll_write(cx, buf))?;
if n == 0 {
return Poll::Ready(Err(io::ErrorKind::WriteZero.into()));
}
let (_, rest) = std::mem::take(buf).split_at(n);
*buf = rest;
}
*this.state = State::Idle;
return Poll::Ready(Ok(4 + this.prefix.len() + data.len() + this.suffix.len()));
}
}
}
}

fn poll_flush(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
let this = self.project();
this.writer.poll_flush(cx)
}

fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<io::Result<()>> {
let this = self.project();
this.writer.poll_close(cx)
}
}

async fn prefixed_and_suffixed_data_to_write(
prefix: &[u8],
data: &[u8],
suffix: &[u8],
mut out: impl AsyncWrite + Unpin,
) -> io::Result<usize> {
let data_len = prefix.len() + data.len() + suffix.len();
if data_len > MAX_DATA_LEN {
return Err(into_io_err(Error::DataLengthLimitExceeded {
length_in_bytes: data_len,
}));
}
if data.is_empty() {
return Err(into_io_err(Error::DataIsEmpty));
}

let data_len = data_len + 4;
let buf = u16_to_hex(data_len as u16);

out.write_all(&buf).await?;
if !prefix.is_empty() {
out.write_all(prefix).await?;
}
out.write_all(data).await?;
if !suffix.is_empty() {
out.write_all(suffix).await?;
}
Ok(data_len)
}

async fn prefixed_data_to_write(prefix: &[u8], data: &[u8], out: impl AsyncWrite + Unpin) -> io::Result<usize> {
prefixed_and_suffixed_data_to_write(prefix, data, &[], out).await
}

/// Write a `text` message to `out`, which is assured to end in a newline.
pub async fn text_to_write(text: &[u8], out: impl AsyncWrite + Unpin) -> io::Result<usize> {
prefixed_and_suffixed_data_to_write(&[], text, &[b'\n'], out).await
}

/// Write a `data` message to `out`.
pub async fn data_to_write(data: &[u8], out: impl AsyncWrite + Unpin) -> io::Result<usize> {
prefixed_data_to_write(&[], data, out).await
}

/// Write an error `message` to `out`.
pub async fn error_to_write(message: &[u8], out: impl AsyncWrite + Unpin) -> io::Result<usize> {
prefixed_data_to_write(ERR_PREFIX, message, out).await
}

/// Write a response-end message to `out`.
pub async fn response_end_to_write(mut out: impl AsyncWrite + Unpin) -> io::Result<usize> {
out.write_all(RESPONSE_END_LINE).await?;
Ok(4)
}

/// Write a delim message to `out`.
pub async fn delim_to_write(mut out: impl AsyncWrite + Unpin) -> io::Result<usize> {
out.write_all(DELIMITER_LINE).await?;
Ok(4)
}

/// Write a flush message to `out`.
pub async fn flush_to_write(mut out: impl AsyncWrite + Unpin) -> io::Result<usize> {
out.write_all(FLUSH_LINE).await?;
Ok(4)
}

/// Write `data` of `kind` to `out` using side-band encoding.
pub async fn band_to_write(kind: Channel, data: &[u8], out: impl AsyncWrite + Unpin) -> io::Result<usize> {
prefixed_data_to_write(&[kind as u8], data, out).await
}
Loading

0 comments on commit a74573b

Please sign in to comment.