acktor_ipc/remote_actor/
registry.rs

1use std::fmt::{self, Debug};
2use std::sync::Arc;
3
4use ahash::RandomState;
5use dashmap::{DashMap, Entry};
6
7use acktor::{Recipient, Sender, SenderId};
8
9use crate::remote_message::RemoteMessage;
10
11/// A registry of actors in the current process which can receive [`RemoteMessage`]s.
12///
13/// The registry is a cheaply-cloneable concurrent map shared between a [`Node`][crate::Node]
14/// and all of its sessions. It lets sessions route inbound [`RemoteMessage`]s to the proper
15/// actor in the same process.
16///
17/// An actor which implements the [`RemoteActor`][super::RemoteActor] trait will be
18/// automatically registered in a [`Node`][crate::Node]'s registry when its address is encoded
19/// as a [`RemoteMessage`] for the first time. Users can also manually register an actor with
20/// a [`Node`][crate::Node]'s [`AddActor`][crate::node::command::AddActor] command.
21#[derive(Clone, Default)]
22pub struct RemoteActorRegistry {
23    inner: Arc<DashMap<u64, Recipient<RemoteMessage>, RandomState>>,
24}
25
26impl Debug for RemoteActorRegistry {
27    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
28        let mut list = f.debug_list();
29        for entry in self.inner.iter() {
30            list.entry(entry.key());
31        }
32        list.finish()
33    }
34}
35
36impl RemoteActorRegistry {
37    /// Constructs a new empty [`RemoteActorRegistry`].
38    pub fn new() -> Self {
39        Self::default()
40    }
41
42    /// Constructs a new empty [`RemoteActorRegistry`] with the specified capacity.
43    pub fn with_capacity(capacity: usize) -> Self {
44        Self {
45            inner: Arc::new(DashMap::with_capacity_and_hasher(
46                capacity,
47                RandomState::new(),
48            )),
49        }
50    }
51
52    /// Returns the number of actors the registry can hold without reallocating.
53    ///
54    /// **Locking behavior:** May deadlock if called when holding a mutable reference into the
55    /// registry.
56    pub fn capacity(&self) -> usize {
57        self.inner.capacity()
58    }
59
60    /// Returns the number of actors currently in the registry.
61    ///
62    /// **Locking behavior:** May deadlock if called when holding a mutable reference into the
63    /// registry.
64    pub fn len(&self) -> usize {
65        self.inner.len()
66    }
67
68    /// Returns `true` if the registry contains no actors.
69    ///
70    /// **Locking behavior:** May deadlock if called when holding a mutable reference into the
71    /// registry.
72    pub fn is_empty(&self) -> bool {
73        self.inner.is_empty()
74    }
75
76    /// Retains only the actors for which the `predicate` returns `true`.
77    ///
78    /// **Locking behavior:** May deadlock if called when holding any sort of reference into the
79    /// registry.
80    pub fn retain<F>(&self, mut predicate: F)
81    where
82        F: FnMut(u64, &Recipient<RemoteMessage>) -> bool,
83    {
84        self.inner
85            .retain(|index, recipient| predicate(*index, recipient));
86    }
87
88    /// Returns an actor as a `Recipient<RemoteMessage>` corresponding to the given index.
89    ///
90    /// If the actor's mailbox is closed, this method removes it and returns `None`. This is
91    /// different from the behavior of `get` on a normal `HashMap`.
92    ///
93    /// **Locking behavior:** May deadlock if called when holding any sort of reference into the
94    /// registry.
95    pub fn get(&self, index: u64) -> Option<Recipient<RemoteMessage>> {
96        match self.inner.entry(index) {
97            Entry::Occupied(entry) => {
98                let recipient = entry.get();
99                if recipient.is_closed() {
100                    entry.remove();
101                    None
102                } else {
103                    Some(recipient.clone())
104                }
105            }
106            Entry::Vacant(_) => None,
107        }
108    }
109
110    /// Returns `true` if the registry contains an actor for the given index.
111    ///
112    /// **Locking behavior:** May deadlock if called when holding a mutable reference into the
113    /// registry.
114    pub fn contains_index(&self, index: u64) -> bool {
115        self.inner.contains_key(&index)
116    }
117
118    /// Removes an actor by its index, returning the actor as a `Recipient<RemoteMessage>` if it
119    /// was present in the registry.
120    ///
121    /// **Locking behavior:** May deadlock if called when holding any sort of reference into the
122    /// registry.
123    pub fn remove(&self, index: u64) -> Option<Recipient<RemoteMessage>> {
124        self.inner.remove(&index).map(|(_, recipient)| recipient)
125    }
126
127    /// Inserts a new actor keyed by its index.
128    ///
129    /// If the registry did not have this actor present, `true` is returned. If the registry did
130    /// have this actor present, `false` is returned and the registry is not modified. This is
131    /// different from the behavior of `insert` on a normal `HashMap`.
132    ///
133    /// Re-registering the same address is a cheap lookup-no-op.
134    ///
135    /// **Locking behavior:** May deadlock if called when holding any sort of reference into the
136    /// registry.
137    pub fn insert<A>(&self, actor: A) -> bool
138    where
139        A: Into<Recipient<RemoteMessage>> + SenderId,
140    {
141        let index = actor.index();
142        match self.inner.entry(index) {
143            Entry::Occupied(_) => false,
144            Entry::Vacant(entry) => {
145                entry.insert(actor.into());
146                true
147            }
148        }
149    }
150}
acktor_ipc/remote_actor/registry.rs

acktor_ipc/remote_actor/
registry.rs
