refactor(trogon-nats): align JetStream trait signatures with async_nats (#73)

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

Author: yordis

rsworkspace/crates/acp-nats/src/agent/test_support.rs

@@ -42,9 +42,9 @@ impl MockJs {
 impl trogon_nats::jetstream::JetStreamPublisher for MockJs {
     type PublishError = trogon_nats::mocks::MockError;
 
-    async fn js_publish_with_headers(
+    async fn js_publish_with_headers<S: async_nats::subject::ToSubject + Send>(
         &self,
-        subject: String,
+        subject: S,
         headers: async_nats::HeaderMap,
         payload: bytes::Bytes,
     ) -> Result<async_nats::jetstream::publish::PublishAck, Self::PublishError> {

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

@@ -4,6 +4,7 @@ use async_nats::jetstream::AckKind;
 use async_nats::jetstream::consumer::pull;
 use async_nats::jetstream::publish::PublishAck;
 use async_nats::jetstream::stream;
+use async_nats::subject::ToSubject;
 use bytes::Bytes;
 use futures::StreamExt;
 
@@ -41,7 +42,10 @@ impl std::error::Error for JetStreamError {}
 impl JetStreamContext for NatsJetStreamClient {
     type Error = JetStreamError;
 
-    async fn get_or_create_stream(&self, config: stream::Config) -> Result<(), JetStreamError> {
+    async fn get_or_create_stream<S: Into<stream::Config> + Send>(
+        &self,
+        config: S,
+    ) -> Result<(), JetStreamError> {
         self.context
             .get_or_create_stream(config)
             .await
@@ -53,9 +57,9 @@ impl JetStreamContext for NatsJetStreamClient {
 impl JetStreamPublisher for NatsJetStreamClient {
     type PublishError = JetStreamError;
 
-    async fn js_publish_with_headers(
+    async fn js_publish_with_headers<S: ToSubject + Send>(
         &self,
-        subject: String,
+        subject: S,
         headers: HeaderMap,
         payload: Bytes,
     ) -> Result<PublishAck, JetStreamError> {

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

@@ -6,6 +6,7 @@ use async_nats::jetstream::AckKind;
 use async_nats::jetstream::consumer::pull;
 use async_nats::jetstream::publish::PublishAck;
 use async_nats::jetstream::stream;
+use async_nats::subject::ToSubject;
 use bytes::Bytes;
 use futures::channel::mpsc;
 use futures::stream::BoxStream;
@@ -171,7 +172,11 @@ impl Default for MockJetStreamContext {
 impl JetStreamContext for MockJetStreamContext {
     type Error = MockError;
 
-    async fn get_or_create_stream(&self, config: stream::Config) -> Result<(), MockError> {
+    async fn get_or_create_stream<S: Into<stream::Config> + Send>(
+        &self,
+        config: S,
+    ) -> Result<(), MockError> {
+        let config = config.into();
         let should_fail = {
             let mut flag = self.should_fail.lock().unwrap();
             if *flag {
@@ -250,12 +255,13 @@ impl Default for MockJetStreamPublisher {
 impl JetStreamPublisher for MockJetStreamPublisher {
     type PublishError = MockError;
 
-    async fn js_publish_with_headers(
+    async fn js_publish_with_headers<S: ToSubject + Send>(
         &self,
-        subject: String,
+        subject: S,
         headers: HeaderMap,
         payload: Bytes,
     ) -> Result<PublishAck, MockError> {
+        let subject = subject.to_subject().to_string();
         let should_fail = {
             let mut count = self.publish_fail_count.lock().unwrap();
             if *count > 0 {

rsworkspace/crates/trogon-nats/src/jetstream/traits.rs

@@ -5,24 +5,25 @@ use async_nats::HeaderMap;
 use async_nats::jetstream::consumer::pull;
 use async_nats::jetstream::publish::PublishAck;
 use async_nats::jetstream::stream;
+use async_nats::subject::ToSubject;
 use bytes::Bytes;
 use futures::Stream;
 
 pub trait JetStreamContext: Send + Sync + Clone + 'static {
     type Error: Error + Send + Sync;
 
-    fn get_or_create_stream(
+    fn get_or_create_stream<S: Into<stream::Config> + Send>(
         &self,
-        config: stream::Config,
+        config: S,
     ) -> impl Future<Output = Result<(), Self::Error>> + Send;
 }
 
 pub trait JetStreamPublisher: Send + Sync + Clone + 'static {
     type PublishError: Error + Send + Sync;
 
-    fn js_publish_with_headers(
+    fn js_publish_with_headers<S: ToSubject + Send>(
         &self,
-        subject: String,
+        subject: S,
         headers: HeaderMap,
         payload: Bytes,
     ) -> impl Future<Output = Result<PublishAck, Self::PublishError>> + Send;
@@ -64,9 +65,9 @@ impl Error for NoJetStream {}
 impl JetStreamPublisher for () {
     type PublishError = NoJetStream;
 
-    async fn js_publish_with_headers(
+    async fn js_publish_with_headers<S: ToSubject + Send>(
         &self,
-        _subject: String,
+        _subject: S,
         _headers: HeaderMap,
         _payload: Bytes,
     ) -> Result<PublishAck, NoJetStream> {
@@ -193,7 +194,7 @@ mod tests {
 
     #[tokio::test]
     async fn unit_publisher_returns_err() {
-        let result = ().js_publish_with_headers("s".into(), HeaderMap::new(), Bytes::new()).await;
+        let result = ().js_publish_with_headers("s", HeaderMap::new(), Bytes::new()).await;
         assert!(result.is_err());
     }