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
use crate::deserializer::{ParseError, Parser, RawResponse};
use crate::{Query, Response};
pub use std::io::Result as IoResult;
use std::io::{Error, ErrorKind, Read, Write};
use std::net::TcpStream;
const BUF_CAP: usize = 4096;
#[derive(Debug)]
#[cfg_attr(docsrs, doc(cfg(feature = "sync")))]
pub struct Connection {
stream: TcpStream,
buffer: Vec<u8>,
}
impl Connection {
pub fn new(host: &str, port: u16) -> IoResult<Self> {
let stream = TcpStream::connect((host, port))?;
Ok(Connection {
stream: stream,
buffer: Vec::with_capacity(BUF_CAP),
})
}
pub fn run_simple_query(&mut self, query: &Query) -> IoResult<Response> {
assert!(query.__len() != 0, "A `Query` cannot be of zero length!");
query.write_query_to_sync(&mut self.stream)?;
self.stream.flush()?;
loop {
let mut buffer = [0u8; 1024];
match self.stream.read(&mut buffer) {
Ok(0) => return Err(Error::from(ErrorKind::ConnectionReset)),
Ok(read) => {
self.buffer.extend(&buffer[..read]);
}
Err(e) => return Err(e),
}
match self.try_response() {
Ok((query, forward_by)) => {
self.buffer.drain(..forward_by);
match query {
RawResponse::SimpleQuery(s) => return Ok(Response::Item(s)),
RawResponse::PipelinedQuery(_) => {
unimplemented!("Pipelined queries aren't implemented yet")
}
}
}
Err(e) => match e {
ParseError::NotEnough => (),
ParseError::BadPacket | ParseError::UnexpectedByte => {
self.buffer.clear();
return Ok(Response::InvalidResponse);
}
ParseError::DataTypeParseError => return Ok(Response::ParseError),
ParseError::Empty => return Err(Error::from(ErrorKind::ConnectionReset)),
ParseError::UnknownDatatype => return Ok(Response::UnsupportedDataType),
},
}
}
}
fn try_response(&mut self) -> Result<(RawResponse, usize), ParseError> {
if self.buffer.is_empty() {
return Err(ParseError::Empty);
}
Parser::new(&self.buffer).parse()
}
}
impl crate::actions::SyncSocket for crate::sync::Connection {
fn run(&mut self, q: Query) -> std::result::Result<Response, std::io::Error> {
self.run_simple_query(&q)
}
}