// skip that number of bytes + 2 (\r\n). skip(src, len + 2) } } b'*' => { letlen = get_decimal(src)?;
for_in0..len { Frame::check(src)?; }
Ok(()) } actual => Err(MiniRedisParseError::Parse(format!( "protocol error; invalid frame type byte `{}`", actual ))), } }
pubfnparse(src: &mut Cursor<&[u8]>) ->Result<Frame, MiniRedisParseError> { matchget_u8(src)? { b'+' => { // Read the line and convert it to `Vec<u8>` letline = get_line(src)?.to_vec();
// Convert the line to a String letstring = String::from_utf8(line)?;
Ok(Frame::Simple(string)) } b'-' => { // Read the line and convert it to `Vec<u8>` letline = get_line(src)?.to_vec();
// Convert the line to a String letstring = String::from_utf8(line)?;
/// Read a new-line terminated decimal fnget_decimal(src: &mut Cursor<&[u8]>) ->Result<u64, MiniRedisParseError> { use atoi::atoi;
letline = get_line(src)?;
atoi::<u64>(line).ok_or_else(|| { MiniRedisParseError::Parse("protocol error; invalid frame format to get decimal".into()) }) }
/// Find a line in a frame fnget_line<'a>(src: &mut Cursor<&'a [u8]>) ->Result<&'a [u8], MiniRedisParseError> { // Scan the bytes directly letstart = src.position() asusize; // Scan to the second to last byte letend = src.get_ref().len() - 1;
foriin start..end { if src.get_ref()[i] == b'\r' && src.get_ref()[i + 1] == b'\n' { // We found a line, update the position to be *after* the \n src.set_position((i + 2) asu64);
// Return the line returnOk(&src.get_ref()[start..i]); } }
Err(MiniRedisParseError::Incomplete) }
上面定义了几个辅助函数：
skip(src: &mut Cursor<&[u8]>, n: usize)：将当前 Cursor 前移 n 个字节；
/// Utility for parsing a command /// /// Commands are represented as array frames. Each entry in the frame is a /// "token". A `Parse` is initialized with the array frame and provides a /// cursor-like API. Each command struct includes a `parse_frame` method that /// uses a `Parse` to extract its fields. #[derive(Debug)] pub(crate) structParse { /// Array frame iterator. parts: vec::IntoIter<Frame>, }
implParse { /// Create a new `Parse` to parse the contents of `frame`. /// Returns `Err` if `frame` is not an array frame. pub(crate) fnnew(frame: Frame) ->Result<Parse, MiniRedisParseError> { letarray = match frame { Frame::Array(array) => array, frame => { returnErr(MiniRedisParseError::Parse(format!( "protocol error; expected array, got {:?}", frame ))) } };
Ok(Parse { parts: array.into_iter(), }) }
/// Return the next entry. Array frames are arrays of frames, so the next /// entry is a frame. fnnext(&mutself) ->Result<Frame, MiniRedisParseError> { self.parts.next().ok_or(MiniRedisParseError::EndOfStream) }
/// Return the next entry as a string. /// If the next entry cannot be represented as a String, then an error is returned. pub(crate) fnnext_string(&mutself) ->Result<String, MiniRedisParseError> { matchself.next()? { // Both `Simple` and `Bulk` representation may be strings. Strings // are parsed to UTF-8. // // While errors are stored as strings, they are considered separate // types. Frame::Simple(s) => Ok(s), Frame::Bulk(data) => std::str::from_utf8(&data[..]) .map(|s| s.to_string()) .map_err(|_| MiniRedisParseError::Parse("protocol error; invalid string".into())), frame => Err(MiniRedisParseError::Parse(format!( "protocol error; expected simple frame or bulk frame, got {:?}", frame ))), } }
/// Return the next entry as raw bytes. /// If the next entry cannot be represented as raw bytes, an error is /// returned. pub(crate) fnnext_bytes(&mutself) ->Result<Bytes, MiniRedisParseError> { matchself.next()? { // Both `Simple` and `Bulk` representation may be raw bytes. // // Although errors are stored as strings and could be represented as // raw bytes, they are considered separate types. Frame::Simple(s) => Ok(Bytes::from(s.into_bytes())), Frame::Bulk(data) => Ok(data), frame => Err(MiniRedisParseError::Parse(format!( "protocol error; expected simple frame or bulk frame, got {:?}", frame ))), } }
/// Return the next entry as an integer. /// /// This includes `Simple`, `Bulk`, and `Integer` frame types. `Simple` and /// `Bulk` frame types are parsed. /// /// If the next entry cannot be represented as an integer, then an error is /// returned. pub(crate) fnnext_int(&mutself) ->Result<u64, MiniRedisParseError> { use atoi::atoi;
matchself.next()? { // An integer frame type is already stored as an integer. Frame::Integer(v) => Ok(v), // Simple and bulk frames must be parsed as integers. If the parsing // fails, an error is returned. Frame::Simple(data) => atoi::<u64>(data.as_bytes()) .ok_or_else(|| MiniRedisParseError::Parse("protocol error; invalid number".into())), Frame::Bulk(data) => atoi::<u64>(&data) .ok_or_else(|| MiniRedisParseError::Parse("protocol error; invalid number".into())), frame => Err(MiniRedisParseError::Parse(format!( "protocol error; expected int frame but got {:?}", frame ))), } }
/// Ensure there are no more entries in the array pub(crate) fnfinish(&mutself) ->Result<(), MiniRedisParseError> { ifself.parts.next().is_none() { Ok(()) } else { Err(MiniRedisParseError::Parse( "protocol error; expected end of frame, but there was more".into(), )) } } }
解析如下：
new 方法：创建了一个 IntoIter 类型的迭代器，类似于流数据，一旦获取到了下一个 Frame，则交出所有权！
/// Send and receive `Frame` values from a remote peer. /// /// When implementing networking protocols, a message on that protocol is /// often composed of several smaller messages known as frames. The purpose of /// `Connection` is to read and write frames on the underlying `TcpStream`. /// /// To read frames, the `Connection` uses an internal buffer, which is filled /// up until there are enough bytes to create a full frame. Once this happens, /// the `Connection` creates the frame and returns it to the caller. /// /// When sending frames, the frame is first encoded into the write buffer. /// The contents of the write buffer are then written to the socket. #[derive(Debug)] pubstructConnection { /// The `TcpStream`. It is decorated with a `BufWriter`, which provides write /// level buffering. The `BufWriter` implementation provided by Tokio is /// sufficient for our needs. stream: BufWriter<TcpStream>,
// The buffer for reading frames. buffer: BytesMut, }
use tokio::io::{AsyncReadExt}; pubasyncfnread_frame(&mutself) ->Result<Option<Frame>, MiniRedisConnectionError> { loop { // Attempt to parse a frame from the buffered data. If enough data // has been buffered, the frame is returned. ifletSome(frame) = self.parse_frame()? { returnOk(Some(frame)); }
// There is not enough buffered data to read a frame. Attempt to // read more data from the socket. // // On success, the number of bytes is returned. `0` indicates "end // of stream". if0 == self.stream.read_buf(&mutself.buffer).await? { // The remote closed the connection. For this to be a clean // shutdown, there should be no data in the read buffer. If // there is, this means that the peer closed the socket while // sending a frame. returnifself.buffer.is_empty() { Ok(None) } else { Err(MiniRedisConnectionError::Disconnect) }; } } }
fnparse_frame(&mutself) ->Result<Option<Frame>, MiniRedisConnectionError> { // Cursor is used to track the "current" location in the // buffer. Cursor also implements `Buf` from the `bytes` crate // which provides a number of helpful utilities for working // with bytes. letmut buf = Cursor::new(&self.buffer[..]);
// The first step is to check if enough data has been buffered to parse a single frame. // This step is usually much faster than doing a full // parse of the frame, and allows us to skip allocating data structures // to hold the frame data unless we know the full frame has been received. match Frame::check(&mut buf) { Ok(_) => { // The `check` function will have advanced the cursor until the // end of the frame. Since the cursor had position set to zero // before `Frame::check` was called, we obtain the length of the // frame by checking the cursor position. letlen = buf.position() asusize;
// Reset the position to zero before passing the cursor to // `Frame::parse`. buf.set_position(0);
// Parse the frame from the buffer. This allocates the necessary // structures to represent the frame and returns the frame value. // // If the encoded frame representation is invalid, an error is // returned. This should terminate the **current** connection // but should not impact any other connected client. letframe = Frame::parse(&mut buf)?;
// Discard the parsed data from the read buffer. // // When `advance` is called on the read buffer, all of the data // up to `len` is discarded. The details of how this works is // left to `BytesMut`. This is often done by moving an internal // cursor, but it may be done by reallocating and copying data. self.buffer.advance(len);
// Return the parsed frame to the caller. Ok(Some(frame)) } // There is not enough data present in the read buffer to parse a // single frame. We must wait for more data to be received from the // socket. Reading from the socket will be done in the statement // after this `match`. // // We do not want to return `Err` from here as this "error" is an // expected runtime condition. Err(MiniRedisParseError::Incomplete) => Ok(None), // An error was encountered while parsing the frame. The connection // is now in an invalid state. Returning `Err` from here will result // in the connection being closed. Err(e) => Err(e.into()), } }
/// Write a single `Frame` value to the underlying stream. /// /// The `Frame` value is written to the socket using the various `write_*` /// functions provided by `AsyncWrite`. Calling these functions directly on /// a `TcpStream` is **not** advised, as this will result in a large number of /// syscalls. However, it is fine to call these functions on a *buffered* /// write stream. The data will be written to the buffer. Once the buffer is /// full, it is flushed to the underlying socket. pubasyncfnwrite_frame(&mutself, frame: &Frame) ->Result<(), MiniRedisConnectionError> { // Arrays are encoded by encoding each entry. All other frame types are // considered literals. For now, mini-redis is not able to encode // recursive frame structures. See below for more details. match frame { Frame::Array(val) => { // Encode the frame type prefix. For an array, it is `*`. self.stream.write_u8(b'*').await?;
// Encode the length of the array. self.write_decimal(val.len() asu64).await?;
// Iterate and encode each entry in the array. forentryin val { self.write_value(entry).await?; } } // The frame type is a literal. Encode the value directly. _ => self.write_value(frame).await?, }
// Ensure the encoded frame is written to the socket. The calls above // are to the buffered stream and writes. Calling `flush` writes the // remaining contents of the buffer to the socket. self.stream.flush().await.map_err(|e| e.into()) }
self.stream.write_u8(b'$').await?; self.write_decimal(len asu64).await?; self.stream.write_all(val).await?; self.stream.write_all(b"\r\n").await?; } // Encoding an `Array` from within a value cannot be done using a // recursive strategy. In general, async fns do not support // recursion. Mini-redis has not needed to encode nested arrays yet, // so for now it is skipped. Frame::Array(_val) => { warn!("unreachable code: recursive write_value: {:?}", _val); returnErr(MiniRedisParseError::Unimplemented.into()); } }
Ok(()) }
/// Write a decimal frame to the stream asyncfnwrite_decimal(&mutself, val: u64) ->Result<(), MiniRedisConnectionError> { use std::io::Write;
// Convert the value to a string letmut buf = [0u8; 20]; letmut buf = Cursor::new(&mut buf[..]);
