Showing the changes to git-protocol/src (1 of 4 files). Show all
Unified · Split
| @@ -286,6 +286,136 @@ mod blocking_io { · 285 unmodified lines | ||
| 286 | 286 | } |
| 287 | 287 | } |
| 288 | 288 | |
| 289 | +#[cfg(all(not(feature = "blocking-client"), feature = "async-client"))] | |
| 290 | +mod async_io { | |
| 291 | + use crate::fetch::{ | |
| 292 | + response, | |
| 293 | + response::{Acknowledgement, ShallowUpdate}, | |
| 294 | + Response, | |
| 295 | + }; | |
| 296 | + use futures_lite::AsyncBufReadExt; | |
| 297 | + use git_transport::{client, Protocol}; | |
| 298 | + use std::io; | |
| 299 | + | |
| 300 | + async fn parse_v2_section<T>( | |
| 301 | + line: &mut String, | |
| 302 | + reader: &mut (impl client::ExtendedBufRead + Unpin), | |
| 303 | + res: &mut Vec<T>, | |
| 304 | + parse: impl Fn(&str) -> Result<T, response::Error>, | |
| 305 | + ) -> Result<bool, response::Error> { | |
| 306 | + line.clear(); | |
| 307 | + while reader.read_line(line).await? != 0 { | |
| 308 | + res.push(parse(line)?); | |
| 309 | + line.clear(); | |
| 310 | + } | |
| 311 | + // End of message, or end of section? | |
| 312 | + Ok(if reader.stopped_at() == Some(client::MessageKind::Delimiter) { | |
| 313 | + // try reading more sections | |
| 314 | + reader.reset(Protocol::V2); | |
| 315 | + false | |
| 316 | + } else { | |
| 317 | + // we are done, there is no pack | |
| 318 | + true | |
| 319 | + }) | |
| 320 | + } | |
| 321 | + | |
| 322 | + impl Response { | |
| 323 | + /// Parse a response of the given `version` of the protocol from `reader`. | |
| 324 | + pub async fn from_line_reader( | |
| 325 | + version: Protocol, | |
| 326 | + reader: &mut (impl client::ExtendedBufRead + Unpin), | |
| 327 | + ) -> Result<Response, response::Error> { | |
| 328 | + match version { | |
| 329 | + Protocol::V1 => { | |
| 330 | + let mut line = String::new(); | |
| 331 | + let mut acks = Vec::<Acknowledgement>::new(); | |
| 332 | + let mut shallows = Vec::<ShallowUpdate>::new(); | |
| 333 | + let has_pack = 'lines: loop { | |
| 334 | + line.clear(); | |
| 335 | + let peeked_line = match reader.peek_data_line().await { | |
| 336 | + Some(Ok(Ok(line))) => String::from_utf8_lossy(line), | |
| 337 | + // This special case (hang/block forver) deals with a single NAK being a legitimate EOF sometimes | |
| 338 | + // Note that this might block forever in stateful connections as there it's not really clear | |
| 339 | + // if something will be following or not by just looking at the response. Instead you have to know | |
| 340 | + // the arguments sent to the server and count response lines based on intricate knowledge on how the | |
| 341 | + // server works. | |
| 342 | + // For now this is acceptable, as V2 can be used as a workaround, which also is the default. | |
| 343 | + Some(Err(err)) if err.kind() == io::ErrorKind::UnexpectedEof => break 'lines false, | |
| 344 | + Some(Err(err)) => return Err(err.into()), | |
| 345 | + Some(Ok(Err(err))) => return Err(err.into()), | |
| 346 | + None => { | |
| 347 | + // maybe we saw a shallow flush packet, let's reset and retry | |
| 348 | + debug_assert_eq!( | |
| 349 | + reader.stopped_at(), | |
| 350 | + Some(client::MessageKind::Flush), | |
| 351 | + "If this isn't a flush packet, we don't know what's going on" | |
| 352 | + ); | |
| 353 | + reader.read_line(&mut line).await?; | |
| 354 | + reader.reset(Protocol::V1); | |
| 355 | + match reader.peek_data_line().await { | |
| 356 | + Some(Ok(Ok(line))) => String::from_utf8_lossy(line), | |
| 357 | + Some(Err(err)) => return Err(err.into()), | |
| 358 | + Some(Ok(Err(err))) => return Err(err.into()), | |
| 359 | + None => break 'lines false, // EOF | |
| 360 | + } | |
| 361 | + } | |
| 362 | + }; | |
| 363 | + | |
| 364 | + if Response::parse_v1_ack_or_shallow_or_assume_pack(&mut acks, &mut shallows, &peeked_line) { | |
| 365 | + break 'lines true; | |
| 366 | + } | |
| 367 | + assert_ne!(reader.read_line(&mut line).await?, 0, "consuming a peeked line works"); | |
| 368 | + }; | |
| 369 | + Ok(Response { | |
| 370 | + acks, | |
| 371 | + shallows, | |
| 372 | + has_pack, | |
| 373 | + }) | |
| 374 | + } | |
| 375 | + Protocol::V2 => { | |
| 376 | + // NOTE: We only read acknowledgements and scrub to the pack file, until we have use for the other features | |
| 377 | + let mut line = String::new(); | |
| 378 | + reader.reset(Protocol::V2); | |
| 379 | + let mut acks = Vec::<Acknowledgement>::new(); | |
| 380 | + let mut shallows = Vec::<ShallowUpdate>::new(); | |
| 381 | + let has_pack = 'section: loop { | |
| 382 | + line.clear(); | |
| 383 | + if reader.read_line(&mut line).await? == 0 { | |
| 384 | + return Err(response::Error::Io(io::Error::new( | |
| 385 | + io::ErrorKind::UnexpectedEof, | |
| 386 | + "Could not read message headline", | |
| 387 | + ))); | |
| 388 | + }; | |
| 389 | + | |
| 390 | + match line.trim_end() { | |
| 391 | + "acknowledgments" => { | |
| 392 | + if parse_v2_section(&mut line, reader, &mut acks, Acknowledgement::from_line).await? { | |
| 393 | + break 'section false; | |
| 394 | + } | |
| 395 | + } | |
| 396 | + "shallow-info" => { | |
| 397 | + if parse_v2_section(&mut line, reader, &mut shallows, ShallowUpdate::from_line).await? { | |
| 398 | + break 'section false; | |
| 399 | + } | |
| 400 | + } | |
| 401 | + "packfile" => { | |
| 402 | + // what follows is the packfile itself, which can be read with a sideband enabled reader | |
| 403 | + break 'section true; | |
| 404 | + } | |
| 405 | + _ => return Err(response::Error::UnknownSectionHeader(line)), | |
| 406 | + } | |
| 407 | + }; | |
| 408 | + Ok(Response { | |
| 409 | + acks, | |
| 410 | + shallows, | |
| 411 | + has_pack, | |
| 412 | + }) | |
| 413 | + } | |
| 414 | + } | |
| 415 | + } | |
| 416 | + } | |
| 417 | +} | |
| 418 | + | |
| 289 | 419 | #[cfg(any(feature = "async-client", feature = "blocking-client"))] |
| 290 | 420 | impl Response { |
| 291 | 421 | /// with a friendly server, we just assume that a non-ack line is a pack line |