4o1x5/bunbun-worker
1
0
Fork 0
Code Issues Pull requests Actions Packages Projects Releases Wiki Activity

renamed clienterror

Browse source
This commit is contained in:
2005 2024-11-14 19:38:17 +01:00
parent 6b26c1a90c
commit 4012f84886
1 changed files with 15 additions and 28 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

43
src/client.rs
View file

@ -49,7 +49,7 @@ impl Client {
&self, &self,
data: T, data: T,
options: BasicCallOptions, options: BasicCallOptions,
) -> Result<Result<T::Result, T::ErroredResult>, RpcClientError> ) -> Result<Result<T::Result, T::ErroredResult>, ClientError>
where where
T: Serialize + DeserializeOwned, T: Serialize + DeserializeOwned,
{ {
@ -128,7 +128,7 @@ impl Client {
match consumer.next().await { match consumer.next().await {
None => { None => {
tracing::error!("Received empty data after {:?}", now.elapsed()); tracing::error!("Received empty data after {:?}", now.elapsed());
return Err(RpcClientError::NoReply); return Err(ClientError::InvalidResponse);
} }
Some(del) => match del { Some(del) => match del {
Err(error) => { Err(error) => {
@ -138,7 +138,7 @@ impl Client {
now.elapsed() now.elapsed()
); );
// Idk if i should nack it? // Idk if i should nack it?
return Err(RpcClientError::FailedDecode); return Err(ClientError::FailedDecode);
} }
Ok(del) => { Ok(del) => {
tracing::debug!("Received response after {:?}", now.elapsed()); tracing::debug!("Received response after {:?}", now.elapsed());
@ -153,7 +153,7 @@ impl Client {
Some(dur) => match timeout(dur, listen).await { Some(dur) => match timeout(dur, listen).await {
Err(elapsed) => { Err(elapsed) => {
tracing::warn!("RPC job has reached timeout after: {}", elapsed); tracing::warn!("RPC job has reached timeout after: {}", elapsed);
return Err(RpcClientError::TimeoutReached); return Err(ClientError::TimeoutReached);
} }
Ok(r) => match r { Ok(r) => match r {
Err(error) => return Err(error), Err(error) => return Err(error),
@ -169,17 +169,17 @@ impl Client {
tracing::error!( tracing::error!(
"Got a response with no headers, this might be an issue with version mismatch" "Got a response with no headers, this might be an issue with version mismatch"
); );
return Err(RpcClientError::InvalidResponse); return Err(ClientError::InvalidResponse);
} }
Some(headers) => match headers.inner().get("outcome") { Some(headers) => match headers.inner().get("outcome") {
None => { None => {
tracing::error!("Got a response with no outcome header"); tracing::error!("Got a response with no outcome header");
return Err(RpcClientError::InvalidResponse); return Err(ClientError::InvalidResponse);
} }
Some(res) => match res.as_long_string() { Some(res) => match res.as_long_string() {
None => { None => {
tracing::error!("Got a response with no headers"); tracing::error!("Got a response with no headers");
return Err(RpcClientError::InvalidResponse); return Err(ClientError::InvalidResponse);
} }
Some(outcome) => { Some(outcome) => {
match serde_json::from_str::<ResultHeader>(outcome.to_string().as_str()) { match serde_json::from_str::<ResultHeader>(outcome.to_string().as_str()) {
@ -189,7 +189,7 @@ impl Client {
} }
Err(_) => { Err(_) => {
tracing::warn!("Received a result header but it's not a type that can be deserailized "); tracing::warn!("Received a result header but it's not a type that can be deserailized ");
return Err(RpcClientError::InvalidResponse); return Err(ClientError::InvalidResponse);
} }
} }
} }
@ -202,35 +202,36 @@ impl Client {
Ok(r) => r, Ok(r) => r,
Err(error) => { Err(error) => {
tracing::error!("Failed to decode response message to utf8 {error}"); tracing::error!("Failed to decode response message to utf8 {error}");
return Err(RpcClientError::FailedDecode); return Err(ClientError::FailedDecode);
} }
}; };
let _ = channel.close(0, "byebye").await; let _ = channel.close(0, "byebye").await;
// ack message
let _ = del.ack(BasicAckOptions::default()).await; let _ = del.ack(BasicAckOptions::default()).await;
match result_type { match result_type {
ResultHeader::Error => match serde_json::from_str::<T::ErroredResult>(utf8) { ResultHeader::Error => match serde_json::from_str::<T::ErroredResult>(utf8) {
// get result header // get result header
Err(_) => { Err(_) => {
tracing::error!("Failed to decode response message to E"); tracing::error!("Failed to decode response message to E");
return Err(RpcClientError::FailedDecode); return Err(ClientError::FailedDecode);
} }
Ok(res) => return Ok(Err(res)), Ok(res) => return Ok(Err(res)),
}, },
ResultHeader::Panic => return Err(RpcClientError::ServerPanicked), ResultHeader::Panic => return Err(ClientError::ServerPanicked),
ResultHeader::Ok => ResultHeader::Ok =>
// get result // get result
{ {
match serde_json::from_str::<T::Result>(utf8) { match serde_json::from_str::<T::Result>(utf8) {
Err(_) => { Err(_) => {
tracing::error!("Failed to decode response message to R"); tracing::error!("Failed to decode response message to R");
return Err(RpcClientError::FailedDecode); return Err(ClientError::FailedDecode);
} }
Ok(res) => return Ok(Ok(res)), Ok(res) => return Ok(Ok(res)),
} }
} }
} }
// ack message
} }
/// Sends a basic Task to the queue /// Sends a basic Task to the queue
@ -306,27 +307,13 @@ impl BasicCallOptions {
/// An error that the client returns /// An error that the client returns
#[derive(Debug)] #[derive(Debug)]
pub enum RpcClientError { pub enum ClientError {
NoReply,
FailedDecode, FailedDecode,
FailedToSend, FailedToSend,
InvalidResponse, InvalidResponse,
ServerPanicked, ServerPanicked,
TimeoutReached, TimeoutReached,
} }
/// An error for normal calls
#[derive(Debug)]
pub enum ClientError {
FailedToSend,
}
impl Display for ClientError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Self::FailedToSend => write!(f, "Failed to send to queue"),
}
}
}
/// A Client-side trait that needs to be implemented for a type in order for the client to know return types. /// A Client-side trait that needs to be implemented for a type in order for the client to know return types.
/// ///