feat(actor): implement selective receive with pattern matching

Add Erlang-style selective receive support to actors, matching the
semantics of Erlang's receive ... end with pattern matching.

Changes:
- Add pending queue (Mutex<VecDeque<Envelope>>) to Mailbox for non-matching messages
- Implement recv_matching() with timeout support for waiting on specific messages
- Implement try_recv_matching() for non-blocking selective receive
- Add ActorContext methods: receive(), receive_timeout(), try_receive()
- Messages check pending queue first, then channel
- Non-matching messages saved to pending queue for later processing
- Signals always go to pending queue (processed by handle_signal)
- Add comprehensive test suite with 6 tests covering all scenarios
- Fix Send/Sync issues by wrapping pending queue in Mutex

Implementation notes:
- Predicate functions: FnMut(&Message) -> Option<T>
- Timeout using tokio::time::timeout_at
- Allows actors to wait for specific messages while leaving others queued
- Maintains message ordering through VecDeque

Examples:
- RPC with correlation IDs
- Selective message processing
- Timeout-based patterns
- Pending queue verification

Author: am-kantox

joerl/src/actor.rs

@@ -298,6 +298,151 @@ impl ActorContext {
         }
         Err(crate::ActorError::SendFailed(to))
     }
+
+    /// Selectively receive a message matching the predicate.
+    ///
+    /// This is similar to Erlang's `receive` with pattern matching.
+    /// Messages that don't match remain in the mailbox and will be
+    /// checked again on subsequent receive calls or regular message processing.
+    ///
+    /// This allows actors to wait for specific messages while leaving
+    /// other messages in the mailbox for later processing.
+    ///
+    /// In Erlang: `receive Pattern -> Body end`
+    ///
+    /// # Arguments
+    ///
+    /// * `predicate` - Function that returns `Some(T)` if message matches
+    ///
+    /// # Examples
+    ///
+    /// ```rust
+    /// use joerl::{Actor, ActorContext, Message};
+    /// use async_trait::async_trait;
+    ///
+    /// #[derive(Clone)]
+    /// struct Response {
+    ///     id: u64,
+    ///     data: String,
+    /// }
+    ///
+    /// struct RpcActor {
+    ///     next_id: u64,
+    /// }
+    ///
+    /// #[async_trait]
+    /// impl Actor for RpcActor {
+    ///     async fn handle_message(&mut self, msg: Message, ctx: &mut ActorContext) {
+    ///         // Make a request with an ID
+    ///         let req_id = self.next_id;
+    ///         self.next_id += 1;
+    ///         
+    ///         // ... send request somewhere ...
+    ///         
+    ///         // Wait for specific response
+    ///         let response = ctx.receive(|msg| {
+    ///             msg.downcast_ref::<Response>()
+    ///                 .filter(|r| r.id == req_id)
+    ///                 .cloned()
+    ///         }).await;
+    ///         
+    ///         // Meanwhile, other messages stay in mailbox
+    ///     }
+    /// }
+    /// ```
+    pub async fn receive<F, T>(&mut self, predicate: F) -> Option<T>
+    where
+        F: FnMut(&Message) -> Option<T>,
+    {
+        self.mailbox.recv_matching(predicate, None).await
+    }
+
+    /// Selectively receive a message with timeout.
+    ///
+    /// Like `receive()`, but returns `None` if no matching message
+    /// arrives within the timeout duration.
+    ///
+    /// In Erlang: `receive Pattern -> Body after Timeout -> TimeoutBody end`
+    ///
+    /// # Arguments
+    ///
+    /// * `predicate` - Function that returns `Some(T)` if message matches
+    /// * `timeout` - Maximum time to wait
+    ///
+    /// # Examples
+    ///
+    /// ```rust
+    /// use joerl::{Actor, ActorContext, Message};
+    /// use async_trait::async_trait;
+    /// use std::time::Duration;
+    ///
+    /// #[derive(Clone)]
+    /// struct Ack {
+    ///     id: u64,
+    /// }
+    ///
+    /// struct Worker;
+    ///
+    /// #[async_trait]
+    /// impl Actor for Worker {
+    ///     async fn handle_message(&mut self, _msg: Message, ctx: &mut ActorContext) {
+    ///         // Wait for ack, but timeout after 5 seconds
+    ///         let ack = ctx.receive_timeout(
+    ///             |msg| msg.downcast_ref::<Ack>().cloned(),
+    ///             Duration::from_secs(5)
+    ///         ).await;
+    ///         
+    ///         match ack {
+    ///             Some(ack) => println!("Got ack: {}", ack.id),
+    ///             None => println!("Timeout waiting for ack"),
+    ///         }
+    ///     }
+    /// }
+    /// ```
+    pub async fn receive_timeout<F, T>(
+        &mut self,
+        predicate: F,
+        timeout: std::time::Duration,
+    ) -> Option<T>
+    where
+        F: FnMut(&Message) -> Option<T>,
+    {
+        self.mailbox.recv_matching(predicate, Some(timeout)).await
+    }
+
+    /// Try to receive a matching message without blocking.
+    ///
+    /// Returns immediately with `Some(T)` if a matching message is found,
+    /// or `None` if no match is available.
+    ///
+    /// # Examples
+    ///
+    /// ```rust
+    /// use joerl::{Actor, ActorContext, Message};
+    /// use async_trait::async_trait;
+    ///
+    /// struct Worker;
+    ///
+    /// #[async_trait]
+    /// impl Actor for Worker {
+    ///     async fn handle_message(&mut self, _msg: Message, ctx: &mut ActorContext) {
+    ///         // Check if there's a ready message without waiting
+    ///         if let Some(ready) = ctx.try_receive(|msg| {
+    ///             msg.downcast_ref::<String>()
+    ///                 .filter(|s| s == "ready")
+    ///                 .cloned()
+    ///         }) {
+    ///             println!("Got ready signal: {}", ready);
+    ///         }
+    ///     }
+    /// }
+    /// ```
+    pub fn try_receive<F, T>(&mut self, predicate: F) -> Option<T>
+    where
+        F: FnMut(&Message) -> Option<T>,
+    {
+        self.mailbox.try_recv_matching(predicate)
+    }
 }
 
 #[cfg(test)]

joerl/src/mailbox.rs

@@ -5,17 +5,26 @@
 
 use crate::message::Envelope;
 use crate::telemetry::MessageMetrics;
+use std::collections::VecDeque;
+use std::sync::Mutex;
+use std::time::Duration;
 use tokio::sync::mpsc;
-
-/// Default mailbox capacity.
 pub const DEFAULT_MAILBOX_CAPACITY: usize = 100;
 
 /// Actor mailbox for receiving messages.
 ///
 /// Mailboxes are bounded channels that provide backpressure when full.
 /// This prevents fast senders from overwhelming slow receivers.
+///
+/// The mailbox now supports selective receive (similar to Erlang's `receive`)
+/// by maintaining a pending queue for messages that don't match the current
+/// receive pattern.
 pub struct Mailbox {
     rx: mpsc::Receiver<Envelope>,
+    /// Messages that didn't match a selective receive predicate.
+    /// These are checked first on subsequent receives.
+    /// Wrapped in Mutex to maintain Sync for ActorContext.
+    pending: Mutex<VecDeque<Envelope>>,
 }
 
 impl Mailbox {
@@ -28,7 +37,10 @@ impl Mailbox {
     /// Creates a new mailbox with actor type for telemetry.
     pub(crate) fn new_with_type(capacity: usize, actor_type: String) -> (Self, MailboxSender) {
         let (tx, rx) = mpsc::channel(capacity);
-        let mailbox = Mailbox { rx };
+        let mailbox = Mailbox {
+            rx,
+            pending: Mutex::new(VecDeque::new()),
+        };
         let sender = MailboxSender {
             tx,
             actor_type,
@@ -40,7 +52,13 @@ impl Mailbox {
     /// Receives the next message from the mailbox.
     ///
     /// Returns `None` if all senders have been dropped.
+    ///
+    /// This now checks the pending queue first before receiving from the channel.
     pub(crate) async fn recv(&mut self) -> Option<Envelope> {
+        // Check pending queue first
+        if let Some(envelope) = self.pending.lock().unwrap().pop_front() {
+            return Some(envelope);
+        }
         self.rx.recv().await
     }
 
@@ -56,6 +74,120 @@ impl Mailbox {
     pub fn close(&mut self) {
         self.rx.close();
     }
+
+    /// Selectively receive a message matching the predicate.
+    ///
+    /// This is similar to Erlang's `receive` with pattern matching.
+    /// Messages that don't match are saved in the pending queue and
+    /// will be checked again on subsequent receives.
+    ///
+    /// # Arguments
+    ///
+    /// * `predicate` - Function that returns `Some(T)` if the message matches
+    /// * `timeout` - Optional timeout duration
+    ///
+    /// # Returns
+    ///
+    /// * `Some(T)` - A matching message was found
+    /// * `None` - Timeout expired or all senders dropped
+    pub(crate) async fn recv_matching<F, T>(
+        &mut self,
+        mut predicate: F,
+        timeout: Option<Duration>,
+    ) -> Option<T>
+    where
+        F: FnMut(&crate::message::Message) -> Option<T>,
+    {
+        use crate::message::EnvelopeContent;
+
+        // First check pending messages
+        {
+            let mut pending = self.pending.lock().unwrap();
+            for i in 0..pending.len() {
+                if let Some(envelope) = pending.get(i)
+                    && let EnvelopeContent::Message(msg) = &envelope.content
+                    && let Some(result) = predicate(msg)
+                {
+                    // Found a match - remove it and return
+                    pending.remove(i);
+                    return Some(result);
+                }
+            }
+        }
+
+        // Then check incoming messages
+        let deadline = timeout.map(|d| tokio::time::Instant::now() + d);
+
+        loop {
+            let envelope = if let Some(deadline) = deadline {
+                match tokio::time::timeout_at(deadline, self.rx.recv()).await {
+                    Ok(Some(env)) => env,
+                    Ok(None) => return None, // Channel closed
+                    Err(_) => return None,   // Timeout
+                }
+            } else {
+                self.rx.recv().await?
+            };
+
+            match envelope.content {
+                EnvelopeContent::Message(ref msg) => {
+                    if let Some(result) = predicate(msg) {
+                        // Found a match
+                        return Some(result);
+                    } else {
+                        // Doesn't match - save for later
+                        self.pending.lock().unwrap().push_back(envelope);
+                    }
+                }
+                EnvelopeContent::Signal(_) => {
+                    // Signals always go to pending - they should be processed by handle_signal
+                    self.pending.lock().unwrap().push_back(envelope);
+                }
+            }
+        }
+    }
+
+    /// Try to receive a matching message without blocking.
+    ///
+    /// Only checks the pending queue and tries one receive from the channel.
+    pub(crate) fn try_recv_matching<F, T>(&mut self, mut predicate: F) -> Option<T>
+    where
+        F: FnMut(&crate::message::Message) -> Option<T>,
+    {
+        use crate::message::EnvelopeContent;
+
+        // Check pending messages first
+        {
+            let mut pending = self.pending.lock().unwrap();
+            for i in 0..pending.len() {
+                if let Some(envelope) = pending.get(i)
+                    && let EnvelopeContent::Message(msg) = &envelope.content
+                    && let Some(result) = predicate(msg)
+                {
+                    pending.remove(i);
+                    return Some(result);
+                }
+            }
+        }
+
+        // Try one receive without blocking
+        if let Ok(envelope) = self.rx.try_recv() {
+            match envelope.content {
+                EnvelopeContent::Message(ref msg) => {
+                    if let Some(result) = predicate(msg) {
+                        return Some(result);
+                    } else {
+                        self.pending.lock().unwrap().push_back(envelope);
+                    }
+                }
+                EnvelopeContent::Signal(_) => {
+                    self.pending.lock().unwrap().push_back(envelope);
+                }
+            }
+        }
+
+        None
+    }
 }
 
 /// Handle for sending messages to an actor's mailbox.