1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
|
use std::{
io,
ops::{Deref, DerefMut},
};
use async_trait::async_trait;
use futures_io::{AsyncBufRead, AsyncRead};
use gix_packetline::{read::ProgressAction, PacketLineRef};
use crate::{
client::{Error, MessageKind},
Protocol,
};
/// A function `f(is_error, text)` receiving progress or error information.
/// As it is not a future itself, it must not block. If IO is performed within the function, be sure to spawn
/// it onto an executor.
pub type HandleProgress = Box<dyn FnMut(bool, &[u8]) -> ProgressAction>;
/// This trait exists to get a version of a `gix_packetline::Provider` without type parameters,
/// but leave support for reading lines directly without forcing them through `String`.
///
/// For the sake of usability, it also implements [`std::io::BufRead`] making it trivial to
/// read pack files while keeping open the option to read individual lines with low overhead.
#[async_trait(?Send)]
pub trait ReadlineBufRead: AsyncBufRead {
/// Read a packet line into the internal buffer and return it.
///
/// Returns `None` if the end of iteration is reached because of one of the following:
///
/// * natural EOF
/// * ERR packet line encountered
/// * A `delimiter` packet line encountered
async fn readline(
&mut self,
) -> Option<io::Result<Result<gix_packetline::PacketLineRef<'_>, gix_packetline::decode::Error>>>;
/// Read a line similar to `BufRead::read_line()`, but assure it doesn't try to find newlines
/// which might concatenate multiple distinct packet lines.
///
/// Making this a trait method allows to handle differences between async and blocking.
async fn readline_str(&mut self, line: &mut String) -> io::Result<usize>;
}
/// Provide even more access to the underlying packet reader.
#[async_trait(?Send)]
pub trait ExtendedBufRead: ReadlineBufRead {
/// Set the handler to which progress will be delivered.
///
/// Note that this is only possible if packet lines are sent in side band mode.
fn set_progress_handler(&mut self, handle_progress: Option<HandleProgress>);
/// Peek the next data packet line. Maybe None if the next line is a packet we stop at, queryable using
/// [`stopped_at()`][ExtendedBufRead::stopped_at()].
async fn peek_data_line(&mut self) -> Option<io::Result<Result<&[u8], Error>>>;
/// Resets the reader to allow reading past a previous stop, and sets delimiters according to the
/// given protocol.
fn reset(&mut self, version: Protocol);
/// Return the kind of message at which the reader stopped.
fn stopped_at(&self) -> Option<MessageKind>;
}
#[async_trait(?Send)]
impl<'a, T: ReadlineBufRead + ?Sized + 'a + Unpin> ReadlineBufRead for Box<T> {
async fn readline(&mut self) -> Option<io::Result<Result<PacketLineRef<'_>, gix_packetline::decode::Error>>> {
self.deref_mut().readline().await
}
async fn readline_str(&mut self, line: &mut String) -> io::Result<usize> {
self.deref_mut().readline_str(line).await
}
}
#[async_trait(?Send)]
impl<'a, T: ExtendedBufRead + ?Sized + 'a + Unpin> ExtendedBufRead for Box<T> {
fn set_progress_handler(&mut self, handle_progress: Option<HandleProgress>) {
self.deref_mut().set_progress_handler(handle_progress)
}
async fn peek_data_line(&mut self) -> Option<io::Result<Result<&[u8], Error>>> {
self.deref_mut().peek_data_line().await
}
fn reset(&mut self, version: Protocol) {
self.deref_mut().reset(version)
}
fn stopped_at(&self) -> Option<MessageKind> {
self.deref().stopped_at()
}
}
#[async_trait(?Send)]
impl<T: AsyncRead + Unpin> ReadlineBufRead
for gix_packetline::read::WithSidebands<'_, T, for<'b> fn(bool, &'b [u8]) -> ProgressAction>
{
async fn readline(&mut self) -> Option<io::Result<Result<PacketLineRef<'_>, gix_packetline::decode::Error>>> {
self.read_data_line().await
}
async fn readline_str(&mut self, line: &mut String) -> io::Result<usize> {
self.read_line_to_string(line).await
}
}
#[async_trait(?Send)]
impl<'a, T: AsyncRead + Unpin> ReadlineBufRead for gix_packetline::read::WithSidebands<'a, T, HandleProgress> {
async fn readline(&mut self) -> Option<io::Result<Result<PacketLineRef<'_>, gix_packetline::decode::Error>>> {
self.read_data_line().await
}
async fn readline_str(&mut self, line: &mut String) -> io::Result<usize> {
self.read_line_to_string(line).await
}
}
#[async_trait(?Send)]
impl<'a, T: AsyncRead + Unpin> ExtendedBufRead for gix_packetline::read::WithSidebands<'a, T, HandleProgress> {
fn set_progress_handler(&mut self, handle_progress: Option<HandleProgress>) {
self.set_progress_handler(handle_progress)
}
async fn peek_data_line(&mut self) -> Option<io::Result<Result<&[u8], Error>>> {
match self.peek_data_line().await {
Some(Ok(Ok(line))) => Some(Ok(Ok(line))),
Some(Ok(Err(err))) => Some(Ok(Err(err.into()))),
Some(Err(err)) => Some(Err(err)),
None => None,
}
}
fn reset(&mut self, version: Protocol) {
match version {
Protocol::V1 => self.reset_with(&[gix_packetline::PacketLineRef::Flush]),
Protocol::V2 => self.reset_with(&[
gix_packetline::PacketLineRef::Delimiter,
gix_packetline::PacketLineRef::Flush,
]),
}
}
fn stopped_at(&self) -> Option<MessageKind> {
self.stopped_at().map(|l| match l {
gix_packetline::PacketLineRef::Flush => MessageKind::Flush,
gix_packetline::PacketLineRef::Delimiter => MessageKind::Delimiter,
gix_packetline::PacketLineRef::ResponseEnd => MessageKind::ResponseEnd,
gix_packetline::PacketLineRef::Data(_) => unreachable!("data cannot be a delimiter"),
})
}
}
|