intercom-rs 1.1.1

//! Typed subscriber with Stream trait implementation.

use std::{
    marker::PhantomData,
    pin::Pin,
    task::{Context, Poll},
};

use futures::Stream;
use pin_project_lite::pin_project;
use serde::de::DeserializeOwned;

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

/// A typed message received from a subscription.
#[derive(Debug)]
pub struct Message<T> {
    /// The decoded message payload.
    pub payload: T,
    /// The subject the message was published to.
    pub subject: String,
    /// The reply subject, if any.
    pub reply: Option<String>,
    /// The raw underlying message (for advanced use cases).
    raw: async_nats::Message,
}

impl<T> Message<T> {
    /// Get the raw underlying NATS message.
    pub fn raw(&self) -> &async_nats::Message {
        &self.raw
    }
}

pin_project! {
    /// A typed subscriber that implements [`Stream`].
    ///
    /// # Type Parameters
    ///
    /// * `T` - The message type that will be deserialized from incoming messages.
    /// * `C` - The codec type used for deserialization.
    ///
    /// # 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!("Decode error: {}", e),
    ///     }
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub struct Subscriber<T, C: CodecType> {
        #[pin]
        inner: async_nats::Subscriber,
        _marker: PhantomData<(T, C)>,
    }
}

impl<T, C: CodecType> Subscriber<T, C> {
    /// Create a new subscriber wrapping an async-nats subscriber.
    pub(crate) fn new(inner: async_nats::Subscriber) -> Self {
        Self {
            inner,
            _marker: PhantomData,
        }
    }

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

    /// Unsubscribe from the subject.
    pub async fn unsubscribe(mut self) -> Result<()> {
        self.inner
            .unsubscribe()
            .await
            .map_err(|e| crate::error::Error::Nats(e.into()))
    }
}

impl<T: DeserializeOwned, C: CodecType> Stream for Subscriber<T, C> {
    type Item = Result<Message<T>>;

    fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
        let this = self.project();

        match this.inner.poll_next(cx) {
            Poll::Ready(Some(msg)) => {
                let reply = msg.reply.as_ref().map(|r| r.to_string());
                let result = C::decode(&msg.payload).map(|payload| Message {
                    payload,
                    subject: msg.subject.to_string(),
                    reply,
                    raw: msg,
                });
                Poll::Ready(Some(result))
            }
            Poll::Ready(None) => Poll::Ready(None),
            Poll::Pending => Poll::Pending,
        }
    }
}