Something went wrong. Try again.
Monorepo for Tangled
Something went wrong. Try again.
7.9 kB · 191 lines
Rust
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192use std::ops::ControlFlow;use std::sync::atomic::{AtomicBool, Ordering};
use futures_lite::{AsyncBufReadExt as _, AsyncWriteExt};use gix_features::progress::Progress;use gix_object::bstr::{BString, ByteSlice, ByteVec};use gix_transport::Service;use gix_transport::client::async_io::{ HandleProgress, ReadlineBufRead, SetServiceResponse, Transport,};use gix_transport::client::{MessageKind, WriteMode};
use crate::command::{encode_command_block, select_capabilities};use crate::{ Error, Options, ReceivePackAdvertisement, RefUpdateStatus, Update, advertisement, report,};
pub struct Session<'a, T> { transport: &'a mut T, advertised: ReceivePackAdvertisement, /// Value for `agent=<v>`, emitted only if the remote advertised `agent`. agent: Option<&'a str>,}
impl<'a, T> Session<'a, T>where T: Transport,{ /// Perform the `git-receive-pack` handshake over `transport` and capture what the remote said. pub async fn handshake(transport: &'a mut T, agent: Option<&'a str>) -> Result<Self, Error> { // Scoped because `SetServiceResponse` borrows `transport` mutably: everything it owns has // to become owned data before `Self` can take the transport back — rust-lang/rust#76149. let advertised = { let result = transport.handshake(Service::ReceivePack, &[]).await; let SetServiceResponse { actual_protocol, capabilities, refs, } = result?;
// `refs` is `None` under protocol v2, which has no push command at all; the // `report-status` check below then rejects the session, as v2 never advertises it. let mut lines = BString::default(); if let Some(mut refs) = refs { while let Some(line) = refs.readline().await { // Two layers: transport failure, then framing failure. let Some(line) = line??.as_bstr() else { // A non-data line: the flush ending the advertisement. break; }; lines.push_str(line); if lines.last() != Some(&b'\n') { lines.push(b'\n'); } } }
// let supported_versions: Vec<_> = transport.supported_protocol_versions().into(); // if !supported_versions.is_empty() && !supported_versions.contains(&actual_protocol) { // return Err(Error::TransportProtocolPolicyViolation { // actual_version: actual_protocol, // }); // }
let (remote_refs, advertised_haves) = advertisement::from_v1_lines(lines.as_bstr())?; ReceivePackAdvertisement { server_protocol_version: actual_protocol, capabilities, remote_refs, advertised_haves, } };
// TODO(boltless): support `report-status-v2` with v2 report parser. if !advertised.capabilities.contains("report-status") { return Err(Error::NoReportStatus); }
Ok(Session { transport, advertised, agent, }) }
pub fn advertised(&self) -> &ReceivePackAdvertisement { &self.advertised }
/// Perform one push operation. /// /// `pack` must be `None` for a deletion-only push, not an empty pack: `git-receive-pack` reads /// no pack after a delete-only command block, so an empty one is left unread on the wire. pub async fn push( &mut self, trace_packetlines: bool, updates: &[Update], pack: Option<&[u8]>, opts: Options<'_>, progress: &mut impl Progress, should_interrupt: &AtomicBool, ) -> Result<Vec<RefUpdateStatus>, Error> { let capabilities = select_capabilities(&self.advertised.capabilities, updates, &opts, self.agent)?; let commands = encode_command_block(updates, capabilities.as_bstr(), opts.push_options)?;
// `Binary` maps to `PostBodyDataKind::Unbounded`, which puts an HTTP backend into streaming // mode instead of buffering the whole packfile. `MessageKind::Flush` is inert: only // `into_read()` writes it, and we take `into_parts()` — our flush is already in `commands`. let request = self.transport .request(WriteMode::Binary, MessageKind::Flush, trace_packetlines)?;
// `RequestWriter`'s `AsyncWrite` frames every write as a packet line, even under `Binary`; // `into_parts()` gives the unwrapped writer. The commands are framed already, and the pack // must not be framed at all. let (mut writer, mut reader) = request.into_parts(); writer.write_all(&commands).await?;
if let Some(pack) = pack { // Chunked because cancellation is cooperative: one `write_all` of a large pack would // never look at the flag. for chunk in pack.chunks(64 * 1024) { if should_interrupt.load(Ordering::Relaxed) { return Err(Error::Interrupted); } writer.write_all(chunk).await?; } } writer.flush().await?;
// Before any read: on a streaming POST the two halves are interdependent, and getting this // wrong deadlocks rather than errors. drop(writer);
// Only with a side-band: once a handler is set `poll_fill_buf` runs every line through // `decode_band()`, so on an unmultiplexed stream the `u` of `unpack ok` reads as band 117. let sideband = capabilities.split_str(" ").any(|c| c == b"side-band-64k"); if sideband { let progress = &*progress; reader.set_progress_handler(Some(Box::new(move |is_err: bool, data: &[u8]| { // Bands 2 and 3 verbatim: turning `phase: 42% (7/16)` back into a progress bar is // `gix_protocol::RemoteProgress`'s job, which this crate does not depend on. let text = data.as_bstr().trim_end(); if !text.is_empty() { let text = text.to_str_lossy().into_owned(); if is_err { progress.fail(text); } else { progress.info(text); } } if should_interrupt.load(Ordering::Relaxed) { ControlFlow::Break(()) } else { ControlFlow::Continue(()) } }) as HandleProgress<'_>)); }
// Not `readline()`: on a `WithSidebands` reader it forwards to `read_data_line()`, which // skips all sideband handling. The demultiplexer lives in `poll_fill_buf`, so the report // has to come through the `AsyncBufRead` surface, which ends in an empty slice. let mut lines = BString::default(); loop { let line = reader.fill_buf().await?; if line.is_empty() { break; } let consumed = line.len(); lines.push_str(line); // One read is one report line only on the unmultiplexed stream: a side-band payload is // a pkt-line stream of its own, cut wherever the remote's band buffer ended. if !sideband && lines.last() != Some(&b'\n') { lines.push(b'\n'); } reader.consume(consumed);
if should_interrupt.load(Ordering::Relaxed) { return Err(Error::Interrupted); } }
let lines = if sideband { report::unframe(&lines)? } else { lines }; Ok(report::parse_report_v1(lines.as_bstr())?) }}