Add custom request prefix option

async-nats/src/client.rs

@@ -57,19 +57,22 @@ pub struct Client {
     sender: mpsc::Sender<Command>,
     next_subscription_id: Arc<AtomicU64>,
     subscription_capacity: usize,
+    inbox_prefix: String,
 }
 
 impl Client {
     pub(crate) fn new(
         info: tokio::sync::watch::Receiver<ServerInfo>,
         sender: mpsc::Sender<Command>,
         capacity: usize,
+        inbox_prefix: String,
     ) -> Client {
         Client {
             info,
             sender,
             next_subscription_id: Arc::new(AtomicU64::new(0)),
             subscription_capacity: capacity,
+            inbox_prefix,
         }
     }
 
@@ -262,7 +265,7 @@ impl Client {
     /// # }
     /// ```
     pub fn new_inbox(&self) -> String {
-        format!("_INBOX.{}", nuid::next())
+        format!("{}.{}", self.inbox_prefix, nuid::next())
     }
 
     pub async fn subscribe(&self, subject: String) -> Result<Subscriber, Error> {

async-nats/src/lib.rs

@@ -630,7 +630,12 @@ pub async fn connect_with_options<A: ToServerAddrs>(
     // TODO make channel size configurable
     let (sender, receiver) = mpsc::channel(options.sender_capacity);
 
-    let client = Client::new(info_watcher, sender.clone(), options.subscription_capacity);
+    let client = Client::new(
+        info_watcher,
+        sender.clone(),
+        options.subscription_capacity,
+        options.inbox_prefix,
+    );
     tokio::spawn({
         let sender = sender.clone();
         async move {

async-nats/src/options.rs

@@ -49,6 +49,7 @@ pub struct ConnectOptions {
     pub(crate) subscription_capacity: usize,
     pub(crate) sender_capacity: usize,
     pub(crate) event_callback: CallbackArg1<Event, ()>,
+    pub(crate) inbox_prefix: String,
 }
 
 impl fmt::Debug for ConnectOptions {
@@ -67,6 +68,7 @@ impl fmt::Debug for ConnectOptions {
             .entry(&"flush_interval", &self.flush_interval)
             .entry(&"ping_interval", &self.ping_interval)
             .entry(&"sender_capacity", &self.sender_capacity)
+            .entry(&"inbox_prefix", &self.inbox_prefix)
             .finish()
     }
 }
@@ -94,6 +96,7 @@ impl Default for ConnectOptions {
                     println!("error : {}", error);
                 })
             })),
+            inbox_prefix: "_INBOX".to_string(),
         }
     }
 }
@@ -476,6 +479,22 @@ impl ConnectOptions {
         self.sender_capacity = capacity;
         self
     }
+
+    /// Sets custom prefix instead of default `_INBOX`.
+    ///
+    /// # Examples
+    ///
+    /// ```
+    /// # #[tokio::main]
+    /// # async fn main() -> Result<(), async_nats::Error> {
+    /// async_nats::ConnectOptions::new().custom_inbox_prefix("CUSTOM").connect("demo.nats.io").await?;
+    /// # Ok(())
+    /// # }
+    /// ```
+    pub fn custom_inbox_prefix<T: ToString>(mut self, prefix: T) -> ConnectOptions {
+        self.inbox_prefix = prefix.to_string();
+        self
+    }
 }
 
 type AsyncCallbackArg1<A, T> =

async-nats/tests/client_tests.rs

@@ -532,4 +532,34 @@ mod client {
         rx.recv().await;
         rx.recv().await;
     }
+
+    #[tokio::test]
+    async fn inbox_prefix() {
+        let server = nats_server::run_basic_server();
+        let client = ConnectOptions::new()
+            .custom_inbox_prefix("BOB")
+            .connect(server.client_url())
+            .await
+            .unwrap();
+
+        let mut inbox_wildcard_subscription = client.subscribe("BOB.>".to_string()).await.unwrap();
+        let mut subscription = client.subscribe("request".into()).await.unwrap();
+
+        tokio::task::spawn({
+            let client = client.clone();
+            async move {
+                let msg = subscription.next().await.unwrap();
+                client
+                    .publish(msg.reply.unwrap(), "prefix workes".into())
+                    .await
+                    .unwrap();
+            }
+        });
+
+        client
+            .request("request".into(), "data".into())
+            .await
+            .unwrap();
+        inbox_wildcard_subscription.next().await.unwrap();
+    }
 }