intercom-rs 1.1.1

//! Typed publisher with Sink trait implementation.
//!
//! The [`Publisher`] type provides a typed interface for publishing messages to NATS.
//! It implements the [`Sink`] trait for convenient stream-based publishing.
//!
//! **Note**: The Sink implementation uses fire-and-forget semantics for efficiency.
//! For applications requiring acknowledgment of each message, use the
//! [`Client::publish`](crate::Client::publish) method directly or the JetStream
//! [`JetStreamContext::publish`](crate::JetStreamContext::publish) for guaranteed delivery.

use std::{
    collections::VecDeque,
    marker::PhantomData,
    pin::Pin,
    sync::Arc,
    task::{Context, Poll},
};

use futures::Sink;
use serde::Serialize;

use crate::{codec::CodecType, error::Error};

/// A typed publisher that implements [`Sink`].
///
/// # Type Parameters
///
/// * `T` - The message type that will be serialized for publishing.
/// * `C` - The codec type used for serialization.
///
/// # Note on Error Handling
///
/// The [`Sink`] implementation uses fire-and-forget semantics for efficiency.
/// Publish errors are not propagated back through the Sink interface.
/// For applications requiring guaranteed delivery or error handling:
/// - Use [`Publisher::publish`] for direct publish with error handling
/// - Use [`Client::publish`](crate::Client::publish) for basic NATS
/// - Use [`JetStreamContext::publish`](crate::JetStreamContext::publish) for JetStream
///
/// # 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");
///
/// // Use Sink trait methods
/// publisher.send(MyMessage { content: "hello".into() }).await?;
///
/// // Or use feed + flush for batching
/// publisher.feed(MyMessage { content: "msg1".into() }).await?;
/// publisher.feed(MyMessage { content: "msg2".into() }).await?;
/// publisher.flush().await?;
/// # Ok(())
/// # }
/// ```
pub struct Publisher<T, C: CodecType> {
    client: Arc<async_nats::Client>,
    subject: String,
    buffer: VecDeque<Vec<u8>>,
    _marker: PhantomData<(T, C)>,
}

impl<T, C: CodecType> Publisher<T, C> {
    /// Create a new publisher for a subject.
    pub(crate) fn new(client: Arc<async_nats::Client>, subject: String) -> Self {
        Self {
            client,
            subject,
            buffer: VecDeque::new(),
            _marker: PhantomData,
        }
    }

    /// Get the subject this publisher publishes to.
    pub fn subject(&self) -> &str {
        &self.subject
    }

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

impl<T: Serialize, C: CodecType> Publisher<T, C> {
    /// Publish a message directly with error handling.
    ///
    /// Unlike the [`Sink`] implementation, this method returns any publish errors.
    ///
    /// # 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?;
    /// let publisher = client.publisher::<MyMessage>("subject");
    ///
    /// // Direct publish with error handling
    /// publisher.publish(&MyMessage { content: "hello".into() }).await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn publish(&self, message: &T) -> crate::error::Result<()> {
        let data = C::encode(message)?;
        self.client
            .publish(self.subject.clone(), data.into())
            .await?;
        Ok(())
    }
}

impl<T, C: CodecType> Clone for Publisher<T, C> {
    fn clone(&self) -> Self {
        Self {
            client: self.client.clone(),
            subject: self.subject.clone(),
            buffer: VecDeque::new(),
            _marker: PhantomData,
        }
    }
}

impl<T: Serialize + Unpin, C: CodecType + Unpin> Sink<T> for Publisher<T, C> {
    type Error = Error;

    fn poll_ready(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        // The NATS client is always ready to accept messages
        Poll::Ready(Ok(()))
    }

    fn start_send(self: Pin<&mut Self>, item: T) -> Result<(), Self::Error> {
        let data = C::encode(&item)?;
        let this = self.get_mut();
        this.buffer.push_back(data);
        Ok(())
    }

    fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        let this = self.get_mut();

        // Publish all buffered messages
        // Note: Uses fire-and-forget for Sink compatibility.
        // Use Publisher::publish() for error handling.
        while let Some(data) = this.buffer.pop_front() {
            let client = this.client.clone();
            let subject = this.subject.clone();
            tokio::spawn(async move {
                let _ = client.publish(subject, data.into()).await;
            });
        }

        Poll::Ready(Ok(()))
    }

    fn poll_close(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Result<(), Self::Error>> {
        self.poll_flush(cx)
    }
}