refactor(trogon-nats): use per-operation traits with ToSubject (#6)

Signed-off-by: Yordis Prieto <yordis.prieto@gmail.com>

Author: yordis

rsworkspace/crates/trogon-nats/src/client.rs

@@ -1,3 +1,4 @@
+use async_nats::subject::ToSubject;
 use async_nats::{Client as NatsAsyncClient, HeaderMap, Message, Subscriber};
 use bytes::Bytes;
 use std::error::Error;
@@ -6,18 +7,18 @@ use std::future::Future;
 pub trait SubscribeClient: Send + Sync + Clone + 'static {
     type SubscribeError: Error + Send + Sync;
 
-    fn subscribe(
+    fn subscribe<S: ToSubject + Send>(
         &self,
-        subject: String,
+        subject: S,
     ) -> impl Future<Output = Result<Subscriber, Self::SubscribeError>> + Send;
 }
 
 pub trait RequestClient: Send + Sync + Clone + 'static {
     type RequestError: Error + Send + Sync;
 
-    fn request_with_headers(
+    fn request_with_headers<S: ToSubject + Send>(
         &self,
-        subject: String,
+        subject: S,
         headers: HeaderMap,
         payload: Bytes,
     ) -> impl Future<Output = Result<Message, Self::RequestError>> + Send;
@@ -26,9 +27,9 @@ pub trait RequestClient: Send + Sync + Clone + 'static {
 pub trait PublishClient: Send + Sync + Clone + 'static {
     type PublishError: Error + Send + Sync;
 
-    fn publish_with_headers(
+    fn publish_with_headers<S: ToSubject + Send>(
         &self,
-        subject: String,
+        subject: S,
         headers: HeaderMap,
         payload: Bytes,
     ) -> impl Future<Output = Result<(), Self::PublishError>> + Send;
@@ -43,17 +44,20 @@ pub trait FlushClient: Send + Sync + Clone + 'static {
 impl SubscribeClient for NatsAsyncClient {
     type SubscribeError = async_nats::client::SubscribeError;
 
-    async fn subscribe(&self, subject: String) -> Result<Subscriber, Self::SubscribeError> {
+    async fn subscribe<S: ToSubject + Send>(
+        &self,
+        subject: S,
+    ) -> Result<Subscriber, Self::SubscribeError> {
         self.subscribe(subject).await
     }
 }
 
 impl RequestClient for NatsAsyncClient {
     type RequestError = async_nats::client::RequestError;
 
-    async fn request_with_headers(
+    async fn request_with_headers<S: ToSubject + Send>(
         &self,
-        subject: String,
+        subject: S,
         headers: HeaderMap,
         payload: Bytes,
     ) -> Result<Message, Self::RequestError> {
@@ -64,9 +68,9 @@ impl RequestClient for NatsAsyncClient {
 impl PublishClient for NatsAsyncClient {
     type PublishError = async_nats::client::PublishError;
 
-    async fn publish_with_headers(
+    async fn publish_with_headers<S: ToSubject + Send>(
         &self,
-        subject: String,
+        subject: S,
         headers: HeaderMap,
         payload: Bytes,
     ) -> Result<(), Self::PublishError> {

rsworkspace/crates/trogon-nats/src/lib.rs

@@ -43,6 +43,7 @@ pub mod messaging;
 #[cfg(feature = "test-support")]
 pub mod mocks;
 
+pub use async_nats::subject::ToSubject;
 pub use auth::{NatsAuth, NatsConfig};
 pub use client::{FlushClient, PublishClient, RequestClient, SubscribeClient};
 pub use connect::{ConnectError, connect};

rsworkspace/crates/trogon-nats/src/mocks.rs

@@ -1,4 +1,5 @@
 use crate::client::{FlushClient, PublishClient, RequestClient, SubscribeClient};
+use async_nats::subject::ToSubject;
 use std::sync::{Arc, Mutex};
 
 #[derive(Debug, Clone)]
@@ -118,18 +119,24 @@ impl Default for AdvancedMockNatsClient {
 impl SubscribeClient for MockNatsClient {
     type SubscribeError = MockError;
 
-    async fn subscribe(&self, subject: String) -> Result<async_nats::Subscriber, MockError> {
-        self.subscribed_subjects.lock().unwrap().push(subject);
+    async fn subscribe<S: ToSubject + Send>(
+        &self,
+        subject: S,
+    ) -> Result<async_nats::Subscriber, MockError> {
+        self.subscribed_subjects
+            .lock()
+            .unwrap()
+            .push(subject.to_subject().to_string());
         Err(MockError("mock: subscribe not implemented".to_string()))
     }
 }
 
 impl RequestClient for MockNatsClient {
     type RequestError = MockError;
 
-    async fn request_with_headers(
+    async fn request_with_headers<S: ToSubject + Send>(
         &self,
-        _subject: String,
+        _subject: S,
         _headers: async_nats::HeaderMap,
         _payload: bytes::Bytes,
     ) -> Result<async_nats::Message, MockError> {
@@ -140,16 +147,16 @@ impl RequestClient for MockNatsClient {
 impl PublishClient for MockNatsClient {
     type PublishError = MockError;
 
-    async fn publish_with_headers(
+    async fn publish_with_headers<S: ToSubject + Send>(
         &self,
-        subject: String,
+        subject: S,
         _headers: async_nats::HeaderMap,
         payload: bytes::Bytes,
     ) -> Result<(), MockError> {
-        self.published
-            .lock()
-            .unwrap()
-            .push(PublishedMessage { subject, payload });
+        self.published.lock().unwrap().push(PublishedMessage {
+            subject: subject.to_subject().to_string(),
+            payload,
+        });
         Ok(())
     }
 }
@@ -165,20 +172,24 @@ impl FlushClient for MockNatsClient {
 impl SubscribeClient for AdvancedMockNatsClient {
     type SubscribeError = MockError;
 
-    async fn subscribe(&self, subject: String) -> Result<async_nats::Subscriber, MockError> {
+    async fn subscribe<S: ToSubject + Send>(
+        &self,
+        subject: S,
+    ) -> Result<async_nats::Subscriber, MockError> {
         self.base.subscribe(subject).await
     }
 }
 
 impl RequestClient for AdvancedMockNatsClient {
     type RequestError = MockError;
 
-    async fn request_with_headers(
+    async fn request_with_headers<S: ToSubject + Send>(
         &self,
-        subject: String,
+        subject: S,
         _headers: async_nats::HeaderMap,
         _payload: bytes::Bytes,
     ) -> Result<async_nats::Message, MockError> {
+        let subject = subject.to_subject().to_string();
         let should_fail = *self.should_fail_request.lock().unwrap();
         if should_fail {
             *self.should_fail_request.lock().unwrap() = false;
@@ -207,9 +218,9 @@ impl RequestClient for AdvancedMockNatsClient {
 impl PublishClient for AdvancedMockNatsClient {
     type PublishError = MockError;
 
-    async fn publish_with_headers(
+    async fn publish_with_headers<S: ToSubject + Send>(
         &self,
-        subject: String,
+        subject: S,
         headers: async_nats::HeaderMap,
         payload: bytes::Bytes,
     ) -> Result<(), MockError> {
@@ -242,15 +253,19 @@ mod tests {
     async fn mock_client_tracks_publish() {
         let mock = MockNatsClient::new();
         let _ = mock
-            .publish_with_headers("foo".into(), async_nats::HeaderMap::new(), "bar".into())
+            .publish_with_headers(
+                "foo",
+                async_nats::HeaderMap::new(),
+                bytes::Bytes::from("bar"),
+            )
             .await;
         assert_eq!(mock.published_messages(), vec!["foo"]);
     }
 
     #[tokio::test]
     async fn mock_client_tracks_subscribe() {
         let mock = MockNatsClient::new();
-        let _ = mock.subscribe("test.sub".into()).await;
+        let _ = mock.subscribe("test.sub").await;
         assert_eq!(mock.subscribed_to(), vec!["test.sub"]);
     }
 
@@ -273,7 +288,11 @@ mod tests {
     async fn advanced_mock_request_no_response_configured() {
         let mock = AdvancedMockNatsClient::new();
         let result = mock
-            .request_with_headers("missing".into(), async_nats::HeaderMap::new(), "x".into())
+            .request_with_headers(
+                "missing",
+                async_nats::HeaderMap::new(),
+                bytes::Bytes::from("x"),
+            )
             .await;
         assert!(result.is_err());
         assert!(result.unwrap_err().0.contains("no response configured"));