Subscriber-Driven OutputPort Subscriptions Proposal for Ractor

[dcadenas]

Hi! Exploring Ractor's OutputPort, I found a way to boost decoupling: subscribers managing their subscriptions. It aligns with the essence of event-driven systems — publishers need not know subscriber details.

I introduced OutputPortSubscriberTrait to let subscribers connect directly to OutputPort, simplifying setup and reinforcing separation between event generation and handling. This leans into the event-driven architecture, promoting clearer separation of concerns.

Code Note:
The implementation is straightforward, focusing on message transformations via the From trait and leveraging the inherent knowledge of subscribers about their message types. By shifting the port.subscribe call to subscribers, it minimizes code while maximizing clarity and decoupling.

If this seems like a fit for Ractor, I'm all in for contributing a PR. Keen to hear your take!

use ractor::ActorRef;
use ractor::Message;
use std::fmt::Debug;

pub type OutputPortSubscriber<InputMessage> = Box<dyn OutputPortSubscriberTrait<InputMessage>>;

pub trait OutputPortSubscriberTrait<I>: Debug + Send
where
    I: Send + Clone + Debug + 'static,
{
    fn subscribe_to_port(&self, port: &ractor::OutputPort<I>);
}

impl<I, O> OutputPortSubscriberTrait<I> for ActorRef<O>
where
    I: Message + Debug + Clone,
    O: Message + Debug + From<I>,
{
    fn subscribe_to_port(&self, port: &ractor::OutputPort<I>) {
        port.subscribe(self.clone(), |msg| Some(O::from(msg)));
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use anyhow::Result;
    use ractor::{cast, Actor, ActorProcessingErr, OutputPort};
    use std::sync::Arc;
    use tokio::sync::Mutex;
    use tokio::time::{sleep, Duration};

    enum NumberPublisherMessage {
        Publish(u8),
        Subscribe(OutputPortSubscriber<u8>),
    }

    struct NumberPublisher;

    #[ractor::async_trait]
    impl Actor for NumberPublisher {
        type State = OutputPort<u8>;
        type Msg = NumberPublisherMessage;
        type Arguments = ();

        async fn pre_start(
            &self,
            _myself: ActorRef<Self::Msg>,
            _: (),
        ) -> Result<Self::State, ActorProcessingErr> {
            Ok(OutputPort::default())
        }

        async fn handle(
            &self,
            _myself: ActorRef<Self::Msg>,
            message: Self::Msg,
            state: &mut Self::State,
        ) -> Result<(), ActorProcessingErr> {
            match message {
                NumberPublisherMessage::Subscribe(subscriber) => {
                    // The key here is that it's the subscriber that does the mapping, it could be anything
                    // that implements the trait but most of the time it will be ActorRef instances
                    subscriber.subscribe_to_port(state);
                }
                NumberPublisherMessage::Publish(value) => {
                    state.send(value);
                }
            }
            Ok(())
        }
    }

    #[derive(Debug)]
    enum PlusSubscriberMessage {
        Plus(u8),
    }
    
    // The mapping is done through implementing From in the subscriber message.
    // For more specificity we could use newtypes
    impl From<u8> for PlusSubscriberMessage {
        fn from(value: u8) -> Self {
            PlusSubscriberMessage::Plus(value)
        }
    }

    struct PlusSubscriber;
    #[ractor::async_trait]
    impl Actor for PlusSubscriber {
        type State = Arc<Mutex<u8>>;
        type Msg = PlusSubscriberMessage;
        type Arguments = Self::State;

        async fn pre_start(
            &self,
            _myself: ActorRef<Self::Msg>,
            state: Self::State,
        ) -> Result<Self::State, ActorProcessingErr> {
            Ok(state)
        }

        async fn handle(
            &self,
            _myself: ActorRef<Self::Msg>,
            message: Self::Msg,
            state: &mut Self::State,
        ) -> Result<(), ActorProcessingErr> {
            match message {
                PlusSubscriberMessage::Plus(value) => {
                    let mut state = state.lock().await;
                    *state += value;
                }
            }
            Ok(())
        }
    }

    #[derive(Debug)]
    enum MulSubscriberMessage {
        Mul(u8),
    }
    impl From<u8> for MulSubscriberMessage {
        fn from(value: u8) -> Self {
            MulSubscriberMessage::Mul(value)
        }
    }

    struct MulSubscriber;
    #[ractor::async_trait]
    impl Actor for MulSubscriber {
        type State = Arc<Mutex<u8>>;
        type Msg = MulSubscriberMessage;
        type Arguments = Self::State;

        async fn pre_start(
            &self,
            _myself: ActorRef<Self::Msg>,
            state: Self::State,
        ) -> Result<Self::State, ActorProcessingErr> {
            Ok(state)
        }

        async fn handle(
            &self,
            _myself: ActorRef<Self::Msg>,
            message: Self::Msg,
            state: &mut Self::State,
        ) -> Result<(), ActorProcessingErr> {
            match message {
                MulSubscriberMessage::Mul(value) => {
                    let mut state = state.lock().await;
                    *state *= value;
                }
            }
            Ok(())
        }
    }

    #[tokio::test]
    async fn test_output_port_subscriber() -> Result<()> {
        let (number_publisher_ref, number_publisher_handler) =
            Actor::spawn(None, NumberPublisher, ()).await?;

        let plus_result = Arc::new(Mutex::new(0));
        let (plus_subcriber_ref, plus_subscriber_handler) =
            Actor::spawn(None, PlusSubscriber, plus_result.clone()).await?;

        let mul_result = Arc::new(Mutex::new(1));
        let (mul_subcriber_ref, mul_subscriber_handler) =
            Actor::spawn(None, MulSubscriber, mul_result.clone()).await?;

        cast!(
            number_publisher_ref,
            NumberPublisherMessage::Subscribe(Box::new(plus_subcriber_ref.clone()))
        )?;
        cast!(
            number_publisher_ref,
            NumberPublisherMessage::Subscribe(Box::new(mul_subcriber_ref.clone()))
        )?;

        cast!(number_publisher_ref, NumberPublisherMessage::Publish(2))?;
        cast!(number_publisher_ref, NumberPublisherMessage::Publish(3))?;

        sleep(Duration::from_secs(1)).await;

        assert_eq!(2 + 3, *plus_result.lock().await);
        assert_eq!(2 * 3, *mul_result.lock().await);

        number_publisher_ref.stop(None);
        plus_subcriber_ref.stop(None);
        mul_subcriber_ref.stop(None);

        number_publisher_handler.await?;
        plus_subscriber_handler.await?;
        mul_subscriber_handler.await?;

        Ok(())
    }
}

[slawlor]

Given this is a feature request I think this fits better as an issue. I like the idea and was honestly struggling with a good way to handle the OutputPort. Personally I view it as the weakest, and least-helpful, functionality in ractor today.

Please feel free to open an issue and setup a branch on a fork and I'd love to iterate on it!

Thanks :)