intercom_rs/

client.rs

//! NATS client wrapper with typed publish/subscribe operations.

use std::{marker::PhantomData, sync::Arc};

use async_nats::ServerAddr;
use serde::{de::DeserializeOwned, Serialize};

use crate::{
    codec::CodecType,
    error::{Error, Result},
    jetstream::context::JetStreamContext,
    publisher::Publisher,
    subscriber::Subscriber,
};

/// A typed NATS client wrapper with configurable codec.
///
/// # Type Parameters
///
/// * `C` - The codec type to use for message serialization (e.g., `MsgPackCodec`, `JsonCodec`)
///
/// # Example
///
/// ```no_run
/// use intercom::{Client, MsgPackCodec};
///
/// # async fn example() -> intercom::Result<()> {
/// // Using MessagePack codec (default)
/// let msgpack_client = Client::<MsgPackCodec>::connect("nats://localhost:4222").await?;
/// # Ok(())
/// # }
/// ```
#[derive(Clone)]
pub struct Client<C: CodecType> {
    inner: Arc<async_nats::Client>,
    _codec: PhantomData<C>,
}

impl<C: CodecType> Client<C> {
    /// Connect to a NATS server.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use intercom::{Client, MsgPackCodec};
    ///
    /// # async fn example() -> intercom::Result<()> {
    /// let client = Client::<MsgPackCodec>::connect("nats://localhost:4222").await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn connect<A: ToServerAddrs>(addrs: A) -> Result<Self> {
        let client = async_nats::connect(addrs).await?;
        Ok(Self {
            inner: Arc::new(client),
            _codec: PhantomData,
        })
    }

    /// Connect to a NATS server with custom options.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use intercom::{Client, MsgPackCodec};
    ///
    /// # async fn example() -> intercom::Result<()> {
    /// let client = Client::<MsgPackCodec>::connect_with_options(
    ///     "nats://localhost:4222",
    ///     async_nats::ConnectOptions::new().name("my-app")
    /// ).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn connect_with_options<A: ToServerAddrs>(
        addrs: A,
        options: async_nats::ConnectOptions,
    ) -> Result<Self> {
        let client = options.connect(addrs).await?;
        Ok(Self {
            inner: Arc::new(client),
            _codec: PhantomData,
        })
    }

    /// Get the underlying async-nats client.
    pub fn inner(&self) -> &async_nats::Client {
        &self.inner
    }

    /// Publish a typed message to a subject.
    ///
    /// # Type Parameters
    ///
    /// * `T` - The message type (must implement Serialize)
    ///
    /// # Example
    ///
    /// ```no_run
    /// use intercom::{Client, MsgPackCodec};
    /// use serde::{Deserialize, Serialize};
    ///
    /// #[derive(Serialize, Deserialize)]
    /// struct MyMessage {
    ///     content: String,
    /// }
    ///
    /// # async fn example() -> intercom::Result<()> {
    /// let client = Client::<MsgPackCodec>::connect("nats://localhost:4222").await?;
    ///
    /// // Using turbofish syntax
    /// client.publish::<MyMessage>("subject", &MyMessage { content: "hello".into() }).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn publish<T: Serialize>(&self, subject: &str, message: &T) -> Result<()> {
        let data = C::encode(message)?;
        self.inner
            .publish(subject.to_string(), data.into())
            .await?;
        Ok(())
    }

    /// Publish a typed message and wait for a reply.
    ///
    /// # Type Parameters
    ///
    /// * `T` - The request message type (must implement Serialize)
    /// * `R` - The response message type (must implement DeserializeOwned)
    ///
    /// # Example
    ///
    /// ```no_run
    /// use intercom::{Client, MsgPackCodec};
    /// use serde::{Deserialize, Serialize};
    ///
    /// #[derive(Serialize, Deserialize)]
    /// struct Request { query: String }
    ///
    /// #[derive(Serialize, Deserialize)]
    /// struct Response { result: String }
    ///
    /// # async fn example() -> intercom::Result<()> {
    /// let client = Client::<MsgPackCodec>::connect("nats://localhost:4222").await?;
    ///
    /// let response = client.request::<Request, Response>(
    ///     "service",
    ///     &Request { query: "hello".into() }
    /// ).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn request<T: Serialize, R: DeserializeOwned>(
        &self,
        subject: &str,
        message: &T,
    ) -> Result<R> {
        let data = C::encode(message)?;
        let response = self.inner.request(subject.to_string(), data.into()).await?;
        C::decode(&response.payload)
    }

    /// Publish a message with a specific reply subject.
    ///
    /// # Type Parameters
    ///
    /// * `T` - The message type (must implement Serialize)
    ///
    /// # Example
    ///
    /// ```no_run
    /// use intercom::{Client, MsgPackCodec};
    /// use serde::{Deserialize, Serialize};
    ///
    /// #[derive(Serialize, Deserialize)]
    /// struct MyMessage { content: String }
    ///
    /// # async fn example() -> intercom::Result<()> {
    /// let client = Client::<MsgPackCodec>::connect("nats://localhost:4222").await?;
    ///
    /// client.publish_with_reply::<MyMessage>(
    ///     "subject",
    ///     "reply.subject",
    ///     &MyMessage { content: "hello".into() }
    /// ).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn publish_with_reply<T: Serialize>(
        &self,
        subject: &str,
        reply: &str,
        message: &T,
    ) -> Result<()> {
        let data = C::encode(message)?;
        self.inner
            .publish_with_reply(subject.to_string(), reply.to_string(), data.into())
            .await?;
        Ok(())
    }

    /// Subscribe to a subject with typed messages.
    ///
    /// Returns a [`Subscriber`] that implements [`futures::Stream`].
    ///
    /// # Type Parameters
    ///
    /// * `T` - The message type (must implement DeserializeOwned)
    ///
    /// # Example
    ///
    /// ```no_run
    /// use intercom::{Client, MsgPackCodec};
    /// use serde::{Deserialize, Serialize};
    /// use futures::StreamExt;
    ///
    /// #[derive(Serialize, Deserialize, Debug)]
    /// struct MyMessage { content: String }
    ///
    /// # async fn example() -> intercom::Result<()> {
    /// let client = Client::<MsgPackCodec>::connect("nats://localhost:4222").await?;
    ///
    /// let mut subscriber = client.subscribe::<MyMessage>("subject").await?;
    ///
    /// while let Some(result) = subscriber.next().await {
    ///     match result {
    ///         Ok(msg) => println!("Received: {:?}", msg.payload),
    ///         Err(e) => eprintln!("Error: {}", e),
    ///     }
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn subscribe<T: DeserializeOwned>(&self, subject: &str) -> Result<Subscriber<T, C>> {
        let inner = self.inner.subscribe(subject.to_string()).await?;
        Ok(Subscriber::new(inner))
    }

    /// Subscribe to a subject as part of a queue group.
    ///
    /// Messages are load-balanced across subscribers in the same queue group.
    ///
    /// # Type Parameters
    ///
    /// * `T` - The message type (must implement DeserializeOwned)
    ///
    /// # Example
    ///
    /// ```no_run
    /// use intercom::{Client, MsgPackCodec};
    /// use serde::{Deserialize, Serialize};
    ///
    /// #[derive(Serialize, Deserialize, Debug)]
    /// struct MyMessage { content: String }
    ///
    /// # async fn example() -> intercom::Result<()> {
    /// let client = Client::<MsgPackCodec>::connect("nats://localhost:4222").await?;
    ///
    /// let subscriber = client.queue_subscribe::<MyMessage>("subject", "my-queue").await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn queue_subscribe<T: DeserializeOwned>(
        &self,
        subject: &str,
        queue_group: &str,
    ) -> Result<Subscriber<T, C>> {
        let inner = self
            .inner
            .queue_subscribe(subject.to_string(), queue_group.to_string())
            .await?;
        Ok(Subscriber::new(inner))
    }

    /// Create a typed publisher for a subject.
    ///
    /// Returns a [`Publisher`] that implements [`futures::Sink`].
    ///
    /// # Type Parameters
    ///
    /// * `T` - The message type (must implement Serialize)
    ///
    /// # Example
    ///
    /// ```no_run
    /// use intercom::{Client, MsgPackCodec};
    /// use serde::{Deserialize, Serialize};
    /// use futures::SinkExt;
    ///
    /// #[derive(Serialize, Deserialize)]
    /// struct MyMessage { content: String }
    ///
    /// # async fn example() -> intercom::Result<()> {
    /// let client = Client::<MsgPackCodec>::connect("nats://localhost:4222").await?;
    ///
    /// let mut publisher = client.publisher::<MyMessage>("subject");
    ///
    /// publisher.send(MyMessage { content: "hello".into() }).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub fn publisher<T: Serialize>(&self, subject: &str) -> Publisher<T, C> {
        Publisher::new(self.inner.clone(), subject.to_string())
    }

    /// Create a JetStream context.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use intercom::{Client, MsgPackCodec};
    ///
    /// # async fn example() -> intercom::Result<()> {
    /// let client = Client::<MsgPackCodec>::connect("nats://localhost:4222").await?;
    /// let jetstream = client.jetstream();
    /// # Ok(())
    /// # }
    /// ```
    pub fn jetstream(&self) -> JetStreamContext<C> {
        JetStreamContext::new(async_nats::jetstream::new((*self.inner).clone()))
    }

    /// Create a JetStream context with a custom domain.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use intercom::{Client, MsgPackCodec};
    ///
    /// # async fn example() -> intercom::Result<()> {
    /// let client = Client::<MsgPackCodec>::connect("nats://localhost:4222").await?;
    /// let jetstream = client.jetstream_with_domain("my-domain");
    /// # Ok(())
    /// # }
    /// ```
    pub fn jetstream_with_domain(&self, domain: &str) -> JetStreamContext<C> {
        JetStreamContext::new(async_nats::jetstream::with_domain(
            (*self.inner).clone(),
            domain,
        ))
    }

    /// Create a JetStream context with a custom prefix.
    ///
    /// # Example
    ///
    /// ```no_run
    /// use intercom::{Client, MsgPackCodec};
    ///
    /// # async fn example() -> intercom::Result<()> {
    /// let client = Client::<MsgPackCodec>::connect("nats://localhost:4222").await?;
    /// let jetstream = client.jetstream_with_prefix("my-prefix");
    /// # Ok(())
    /// # }
    /// ```
    pub fn jetstream_with_prefix(&self, prefix: &str) -> JetStreamContext<C> {
        JetStreamContext::new(async_nats::jetstream::with_prefix(
            (*self.inner).clone(),
            prefix,
        ))
    }

    /// Flush any pending messages.
    pub async fn flush(&self) -> Result<()> {
        self.inner
            .flush()
            .await
            .map_err(|e| Error::Nats(e.into()))
    }
}

/// Trait for types that can be converted to server addresses.
pub trait ToServerAddrs: async_nats::ToServerAddrs {}

impl ToServerAddrs for &str {}
impl ToServerAddrs for String {}
impl ToServerAddrs for &String {}
impl ToServerAddrs for ServerAddr {}
impl ToServerAddrs for &ServerAddr {}
impl ToServerAddrs for Vec<ServerAddr> {}
impl ToServerAddrs for &[ServerAddr] {}