renamed clienterror

- also created a new basiccalloptions that includes message versioning and such

server: now message versioning works.

README.md

@@ -47,13 +47,30 @@ bunbun-worker = { git = "https://git.4o1x5.dev/4o1x5/bunbun-worker", branch = "m
 
 ## Usage
 
-Here is a basic implementation of an RPC job in bunbun-worker
+### Message versioning
+
+In this crate message versioning is done by including `v1.0.0` or such on the end of the queue name, instead of including it in the headers of a message. This reduces the amount of redelivered messages.  
+The following example will send a job to a queue named `emailjob-v1.0.0`.
+
+```rust
+let result = client
+        .rpc_call::<EmailJob>(
+            EmailJob {
+                send_to: "someone".into(),
+                contents: "something".into(),
+            },
+            BasicCallOptions::default("emailjob")
+                .timeout(Duration::from_secs(3))
+                .message_version("v1.0.0")
+        )
+        .await
+        .unwrap();
+```
 
 # Limitations
 
 1. Currently some `unwrap()`'s are called inside the code and may results in panics (not in the job-runner).
-2. No settings, and very limited API
-3. The rabbitmq RPC logic is very basic with no message-versioning (aside using different queue names (eg service-class-v1.0.0) )
+2. limited API
 
 # Bugs department
 

src/client.rs

@@ -1,7 +1,7 @@
 use std::{
     fmt::{Debug, Display},
     str::from_utf8,
-    time::Instant,
+    time::{Duration, Instant},
 };
 
 use futures::StreamExt;
@@ -10,6 +10,7 @@ use lapin::{
     ConnectionProperties,
 };
 use serde::{de::DeserializeOwned, Serialize};
+use tokio::time::timeout;
 use uuid::Uuid;
 
 use crate::ResultHeader;
@@ -47,8 +48,8 @@ impl Client {
     pub async fn rpc_call<T: RPCClientTask + Send + Debug>(
         &self,
         data: T,
-        queue_name: &str,
-    ) -> Result<Result<T::Result, T::ErroredResult>, RpcClientError>
+        options: BasicCallOptions,
+    ) -> Result<Result<T::Result, T::ErroredResult>, ClientError>
     where
         T: Serialize + DeserializeOwned,
     {
@@ -94,7 +95,7 @@ impl Client {
         match channel
             .basic_publish(
                 "",
-                queue_name,
+                format!("{}-{}", &options.queue_name, options.message_version).as_str(),
                 BasicPublishOptions::default(),
                 serde_json::to_string(&data).unwrap().as_bytes(),
                 BasicProperties::default()
@@ -112,22 +113,22 @@ impl Client {
                 }
                 Ok(confirmation) => {
                     tracing::info!(
-                        "Sent RPC job of type {} to channel {} Ack: {}",
+                        "Sent RPC job of type {} to channel {} Ack: {} Ver: {}",
                         std::any::type_name::<T>(),
-                        queue_name,
-                        confirmation.is_ack()
+                        options.queue_name,
+                        confirmation.is_ack(),
+                        options.message_version
                     );
                 }
             },
         }
 
-        // TODO implement timeout
         tracing::debug!("Awaiting response from callback queue");
-        let del = loop {
+        let listen = async move {
             match consumer.next().await {
                 None => {
                     tracing::error!("Received empty data after {:?}", now.elapsed());
-                    return Err(RpcClientError::NoReply);
+                    return Err(ClientError::InvalidResponse);
                 }
                 Some(del) => match del {
                     Err(error) => {
@@ -137,16 +138,30 @@ impl Client {
                             now.elapsed()
                         );
                         // Idk if i should nack it?
-                        return Err(RpcClientError::FailedDecode);
+                        return Err(ClientError::FailedDecode);
                     }
                     Ok(del) => {
                         tracing::debug!("Received response after {:?}", now.elapsed());
-                        break del;
+                        return Ok(del);
                     }
                 },
             };
         };
 
+        let del = match options.timeout {
+            None => listen.await?,
+            Some(dur) => match timeout(dur, listen).await {
+                Err(elapsed) => {
+                    tracing::warn!("RPC job has reached timeout after: {}", elapsed);
+                    return Err(ClientError::TimeoutReached);
+                }
+                Ok(r) => match r {
+                    Err(error) => return Err(error),
+                    Ok(r) => r,
+                },
+            },
+        };
+
         // TODO better implementation of this
         tracing::debug!("Decoding headers");
         let result_type = match del.properties.headers().to_owned() {
@@ -154,17 +169,17 @@ impl Client {
                 tracing::error!(
                     "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") {
                 None => {
                     tracing::error!("Got a response with no outcome header");
-                    return Err(RpcClientError::InvalidResponse);
+                    return Err(ClientError::InvalidResponse);
                 }
                 Some(res) => match res.as_long_string() {
                     None => {
                         tracing::error!("Got a response with no headers");
-                        return Err(RpcClientError::InvalidResponse);
+                        return Err(ClientError::InvalidResponse);
                     }
                     Some(outcome) => {
                         match serde_json::from_str::<ResultHeader>(outcome.to_string().as_str()) {
@@ -174,58 +189,53 @@ impl Client {
                             }
                             Err(_) => {
                                 tracing::warn!("Received a result header but it's not a type that can be deserailized ");
-                                return Err(RpcClientError::InvalidResponse);
+                                return Err(ClientError::InvalidResponse);
                             }
                         }
                     }
                 },
             },
         };
+
         tracing::debug!("Result type is: {result_type}, decoding...");
         let utf8 = match from_utf8(&del.data) {
             Ok(r) => r,
             Err(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;
-        // acking for idk reason
+
+        // ack message
         let _ = del.ack(BasicAckOptions::default()).await;
+
         match result_type {
             ResultHeader::Error => match serde_json::from_str::<T::ErroredResult>(utf8) {
                 // get result header
                 Err(_) => {
                     tracing::error!("Failed to decode response message to E");
-                    return Err(RpcClientError::FailedDecode);
+                    return Err(ClientError::FailedDecode);
                 }
                 Ok(res) => return Ok(Err(res)),
             },
-            ResultHeader::Panic => return Err(RpcClientError::ServerPanicked),
+            ResultHeader::Panic => return Err(ClientError::ServerPanicked),
             ResultHeader::Ok =>
             // get result
             {
                 match serde_json::from_str::<T::Result>(utf8) {
                     Err(_) => {
                         tracing::error!("Failed to decode response message to R");
-                        return Err(RpcClientError::FailedDecode);
+                        return Err(ClientError::FailedDecode);
                     }
                     Ok(res) => return Ok(Ok(res)),
                 }
             }
         }
-        // ack message
     }
-    /// Sends a message to the queue
-    ///
-    /// # Examples
-    ///
-    /// ```
-    /// use bunbun_worker::client::Client;
-    /// let client = Client::new("amqp://127.0.0.1:5672");
-    /// let result = client.call(EmailJob::new("someone@example.com", "Hello there"), "email-emailjob-v1.0.0");
-    /// ```
-    pub async fn call<T>(&self, data: T, queue_name: &str) -> Result<(), ClientError>
+
+    /// Sends a basic Task to the queue
+    pub async fn call<T>(&self, data: T, options: BasicCallOptions) -> Result<(), ClientError>
     where
         T: Serialize + DeserializeOwned,
     {
@@ -233,7 +243,7 @@ impl Client {
         match channel
             .basic_publish(
                 "",
-                queue_name,
+                format!("{}-{}", &options.queue_name, options.message_version).as_str(),
                 BasicPublishOptions::default(),
                 serde_json::to_string(&data).unwrap().as_bytes(),
                 BasicProperties::default(),
@@ -254,10 +264,11 @@ impl Client {
                 Ok(confirmation) => {
                     let _ = channel.close(0, "byebye").await;
                     tracing::info!(
-                        "Sent nonRPC job of type {} to channel {} Ack: {}",
+                        "Sent nonRPC job of type {} to channel {} Ack: {} Ver: {}",
                         std::any::type_name::<T>(),
-                        queue_name,
-                        confirmation.is_ack()
+                        options.queue_name,
+                        confirmation.is_ack(),
+                        options.message_version
                     );
                     tracing::debug!(
                         "AMQP confirmed dispatch of job |  Acknowledged? {}",
@@ -269,28 +280,39 @@ impl Client {
         Ok(())
     }
 }
+/// A call option class that is used to control how calls are handled  
+/// You can define the timeout, and the message versions
+pub struct BasicCallOptions {
+    timeout: Option<Duration>,
+    queue_name: String,
+    message_version: String,
+}
+impl BasicCallOptions {
+    pub fn default(queue_name: impl Into<String>) -> Self {
+        Self {
+            timeout: None,
+            queue_name: queue_name.into(),
+            message_version: "v1.0.0".into(),
+        }
+    }
+    pub fn message_version(mut self, message_version: impl Into<String>) -> Self {
+        self.message_version = message_version.into();
+        self
+    }
+    pub fn timeout(mut self, timeout: Duration) -> Self {
+        self.timeout = Some(timeout);
+        self
+    }
+}
 
 /// An error that the client returns
 #[derive(Debug)]
-pub enum RpcClientError {
-    NoReply, // TODO timeout
+pub enum ClientError {
     FailedDecode,
     FailedToSend,
     InvalidResponse,
     ServerPanicked,
-}
-/// 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"),
-        }
-    }
+    TimeoutReached,
 }
 
 /// A Client-side trait that needs to be implemented for a type in order for the client to know return types.

src/lib.rs

@@ -90,7 +90,7 @@ impl WorkerConfig {
     ///
     /// # Arguments
     /// * `custom_tls` - Optional TLSconfig (if none defaults to lapins choice)
-    pub fn enable_tls(&mut self, custom_tls: Option<TlsConfig>) {
+    pub fn enable_tls(mut self, custom_tls: Option<TlsConfig>) -> Self {
         match custom_tls {
             Some(tls) => {
                 let tls = OwnedTLSConfig {
@@ -104,6 +104,45 @@ impl WorkerConfig {
             }
             None => self.tls = OwnedTLSConfig::default().into(),
         }
+        self
+    }
+}
+
+/// A worker configuration
+
+pub struct ListenerConfig {
+    prefetch_count: u16,
+    queue_name: String,
+    consumer_tag: String,
+    message_version: String,
+}
+
+impl ListenerConfig {
+    /// Create a new listener config
+    /// # Arguments
+    /// * `queue_name` - The name of the queue to listen to (e.g. service-serviceJobName-v1.0.0)
+    pub fn default(queue_name: impl Into<String>) -> Self {
+        Self {
+            prefetch_count: 0,
+            queue_name: queue_name.into(),
+            consumer_tag: "".into(),
+            message_version: "v1.0.0".into(),
+        }
+    }
+    /// Set the prefetch count for the listener  
+    ///  This serves as a maximum job count that can be processed at a time. (0 is unlimited)
+    pub fn set_prefetch_count(mut self, prefetch_count: u16) -> Self {
+        self.prefetch_count = prefetch_count;
+        self
+    }
+    /// Set the consumer tag for the listener  
+    pub fn set_consumer_tag(mut self, consumer_tag: impl Into<String>) -> Self {
+        self.consumer_tag = consumer_tag.into();
+        self
+    }
+    pub fn set_message_version(mut self, version: impl Into<String>) -> Self {
+        self.message_version = version.into();
+        self
     }
 }
 
@@ -154,20 +193,24 @@ impl Worker {
     /// Add a non-rpc listener to the worker object
     ///
     /// # Arguments
-    /// * `queue_name` - A string slice that holds the name of the queue to listen to (e.g. service-serviceJobName-v1.0.0)
     /// * `state` - An Arc of the state object that will be passed to the listener
+    /// * `listener_config` - An Arc of the state object that will be passed to the listener
     pub async fn add_non_rpc_consumer<J: Task + 'static + Send>(
         &mut self,
-        queue_name: &str,
         state: Arc<J::State>,
+        listener_config: ListenerConfig,
     ) where
         <J as Task>::State: std::marker::Send + Sync,
     {
         let consumer = self
             .channel
             .basic_consume(
-                queue_name,
-                "",
+                format!(
+                    "{}-{}",
+                    listener_config.queue_name, listener_config.message_version
+                )
+                .as_str(),
+                &listener_config.consumer_tag,
                 BasicConsumeOptions::default(),
                 FieldTable::default(),
             )
@@ -231,28 +274,33 @@ impl Worker {
     ///
     /// ```
     ///    let server = BunBunWorker::new("amqp://localhost:5672", None).await;
-    ///    server.add_rpc_consumer::<MyRPCTask>("service-serviceJobName-v1.0.0", SomeState{} )).await;
+    ///    server.add_rpc_consumer::<MyRPCTask>(ListenerConfig::default("service-jobname-v1.0.0") )).await;
     ///    server.start_all_listeners().await;
     /// ```
     pub async fn add_rpc_consumer<J: RPCTask + 'static + Send>(
         &mut self,
-        queue_name: &str,
         state: Arc<J::State>,
+        listener_config: ListenerConfig,
     ) where
         <J as RPCTask>::State: std::marker::Send + Sync,
         <J as RPCTask>::Result: std::marker::Send + Sync,
         <J as RPCTask>::ErroredResult: std::marker::Send + Sync,
     {
-        let consumer = self
-            .channel
-            .basic_consume(
-                queue_name,
-                "",
-                BasicConsumeOptions::default(),
-                FieldTable::default(),
+        let consumer = create_consumer(
+            self.channel.clone(),
+            format!(
+                "{}-{}",
+                listener_config.queue_name, listener_config.message_version
             )
-            .await
-            .expect("basic_consume error");
+            .as_str(),
+            &listener_config.consumer_tag,
+            listener_config.prefetch_count,
+        )
+        .await
+        .map_err(|e| {
+            tracing::error!("Failed to create consumer: {}", e);
+        })
+        .expect("Failed to create consumer");
 
         let channel = self.channel.clone();
         let handler: Arc<
@@ -633,3 +681,24 @@ impl Display for ResultHeader {
         }
     }
 }
+
+async fn create_consumer(
+    channel: Channel,
+    queue_name: &str,
+    consumer_tag: &str,
+    prefect_count: u16,
+) -> Result<Consumer, lapin::Error> {
+    let channel = channel.clone();
+    channel
+        .basic_qos(prefect_count, BasicQosOptions::default())
+        .await?;
+
+    channel
+        .basic_consume(
+            queue_name,
+            consumer_tag,
+            BasicConsumeOptions::default(),
+            FieldTable::default(),
+        )
+        .await
+}

src/test/mod.rs

@@ -15,8 +15,8 @@ mod test {
     use tracing_test::traced_test;
 
     use crate::{
-        client::{Client, RPCClientTask},
-        RPCTask, Worker, WorkerConfig,
+        client::{BasicCallOptions, Client, RPCClientTask},
+        ListenerConfig, RPCTask, Worker, WorkerConfig,
     };
 
     #[derive(Clone, Debug)]
@@ -108,10 +108,10 @@ mod test {
         .await;
         listener
             .add_rpc_consumer::<EmailJob>(
-                "email-emailjob-v1.0.0",
                 Arc::new(State {
                     something: "test".into(),
                 }),
+                ListenerConfig::default("emailjob").set_prefetch_count(100),
             )
             .await;
         tracing::debug!("Starting listener");
@@ -129,10 +129,10 @@ mod test {
         .await;
         listener
             .add_rpc_consumer::<PanickingEmailJob>(
-                "email-emailjob-v1.0.0",
                 Arc::new(State {
                     something: "test".into(),
                 }),
+                ListenerConfig::default("emailjob").set_prefetch_count(100),
             )
             .await;
         tracing::debug!("Starting listener");
@@ -143,7 +143,7 @@ mod test {
     #[traced_test]
     async fn rpc_client() {
         //
-        let mut client = Client::new(env::var("AMQP_SERVER_URL").unwrap().as_str())
+        let client = Client::new(env::var("AMQP_SERVER_URL").unwrap().as_str())
             .await
             .unwrap();
         let result = client
@@ -152,7 +152,31 @@ mod test {
                     send_to: "someone".into(),
                     contents: "something".into(),
                 },
-                "email-emailjob-v1.0.0",
+                BasicCallOptions::default("emailjob"),
+            )
+            .await
+            .unwrap();
+        assert_eq!(
+            result,
+            Ok(EmailJobResult {
+                contents: "something".to_string()
+            })
+        )
+    }
+    #[test(tokio::test)]
+    #[traced_test]
+    async fn rpc_client_timeout() {
+        //
+        let client = Client::new(env::var("AMQP_SERVER_URL").unwrap().as_str())
+            .await
+            .unwrap();
+        let result = client
+            .rpc_call::<EmailJob>(
+                EmailJob {
+                    send_to: "someone".into(),
+                    contents: "something".into(),
+                },
+                BasicCallOptions::default("emailjob").timeout(Duration::from_secs(3)),
             )
             .await
             .unwrap();
@@ -184,7 +208,7 @@ mod test {
                             send_to: "someone".into(),
                             contents: "something".into(),
                         },
-                        "email-emailjob-v1.0.0",
+                        BasicCallOptions::default("emailjob"),
                     )
                     .await
             }));