add unknown commands handling without breaking client connection (#26)
- fix client unsubscribe when subscribe list in the event of the received unsubscribe list is in a different order than subscribed - add tests for subscribe and unsubscribe commands - add tests for unknown command handling
This commit is contained in:
@@ -273,16 +273,16 @@ impl Subscriber {
|
||||
}
|
||||
|
||||
// Read the response
|
||||
for channel in &channels {
|
||||
for _channel in &channels {
|
||||
let response = self.read_response().await?;
|
||||
match response {
|
||||
Frame::Array(ref frame) => match frame.as_slice() {
|
||||
[unsubscribe, uchannel]
|
||||
if &unsubscribe.to_string() == "unsubscribe"
|
||||
&& &uchannel.to_string() == channel =>
|
||||
{
|
||||
self.subscribed_channels.remove(&uchannel.to_string());
|
||||
}
|
||||
[unsubscribe, uchannel] if &unsubscribe.to_string() == "unsubscribe" => {
|
||||
//unsubscribed channel should exist in the subscribed list at this point
|
||||
if self.subscribed_channels.remove(&uchannel.to_string()) == false {
|
||||
return Err(response.to_error());
|
||||
}
|
||||
},
|
||||
_ => return Err(response.to_error()),
|
||||
},
|
||||
frame => return Err(frame.to_error()),
|
||||
|
||||
@@ -10,6 +10,9 @@ pub use set::Set;
|
||||
mod subscribe;
|
||||
pub use subscribe::{Subscribe, Unsubscribe};
|
||||
|
||||
mod unknown;
|
||||
pub use unknown::Unknown;
|
||||
|
||||
use crate::{Connection, Db, Frame, Parse, ParseError, Shutdown};
|
||||
|
||||
#[derive(Debug)]
|
||||
@@ -19,6 +22,7 @@ pub(crate) enum Command {
|
||||
Set(Set),
|
||||
Subscribe(Subscribe),
|
||||
Unsubscribe(Unsubscribe),
|
||||
Unknown(Unknown)
|
||||
}
|
||||
|
||||
impl Command {
|
||||
@@ -33,7 +37,10 @@ impl Command {
|
||||
"set" => Command::Set(Set::parse_frames(&mut parse)?),
|
||||
"subscribe" => Command::Subscribe(Subscribe::parse_frames(&mut parse)?),
|
||||
"unsubscribe" => Command::Unsubscribe(Unsubscribe::parse_frames(&mut parse)?),
|
||||
_ => return Err(ParseError::UnknownCommand(command_name)),
|
||||
_ => {
|
||||
parse.next_string()?;
|
||||
Command::Unknown(Unknown::new(command_name))
|
||||
},
|
||||
};
|
||||
|
||||
parse.finish()?;
|
||||
@@ -53,9 +60,21 @@ impl Command {
|
||||
Publish(cmd) => cmd.apply(db, dst).await,
|
||||
Set(cmd) => cmd.apply(db, dst).await,
|
||||
Subscribe(cmd) => cmd.apply(db, dst, shutdown).await,
|
||||
Unknown(cmd) => cmd.apply(dst).await,
|
||||
// `Unsubscribe` cannot be applied. It may only be received from the
|
||||
// context of a `Subscribe` command.
|
||||
Unsubscribe(_) => unimplemented!(),
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) fn get_name(&self) -> &str {
|
||||
match self {
|
||||
Command::Get(_) => "get",
|
||||
Command::Publish(_) => "pub",
|
||||
Command::Set(_) => "set",
|
||||
Command::Subscribe(_) => "subscribe",
|
||||
Command::Unsubscribe(_) => "unsubscribe",
|
||||
Command::Unknown(cmd) => &cmd.command_name,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,4 +1,4 @@
|
||||
use crate::cmd::{Parse, ParseError};
|
||||
use crate::cmd::{Parse, ParseError, Unknown};
|
||||
use crate::{Command, Connection, Db, Frame, Shutdown};
|
||||
|
||||
use bytes::Bytes;
|
||||
@@ -134,9 +134,9 @@ impl Subscribe {
|
||||
dst.write_frame(&response).await?;
|
||||
}
|
||||
}
|
||||
_ => {
|
||||
// TODO: received invalid command
|
||||
unimplemented!();
|
||||
command => {
|
||||
let cmd = Unknown::new(command.get_name());
|
||||
cmd.apply(dst).await?;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
26
src/cmd/unknown.rs
Normal file
26
src/cmd/unknown.rs
Normal file
@@ -0,0 +1,26 @@
|
||||
use crate::{Connection, Frame};
|
||||
|
||||
use tracing::{debug, instrument};
|
||||
|
||||
#[derive(Debug)]
|
||||
pub struct Unknown {
|
||||
pub command_name: String,
|
||||
}
|
||||
|
||||
impl Unknown {
|
||||
/// Create a new `Unknown` command which responds to unknown commands
|
||||
/// issued by clients
|
||||
pub(crate) fn new(key: impl ToString) -> Unknown {
|
||||
Unknown { command_name: key.to_string() }
|
||||
}
|
||||
|
||||
#[instrument(skip(self, dst))]
|
||||
pub(crate) async fn apply(self, dst: &mut Connection) -> crate::Result<()> {
|
||||
let response = Frame::Error(format!("ERR unknown command '{}'", self.command_name));
|
||||
|
||||
debug!(?response);
|
||||
|
||||
dst.write_frame(&response).await?;
|
||||
Ok(())
|
||||
}
|
||||
}
|
||||
@@ -13,7 +13,6 @@ pub(crate) struct Parse {
|
||||
pub(crate) enum ParseError {
|
||||
EndOfStream,
|
||||
Invalid,
|
||||
UnknownCommand(String),
|
||||
}
|
||||
|
||||
impl Parse {
|
||||
@@ -85,7 +84,6 @@ impl fmt::Display for ParseError {
|
||||
let msg = match self {
|
||||
ParseError::EndOfStream => "end of stream".to_string(),
|
||||
ParseError::Invalid => "invalid".to_string(),
|
||||
ParseError::UnknownCommand(cmd) => format!("unknown command `{}`", cmd),
|
||||
};
|
||||
write!(f, "{}", &msg)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user