Sayan Nandan
3 years ago
committed by
GitHub
commit
246b974bd6
@ -1,341 +0,0 @@
|
||||
/*
|
||||
* Created on Tue Aug 04 2020
|
||||
*
|
||||
* This file is a part of Skytable
|
||||
* Skytable (formerly known as TerrabaseDB or Skybase) is a free and open-source
|
||||
* NoSQL database written by Sayan Nandan ("the Author") with the
|
||||
* vision to provide flexibility in data modelling without compromising
|
||||
* on performance, queryability or scalability.
|
||||
*
|
||||
* Copyright (c) 2020, Sayan Nandan <ohsayan@outlook.com>
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*
|
||||
*/
|
||||
|
||||
//! This module provides methods to deserialize an incoming response packet
|
||||
|
||||
use libsky::terrapipe::RespCodes;
|
||||
use libsky::util::terminal;
|
||||
use std::fmt;
|
||||
|
||||
#[derive(Debug, PartialEq)]
|
||||
/// A response datagroup
|
||||
///
|
||||
/// This contains all the elements returned by a certain action. So let's say you did
|
||||
/// something like `MGET x y`, then the values of x and y will be in a single datagroup.
|
||||
pub struct DataGroup(Vec<DataType>);
|
||||
|
||||
/// A data type as defined by the Terrapipe protocol
|
||||
///
|
||||
///
|
||||
/// Every variant stays in an `Option` for convenience while parsing. It's like we first
|
||||
/// create a `Variant(None)` variant. Then we read the data which corresponds to it, and then we
|
||||
/// replace `None` with the appropriate object. When we first detect the type, we use this as a way of matching
|
||||
/// avoiding duplication by writing another `DataType` enum
|
||||
#[derive(Debug, PartialEq)]
|
||||
#[non_exhaustive]
|
||||
pub enum DataType {
|
||||
/// A string value
|
||||
Str(Option<String>),
|
||||
/// A response code (it is kept as `String` for "other error" types)
|
||||
RespCode(Option<String>),
|
||||
/// An unsigned 64-bit integer, equivalent to an `u64`
|
||||
UnsignedInt(Option<Result<u64, std::num::ParseIntError>>),
|
||||
}
|
||||
|
||||
impl fmt::Display for DataGroup {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
for element in self.0.iter() {
|
||||
match element {
|
||||
DataType::Str(Some(val)) => write!(f, "\"{}\" ", val)?,
|
||||
DataType::Str(None) => (),
|
||||
DataType::UnsignedInt(Some(Ok(int))) => write!(f, "{}", int)?,
|
||||
DataType::UnsignedInt(Some(Err(_))) => terminal::write_error("[Parse Error]")?,
|
||||
DataType::UnsignedInt(None) => (),
|
||||
DataType::RespCode(Some(rc)) => {
|
||||
if rc.len() == 1 {
|
||||
if let Some(rcode) = RespCodes::from_str(&rc, None) {
|
||||
match rcode {
|
||||
RespCodes::Okay => terminal::write_info("(Okay) ")?,
|
||||
RespCodes::NotFound => terminal::write_info("(Nil) ")?,
|
||||
RespCodes::OverwriteError => {
|
||||
terminal::write_error("(Overwrite Error) ")?
|
||||
}
|
||||
RespCodes::ActionError => terminal::write_error("(Action Error) ")?,
|
||||
RespCodes::PacketError => terminal::write_error("(Packet Error) ")?,
|
||||
RespCodes::ServerError => terminal::write_error("(Server Error) ")?,
|
||||
RespCodes::OtherError(_) => {
|
||||
terminal::write_error("(Other Error) ")?
|
||||
}
|
||||
}
|
||||
}
|
||||
} else {
|
||||
terminal::write_error(format!("[ERROR: '{}'] ", rc))?;
|
||||
}
|
||||
}
|
||||
_ => unimplemented!(),
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
|
||||
/// Errors that may occur while parsing responses from the server
|
||||
///
|
||||
/// Every variant, except `Incomplete` has an `usize` field, which is used to advance the
|
||||
/// buffer
|
||||
#[derive(Debug, PartialEq)]
|
||||
pub enum ClientResult {
|
||||
/// The response was Invalid
|
||||
InvalidResponse,
|
||||
/// The response is a valid response and has been parsed into a vector of datagroups
|
||||
Response(Vec<DataGroup>, usize),
|
||||
/// The response was empty, which means that the remote end closed the connection
|
||||
Empty(usize),
|
||||
/// The response is incomplete
|
||||
Incomplete,
|
||||
}
|
||||
|
||||
impl fmt::Display for ClientResult {
|
||||
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
|
||||
use ClientResult::*;
|
||||
match self {
|
||||
InvalidResponse => write!(f, "ERROR: The server sent an invalid response"),
|
||||
Response(_, _) => unimplemented!(),
|
||||
Empty(_) => write!(f, ""),
|
||||
Incomplete => write!(f, "ERROR: The server sent an incomplete response"),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Parse a response packet
|
||||
pub fn parse(buf: &[u8]) -> ClientResult {
|
||||
if buf.len() < 6 {
|
||||
// A packet that has less than 6 characters? Nonsense!
|
||||
return ClientResult::Incomplete;
|
||||
}
|
||||
/*
|
||||
We first get the metaframe, which looks something like:
|
||||
```
|
||||
#<numchars_in_next_line>\n
|
||||
!<num_of_datagroups>\n
|
||||
```
|
||||
*/
|
||||
let mut pos = 0;
|
||||
if buf[pos] != b'#' {
|
||||
return ClientResult::InvalidResponse;
|
||||
} else {
|
||||
pos += 1;
|
||||
}
|
||||
let next_line = match read_line_and_return_next_line(&mut pos, &buf) {
|
||||
Some(line) => line,
|
||||
None => {
|
||||
// This is incomplete
|
||||
return ClientResult::Incomplete;
|
||||
}
|
||||
};
|
||||
pos += 1; // Skip LF
|
||||
// Find out the number of actions that we have to do
|
||||
let mut action_size = 0usize;
|
||||
if next_line[0] == b'*' {
|
||||
let mut line_iter = next_line.into_iter().skip(1).peekable();
|
||||
while let Some(dig) = line_iter.next() {
|
||||
let curdig: usize = match dig.checked_sub(48) {
|
||||
Some(dig) => {
|
||||
if dig > 9 {
|
||||
return ClientResult::InvalidResponse;
|
||||
} else {
|
||||
dig.into()
|
||||
}
|
||||
}
|
||||
None => return ClientResult::InvalidResponse,
|
||||
};
|
||||
action_size = (action_size * 10) + curdig;
|
||||
}
|
||||
// This line gives us the number of actions
|
||||
} else {
|
||||
return ClientResult::InvalidResponse;
|
||||
}
|
||||
let mut items: Vec<DataGroup> = Vec::with_capacity(action_size);
|
||||
while pos < buf.len() && items.len() <= action_size {
|
||||
match buf[pos] {
|
||||
b'#' => {
|
||||
pos += 1; // Skip '#'
|
||||
let next_line = match read_line_and_return_next_line(&mut pos, &buf) {
|
||||
Some(line) => line,
|
||||
None => {
|
||||
// This is incomplete
|
||||
return ClientResult::Incomplete;
|
||||
}
|
||||
}; // Now we have the current line
|
||||
pos += 1; // Skip the newline
|
||||
// Move the cursor ahead by the number of bytes that we just read
|
||||
// Let us check the current char
|
||||
match next_line[0] {
|
||||
b'&' => {
|
||||
// This is an array
|
||||
// Now let us parse the array size
|
||||
let mut current_array_size = 0usize;
|
||||
let mut linepos = 1; // Skip the '&' character
|
||||
while linepos < next_line.len() {
|
||||
let curdg: usize = match next_line[linepos].checked_sub(48) {
|
||||
Some(dig) => {
|
||||
if dig > 9 {
|
||||
// If `dig` is greater than 9, then the current
|
||||
// UTF-8 char isn't a number
|
||||
return ClientResult::InvalidResponse;
|
||||
} else {
|
||||
dig.into()
|
||||
}
|
||||
}
|
||||
None => return ClientResult::InvalidResponse,
|
||||
};
|
||||
current_array_size = (current_array_size * 10) + curdg; // Increment the size
|
||||
linepos += 1; // Move the position ahead, since we just read another char
|
||||
}
|
||||
// Now we know the array size, good!
|
||||
let mut actiongroup: Vec<DataType> = Vec::with_capacity(current_array_size);
|
||||
// Let's loop over to get the elements till the size of this array
|
||||
while pos < buf.len() && actiongroup.len() < current_array_size {
|
||||
let mut element_size = 0usize;
|
||||
let datatype = match buf[pos] {
|
||||
b'+' => DataType::Str(None),
|
||||
b'!' => DataType::RespCode(None),
|
||||
b':' => DataType::UnsignedInt(None),
|
||||
x @ _ => unimplemented!("Type '{}' not implemented", char::from(x)),
|
||||
};
|
||||
pos += 1; // We've got the tsymbol above, so skip it
|
||||
while pos < buf.len() && buf[pos] != b'\n' {
|
||||
let curdig: usize = match buf[pos].checked_sub(48) {
|
||||
Some(dig) => {
|
||||
if dig > 9 {
|
||||
// If `dig` is greater than 9, then the current
|
||||
// UTF-8 char isn't a number
|
||||
return ClientResult::InvalidResponse;
|
||||
} else {
|
||||
dig.into()
|
||||
}
|
||||
}
|
||||
None => return ClientResult::InvalidResponse,
|
||||
};
|
||||
element_size = (element_size * 10) + curdig; // Increment the size
|
||||
pos += 1; // Move the position ahead, since we just read another char
|
||||
}
|
||||
pos += 1;
|
||||
// We now know the item size
|
||||
let mut value = String::with_capacity(element_size);
|
||||
let extracted = match buf.get(pos..pos + element_size) {
|
||||
Some(s) => s,
|
||||
None => return ClientResult::Incomplete,
|
||||
};
|
||||
pos += element_size; // Move the position ahead
|
||||
value.push_str(&String::from_utf8_lossy(extracted));
|
||||
pos += 1; // Skip the newline
|
||||
actiongroup.push(match datatype {
|
||||
DataType::Str(_) => DataType::Str(Some(value)),
|
||||
DataType::RespCode(_) => DataType::RespCode(Some(value)),
|
||||
DataType::UnsignedInt(_) => {
|
||||
DataType::UnsignedInt(Some(value.parse()))
|
||||
}
|
||||
});
|
||||
}
|
||||
items.push(DataGroup(actiongroup));
|
||||
}
|
||||
_ => return ClientResult::InvalidResponse,
|
||||
}
|
||||
continue;
|
||||
}
|
||||
_ => {
|
||||
// Since the variant '#' would does all the array
|
||||
// parsing business, we should never reach here unless
|
||||
// the packet is invalid
|
||||
return ClientResult::InvalidResponse;
|
||||
}
|
||||
}
|
||||
}
|
||||
if buf.get(pos).is_none() {
|
||||
// Either more data was sent or some data was missing
|
||||
if items.len() == action_size {
|
||||
if items.len() == 1 {
|
||||
ClientResult::Response(items, pos)
|
||||
} else {
|
||||
// The CLI does not support batch queries
|
||||
unimplemented!();
|
||||
}
|
||||
} else {
|
||||
ClientResult::Incomplete
|
||||
}
|
||||
} else {
|
||||
ClientResult::InvalidResponse
|
||||
}
|
||||
}
|
||||
/// Read a size line and return the following line
|
||||
///
|
||||
/// This reads a line that begins with the number, i.e make sure that
|
||||
/// the **`#` character is skipped**
|
||||
///
|
||||
fn read_line_and_return_next_line<'a>(pos: &mut usize, buf: &'a [u8]) -> Option<&'a [u8]> {
|
||||
let mut next_line_size = 0usize;
|
||||
while pos < &mut buf.len() && buf[*pos] != b'\n' {
|
||||
// 48 is the UTF-8 code for '0'
|
||||
let curdig: usize = match buf[*pos].checked_sub(48) {
|
||||
Some(dig) => {
|
||||
if dig > 9 {
|
||||
// If `dig` is greater than 9, then the current
|
||||
// UTF-8 char isn't a number
|
||||
return None;
|
||||
} else {
|
||||
dig.into()
|
||||
}
|
||||
}
|
||||
None => return None,
|
||||
};
|
||||
next_line_size = (next_line_size * 10) + curdig; // Increment the size
|
||||
*pos += 1; // Move the position ahead, since we just read another char
|
||||
}
|
||||
*pos += 1; // Skip the newline
|
||||
// We now know the size of the next line
|
||||
let next_line = match buf.get(*pos..*pos + next_line_size) {
|
||||
Some(line) => line,
|
||||
None => {
|
||||
// This is incomplete
|
||||
return None;
|
||||
}
|
||||
}; // Now we have the current line
|
||||
// Move the cursor ahead by the number of bytes that we just read
|
||||
*pos += next_line_size;
|
||||
Some(next_line)
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
#[test]
|
||||
fn test_parser() {
|
||||
let res = "#2\n*1\n#2\n&1\n+4\nHEY!\n".as_bytes().to_owned();
|
||||
assert_eq!(
|
||||
parse(&res),
|
||||
ClientResult::Response(
|
||||
vec![DataGroup(vec![DataType::Str(Some("HEY!".to_owned()))])],
|
||||
res.len()
|
||||
)
|
||||
);
|
||||
let res = "#2\n*1\n#2\n&1\n!1\n0\n".as_bytes().to_owned();
|
||||
assert_eq!(
|
||||
parse(&res),
|
||||
ClientResult::Response(
|
||||
vec![DataGroup(vec![DataType::RespCode(Some("0".to_owned()))])],
|
||||
res.len()
|
||||
)
|
||||
);
|
||||
}
|
||||
@ -1,252 +0,0 @@
|
||||
/*
|
||||
* Created on Tue Aug 04 2020
|
||||
*
|
||||
* This file is a part of Skytable
|
||||
* Skytable (formerly known as TerrabaseDB or Skybase) is a free and open-source
|
||||
* NoSQL database written by Sayan Nandan ("the Author") with the
|
||||
* vision to provide flexibility in data modelling without compromising
|
||||
* on performance, queryability or scalability.
|
||||
*
|
||||
* Copyright (c) 2020, Sayan Nandan <ohsayan@outlook.com>
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*
|
||||
*/
|
||||
|
||||
mod deserializer;
|
||||
use bytes::{Buf, BytesMut};
|
||||
use deserializer::ClientResult;
|
||||
use libsky::terrapipe;
|
||||
use libsky::TResult;
|
||||
use libsky::BUF_CAP;
|
||||
use openssl::ssl::Ssl;
|
||||
use openssl::ssl::SslContext;
|
||||
use openssl::ssl::SslMethod;
|
||||
use std::io::Result as IoResult;
|
||||
use std::net::SocketAddr;
|
||||
use std::pin::Pin;
|
||||
use tokio::io::{AsyncReadExt, AsyncWriteExt};
|
||||
use tokio::net::TcpStream;
|
||||
use tokio_openssl::SslStream;
|
||||
|
||||
pub enum Con {
|
||||
Secure(SslConnection),
|
||||
Insecure(Connection),
|
||||
}
|
||||
|
||||
impl Con {
|
||||
pub async fn execute_query(&mut self, query: String) {
|
||||
match self {
|
||||
Con::Insecure(con) => con.run_query(query).await,
|
||||
Con::Secure(con) => con.run_query(query).await,
|
||||
}
|
||||
}
|
||||
pub async fn shutdown(&mut self) -> TResult<()> {
|
||||
match self {
|
||||
Con::Insecure(con) => con.shutdown().await,
|
||||
Con::Secure(con) => con.shutdown().await,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// A `Connection` is a wrapper around a`TcpStream` and a read buffer
|
||||
pub struct Connection {
|
||||
stream: TcpStream,
|
||||
buffer: BytesMut,
|
||||
}
|
||||
|
||||
impl Connection {
|
||||
/// Create a new connection, creating a connection to `host`
|
||||
pub async fn new(host: &str) -> TResult<Self> {
|
||||
let stream = TcpStream::connect(host).await?;
|
||||
println!("Connected to tp://{}", host);
|
||||
Ok(Connection {
|
||||
stream,
|
||||
buffer: BytesMut::with_capacity(BUF_CAP),
|
||||
})
|
||||
}
|
||||
/// This function will write a query to the stream and read the response from the
|
||||
/// server. It will then determine if the returned response is complete or incomplete
|
||||
/// or invalid.
|
||||
///
|
||||
/// - If it is complete, then the return is parsed into a `Display`able form
|
||||
/// and written to the output stream. If any parsing errors occur, they're also handled
|
||||
/// by this function (usually, "Invalid Response" is written to the terminal).
|
||||
/// - If the packet is incomplete, it will wait to read the entire response from the stream
|
||||
/// - If the packet is corrupted, it will output "Invalid Response"
|
||||
pub async fn run_query(&mut self, query: String) {
|
||||
let query = terrapipe::proc_query(query);
|
||||
match self.stream.write_all(&query).await {
|
||||
Ok(_) => (),
|
||||
Err(_) => {
|
||||
eprintln!("ERROR: Couldn't write data to socket");
|
||||
return;
|
||||
}
|
||||
};
|
||||
loop {
|
||||
match self.stream.read_buf(&mut self.buffer).await {
|
||||
Ok(_) => (),
|
||||
Err(e) => {
|
||||
eprintln!("ERROR: {}", e);
|
||||
return;
|
||||
}
|
||||
}
|
||||
match self.try_response().await {
|
||||
ClientResult::Empty(f) => {
|
||||
self.buffer.advance(f);
|
||||
eprintln!("ERROR: The remote end reset the connection");
|
||||
return;
|
||||
}
|
||||
ClientResult::Incomplete => {
|
||||
continue;
|
||||
}
|
||||
ClientResult::Response(r, f) => {
|
||||
self.buffer.advance(f);
|
||||
if r.len() == 0 {
|
||||
return;
|
||||
}
|
||||
for group in r {
|
||||
println!("{}", group);
|
||||
}
|
||||
return;
|
||||
}
|
||||
ClientResult::InvalidResponse => {
|
||||
self.buffer.clear();
|
||||
eprintln!("{}", ClientResult::InvalidResponse);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
/// This function is a subroutine of `run_query` used to parse the response packet
|
||||
async fn try_response(&mut self) -> ClientResult {
|
||||
if self.buffer.is_empty() {
|
||||
// The connection was possibly reset
|
||||
return ClientResult::Empty(0);
|
||||
}
|
||||
deserializer::parse(&self.buffer)
|
||||
}
|
||||
pub async fn shutdown(&mut self) -> TResult<()> {
|
||||
self.stream.shutdown().await.map_err(|e| e.into())
|
||||
}
|
||||
}
|
||||
|
||||
/// An `SslConnection` is a wrapper around a `SslStream<TcpStream>` provided by OpenSSL and a
|
||||
/// read buffer
|
||||
pub struct SslConnection {
|
||||
stream: SslStream<TcpStream>,
|
||||
buffer: BytesMut,
|
||||
}
|
||||
|
||||
impl SslConnection {
|
||||
/// Create a new connection, creating a connection to `host`
|
||||
pub async fn new(host: &str, sslcert: &str) -> TResult<Self> {
|
||||
let mut ctx = SslContext::builder(SslMethod::tls_client())?;
|
||||
ctx.set_ca_file(sslcert)?;
|
||||
let ssl = Ssl::new(&ctx.build())?;
|
||||
let stream = TcpStream::connect(host).await?;
|
||||
let mut stream = SslStream::new(ssl, stream)?;
|
||||
Pin::new(&mut stream).connect().await.unwrap();
|
||||
println!("Connected to tps://{}", host);
|
||||
Ok(SslConnection {
|
||||
stream,
|
||||
buffer: BytesMut::with_capacity(BUF_CAP),
|
||||
})
|
||||
}
|
||||
/// This function will write a query to the stream and read the response from the
|
||||
/// server. It will then determine if the returned response is complete or incomplete
|
||||
/// or invalid.
|
||||
///
|
||||
/// - If it is complete, then the return is parsed into a `Display`able form
|
||||
/// and written to the output stream. If any parsing errors occur, they're also handled
|
||||
/// by this function (usually, "Invalid Response" is written to the terminal).
|
||||
/// - If the packet is incomplete, it will wait to read the entire response from the stream
|
||||
/// - If the packet is corrupted, it will output "Invalid Response"
|
||||
pub async fn run_query(&mut self, query: String) {
|
||||
let query = terrapipe::proc_query(query);
|
||||
match self.stream.write_all(&query).await {
|
||||
Ok(_) => (),
|
||||
Err(e) => {
|
||||
eprintln!("ERROR: Couldn't write data to socket with '{}'", e);
|
||||
return;
|
||||
}
|
||||
};
|
||||
loop {
|
||||
if let Err(e) = self.read_again().await {
|
||||
eprintln!("ERROR: Reading from stream failed with: '{}'", e);
|
||||
return;
|
||||
}
|
||||
match self.try_response().await {
|
||||
ClientResult::Empty(f) => {
|
||||
self.buffer.advance(f);
|
||||
eprintln!("ERROR: The remote end reset the connection");
|
||||
return;
|
||||
}
|
||||
ClientResult::Incomplete => {
|
||||
continue;
|
||||
}
|
||||
ClientResult::Response(r, f) => {
|
||||
self.buffer.advance(f);
|
||||
if r.len() == 0 {
|
||||
return;
|
||||
}
|
||||
for group in r {
|
||||
println!("{}", group);
|
||||
}
|
||||
return;
|
||||
}
|
||||
ClientResult::InvalidResponse => {
|
||||
self.buffer.clear();
|
||||
eprintln!("{}", ClientResult::InvalidResponse);
|
||||
return;
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
/// This function is a subroutine of `run_query` used to parse the response packet
|
||||
async fn try_response(&mut self) -> ClientResult {
|
||||
if self.buffer.is_empty() {
|
||||
// The connection was possibly reset
|
||||
return ClientResult::Empty(0);
|
||||
}
|
||||
deserializer::parse(&self.buffer)
|
||||
}
|
||||
async fn read_again(&mut self) -> Result<(), String> {
|
||||
match self.stream.read_buf(&mut self.buffer).await {
|
||||
Ok(0) => {
|
||||
if self.buffer.is_empty() {
|
||||
return Ok(());
|
||||
} else {
|
||||
return Err(format!(
|
||||
"Connection reset while reading from {}",
|
||||
if let Ok(p) = self.get_peer() {
|
||||
p.to_string()
|
||||
} else {
|
||||
"peer".to_owned()
|
||||
}
|
||||
)
|
||||
.into());
|
||||
}
|
||||
}
|
||||
Ok(_) => Ok(()),
|
||||
Err(e) => return Err(format!("{}", e)),
|
||||
}
|
||||
}
|
||||
fn get_peer(&self) -> IoResult<SocketAddr> {
|
||||
self.stream.get_ref().peer_addr()
|
||||
}
|
||||
pub async fn shutdown(&mut self) -> TResult<()> {
|
||||
self.stream.shutdown().await.map_err(|e| e.into())
|
||||
}
|
||||
}
|
||||
@ -0,0 +1,167 @@
|
||||
/*
|
||||
* Created on Wed May 12 2021
|
||||
*
|
||||
* This file is a part of Skytable
|
||||
* Skytable (formerly known as TerrabaseDB or Skybase) is a free and open-source
|
||||
* NoSQL database written by Sayan Nandan ("the Author") with the
|
||||
* vision to provide flexibility in data modelling without compromising
|
||||
* on performance, queryability or scalability.
|
||||
*
|
||||
* Copyright (c) 2021, Sayan Nandan <ohsayan@outlook.com>
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*
|
||||
*/
|
||||
use crossterm::style::{Color, Print, ResetColor, SetForegroundColor};
|
||||
use skytable::{AsyncConnection, Element, RespCode, Response};
|
||||
|
||||
pub struct Runner {
|
||||
con: AsyncConnection,
|
||||
}
|
||||
|
||||
macro_rules! write_string {
|
||||
($st:ident) => {
|
||||
println!("\"{}\"", $st);
|
||||
};
|
||||
($idx:ident, $st:ident) => {
|
||||
println!("({}) \"{}\"", $idx, $st);
|
||||
};
|
||||
}
|
||||
|
||||
macro_rules! write_int {
|
||||
($int:ident) => {
|
||||
println!("{}", $int);
|
||||
};
|
||||
($idx:ident, $st:ident) => {
|
||||
println!("({}) \"{}\"", $idx, $st);
|
||||
};
|
||||
}
|
||||
|
||||
macro_rules! write_err {
|
||||
($idx:expr, $err:ident) => {
|
||||
crossterm::execute!(
|
||||
std::io::stdout(),
|
||||
SetForegroundColor(Color::Red),
|
||||
Print(if let Some(idx) = $idx {
|
||||
format!("({}) ({})\n", idx, $err)
|
||||
} else {
|
||||
format!("({})\n", $err)
|
||||
}),
|
||||
ResetColor
|
||||
)
|
||||
.expect("Failed to write to stdout");
|
||||
};
|
||||
($idx:ident, $err:literal) => {
|
||||
crossterm::execute!(
|
||||
std::io::stdout(),
|
||||
SetForegroundColor(Color::Red),
|
||||
Print(
|
||||
(if let Some(idx) = $idx {
|
||||
format!("({}) ({})\n", idx, $err)
|
||||
} else {
|
||||
format!("({})\n", $err)
|
||||
})
|
||||
),
|
||||
ResetColor
|
||||
)
|
||||
.expect("Failed to write to stdout");
|
||||
};
|
||||
}
|
||||
|
||||
macro_rules! write_okay {
|
||||
() => {
|
||||
crossterm::execute!(
|
||||
std::io::stdout(),
|
||||
SetForegroundColor(Color::Cyan),
|
||||
Print("(Okay)\n".to_string()),
|
||||
ResetColor
|
||||
)
|
||||
.expect("Failed to write to stdout");
|
||||
};
|
||||
($idx:ident) => {
|
||||
crossterm::execute!(
|
||||
std::io::stdout(),
|
||||
SetForegroundColor(Color::Cyan),
|
||||
Print(format!("({}) (Okay)\n", $idx)),
|
||||
ResetColor
|
||||
)
|
||||
.expect("Failed to write to stdout");
|
||||
};
|
||||
}
|
||||
|
||||
impl Runner {
|
||||
pub const fn new(con: AsyncConnection) -> Self {
|
||||
Runner { con }
|
||||
}
|
||||
pub async fn run_query(&mut self, unescaped_items: &str) {
|
||||
let query = libsky::turn_into_query(unescaped_items);
|
||||
match self.con.run_simple_query(query).await {
|
||||
Ok(resp) => match resp {
|
||||
Response::InvalidResponse => {
|
||||
println!("ERROR: The server sent an invalid response");
|
||||
}
|
||||
Response::Item(element) => match element {
|
||||
Element::String(st) => write_string!(st),
|
||||
Element::FlatArray(arr) => print_flat_array(arr),
|
||||
Element::RespCode(r) => print_rcode(r, None),
|
||||
Element::UnsignedInt(int) => write_int!(int),
|
||||
Element::Array(a) => print_array(a),
|
||||
_ => unimplemented!(),
|
||||
},
|
||||
Response::ParseError => {
|
||||
println!("ERROR: The client failed to deserialize data sent by the server")
|
||||
}
|
||||
x @ _ => {
|
||||
println!(
|
||||
"The server possibly sent a newer data type that we can't parse: {:?}",
|
||||
x
|
||||
)
|
||||
}
|
||||
},
|
||||
Err(e) => {
|
||||
eprintln!("An I/O error occurred while querying: {}", e);
|
||||
std::process::exit(1);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn print_rcode(rcode: RespCode, idx: Option<usize>) {
|
||||
match rcode {
|
||||
RespCode::Okay => write_okay!(),
|
||||
RespCode::ActionError => write_err!(idx, "Action Error"),
|
||||
RespCode::ErrorString(st) => write_err!(idx, st),
|
||||
RespCode::OtherError => write_err!(idx, "Other Error"),
|
||||
RespCode::NotFound => write_err!(idx, "Not Found"),
|
||||
RespCode::OverwriteError => write_err!(idx, "Overwrite Error"),
|
||||
RespCode::PacketError => write_err!(idx, "Packet Error"),
|
||||
RespCode::ServerError => write_err!(idx, "Server Error"),
|
||||
}
|
||||
}
|
||||
|
||||
fn print_flat_array(flat_array: Vec<String>) {
|
||||
flat_array.into_iter().for_each(|item| write_string!(item))
|
||||
}
|
||||
fn print_array(array: Vec<Element>) {
|
||||
for (idx, item) in array.into_iter().enumerate() {
|
||||
let idx = idx + 1;
|
||||
match item {
|
||||
Element::String(st) => write_string!(idx, st),
|
||||
Element::RespCode(rc) => print_rcode(rc, Some(idx)),
|
||||
Element::UnsignedInt(int) => write_int!(idx, int),
|
||||
Element::FlatArray(a) => print_flat_array(a),
|
||||
_ => unimplemented!("Nested arrays cannot be printed just yet"),
|
||||
}
|
||||
}
|
||||
}
|
||||
@ -1,177 +0,0 @@
|
||||
/*
|
||||
* Created on Sat Jul 18 2020
|
||||
*
|
||||
* This file is a part of Skytable
|
||||
* Skytable (formerly known as TerrabaseDB or Skybase) is a free and open-source
|
||||
* NoSQL database written by Sayan Nandan ("the Author") with the
|
||||
* vision to provide flexibility in data modelling without compromising
|
||||
* on performance, queryability or scalability.
|
||||
*
|
||||
* Copyright (c) 2020, Sayan Nandan <ohsayan@outlook.com>
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*
|
||||
*/
|
||||
|
||||
//! # The Terrapipe protocol
|
||||
//! This module implements primitives for the Terrapipe protocol
|
||||
//!
|
||||
|
||||
pub const ADDR: &'static str = "127.0.0.1";
|
||||
use std::str::FromStr;
|
||||
|
||||
lazy_static::lazy_static! {
|
||||
static ref RE: regex::Regex = regex::Regex::from_str(r#"("[^"]*"|'[^']*'|[\S]+)+"#).unwrap();
|
||||
}
|
||||
|
||||
/// Response codes returned by the server
|
||||
#[derive(Debug, PartialEq)]
|
||||
pub enum RespCodes {
|
||||
/// `0`: Okay (Empty Response) - use the `ResponseBuilder` for building
|
||||
/// responses that contain data
|
||||
Okay,
|
||||
/// `1`: Not Found
|
||||
NotFound,
|
||||
/// `2`: Overwrite Error
|
||||
OverwriteError,
|
||||
/// `3`: Action Error
|
||||
ActionError,
|
||||
/// `4`: Packet Error
|
||||
PacketError,
|
||||
/// `5`: Server Error
|
||||
ServerError,
|
||||
/// `6`: Some other error - the wrapped `String` will be returned in the response body.
|
||||
/// Just a note, this gets quite messy, especially when we're using it for deconding responses
|
||||
OtherError(Option<String>),
|
||||
}
|
||||
|
||||
impl From<RespCodes> for u8 {
|
||||
fn from(rcode: RespCodes) -> u8 {
|
||||
use RespCodes::*;
|
||||
match rcode {
|
||||
Okay => 0,
|
||||
NotFound => 1,
|
||||
OverwriteError => 2,
|
||||
ActionError => 3,
|
||||
PacketError => 4,
|
||||
ServerError => 5,
|
||||
OtherError(_) => 6,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl From<RespCodes> for char {
|
||||
fn from(rcode: RespCodes) -> char {
|
||||
use RespCodes::*;
|
||||
match rcode {
|
||||
Okay => '0',
|
||||
NotFound => '1',
|
||||
OverwriteError => '2',
|
||||
ActionError => '3',
|
||||
PacketError => '4',
|
||||
ServerError => '5',
|
||||
OtherError(_) => '6',
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl RespCodes {
|
||||
pub fn from_str(val: &str, extra: Option<String>) -> Option<Self> {
|
||||
use RespCodes::*;
|
||||
let res = match val.parse::<u8>() {
|
||||
Ok(val) => match val {
|
||||
0 => Okay,
|
||||
1 => NotFound,
|
||||
2 => OverwriteError,
|
||||
3 => ActionError,
|
||||
4 => PacketError,
|
||||
5 => ServerError,
|
||||
6 => OtherError(extra),
|
||||
_ => return None,
|
||||
},
|
||||
Err(_) => return None,
|
||||
};
|
||||
Some(res)
|
||||
}
|
||||
pub fn from_u8(val: u8, extra: Option<String>) -> Option<Self> {
|
||||
use RespCodes::*;
|
||||
let res = match val {
|
||||
0 => Okay,
|
||||
1 => NotFound,
|
||||
2 => OverwriteError,
|
||||
3 => ActionError,
|
||||
4 => PacketError,
|
||||
5 => ServerError,
|
||||
6 => OtherError(extra),
|
||||
_ => return None,
|
||||
};
|
||||
Some(res)
|
||||
}
|
||||
pub fn from_utf8(val: u8) -> Option<Self> {
|
||||
let result = match val.checked_sub(48) {
|
||||
Some(r) => r,
|
||||
None => return None,
|
||||
};
|
||||
if result > 6 {
|
||||
return None;
|
||||
}
|
||||
return RespCodes::from_u8(result, None);
|
||||
}
|
||||
}
|
||||
|
||||
/// Prepare a query packet from a string of whitespace separated values
|
||||
pub fn proc_query<T>(querystr: T) -> Vec<u8>
|
||||
where
|
||||
T: AsRef<str>,
|
||||
{
|
||||
let mut bytes = Vec::with_capacity(querystr.as_ref().len());
|
||||
let args: Vec<String> = RE
|
||||
.find_iter(&querystr.as_ref())
|
||||
.map(|val| val.as_str().replace("'", "").replace("\"", "").to_owned())
|
||||
.collect();
|
||||
bytes.extend(b"#2\n*1\n#");
|
||||
let arg_len_bytes = args.len().to_string().into_bytes();
|
||||
let arg_len_bytes_len = (arg_len_bytes.len() + 1).to_string().into_bytes();
|
||||
bytes.extend(arg_len_bytes_len);
|
||||
bytes.extend(b"\n&");
|
||||
bytes.extend(arg_len_bytes);
|
||||
bytes.push(b'\n');
|
||||
args.into_iter().for_each(|arg| {
|
||||
bytes.push(b'#');
|
||||
let len_bytes = arg.len().to_string().into_bytes();
|
||||
bytes.extend(len_bytes);
|
||||
bytes.push(b'\n');
|
||||
bytes.extend(arg.as_bytes());
|
||||
bytes.push(b'\n');
|
||||
});
|
||||
bytes
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_queryproc() {
|
||||
let query = "GET x y".to_owned();
|
||||
assert_eq!(
|
||||
"#2\n*1\n#2\n&3\n#3\nGET\n#1\nx\n#1\ny\n"
|
||||
.as_bytes()
|
||||
.to_owned(),
|
||||
proc_query(query)
|
||||
);
|
||||
let q_escaped = proc_query(r#"SET X 'sayan with spaces'"#);
|
||||
assert_eq!(
|
||||
"#2\n*1\n#2\n&3\n#3\nSET\n#1\nX\n#17\nsayan with spaces\n"
|
||||
.as_bytes()
|
||||
.to_owned(),
|
||||
q_escaped
|
||||
);
|
||||
}
|
||||
@ -0,0 +1,67 @@
|
||||
/*
|
||||
* Created on Tue May 11 2021
|
||||
*
|
||||
* This file is a part of Skytable
|
||||
* Skytable (formerly known as TerrabaseDB or Skybase) is a free and open-source
|
||||
* NoSQL database written by Sayan Nandan ("the Author") with the
|
||||
* vision to provide flexibility in data modelling without compromising
|
||||
* on performance, queryability or scalability.
|
||||
*
|
||||
* Copyright (c) 2021, Sayan Nandan <ohsayan@outlook.com>
|
||||
*
|
||||
* This program is free software: you can redistribute it and/or modify
|
||||
* it under the terms of the GNU Affero General Public License as published by
|
||||
* the Free Software Foundation, either version 3 of the License, or
|
||||
* (at your option) any later version.
|
||||
*
|
||||
* This program is distributed in the hope that it will be useful,
|
||||
* but WITHOUT ANY WARRANTY; without even the implied warranty of
|
||||
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
||||
* GNU Affero General Public License for more details.
|
||||
*
|
||||
* You should have received a copy of the GNU Affero General Public License
|
||||
* along with this program. If not, see <https://www.gnu.org/licenses/>.
|
||||
*
|
||||
*/
|
||||
|
||||
use std::borrow::Cow;
|
||||
|
||||
#[derive(Debug, PartialEq)]
|
||||
#[non_exhaustive]
|
||||
/// # Data Types
|
||||
///
|
||||
/// This enum represents the data types supported by the Skyhash Protocol
|
||||
pub enum Element {
|
||||
/// Arrays can be nested! Their `<tsymbol>` is `&`
|
||||
Array(Vec<Element>),
|
||||
/// A String value; `<tsymbol>` is `+`
|
||||
String(String),
|
||||
/// An unsigned integer value; `<tsymbol>` is `:`
|
||||
UnsignedInt(u64),
|
||||
/// A non-recursive String array; tsymbol: `_`
|
||||
FlatArray(Vec<String>),
|
||||
}
|
||||
|
||||
impl Element {
|
||||
/// This will return a reference to the first element in the element
|
||||
///
|
||||
/// If this element is a compound type, it will return a reference to the first element in the compound
|
||||
/// type
|
||||
pub fn get_first(&self) -> Option<Cow<String>> {
|
||||
match self {
|
||||
Self::Array(elem) => match elem.first() {
|
||||
Some(el) => match el {
|
||||
Element::String(st) => Some(Cow::Borrowed(&st)),
|
||||
_ => None,
|
||||
},
|
||||
None => None,
|
||||
},
|
||||
Self::FlatArray(elem) => match elem.first() {
|
||||
Some(el) => Some(Cow::Borrowed(&el)),
|
||||
None => None,
|
||||
},
|
||||
Self::String(ref st) => Some(Cow::Borrowed(&st)),
|
||||
_ => None,
|
||||
}
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
File diff suppressed because it is too large
Load Diff
Loading…
Reference in New Issue