kompact/actors/
paths.rs

1use super::*;
2use crate::{
3    messaging::{DispatchData, DispatchEnvelope, MsgEnvelope, SerialisedFrame},
4    net::buffers::ChunkRef,
5};
6use std::{
7    convert::TryFrom,
8    error::Error,
9    fmt::{self, Debug},
10    net::{AddrParseError, IpAddr, SocketAddr},
11    ops::Div,
12    str::FromStr,
13};
14use uuid::Uuid;
15
16/// Transport protocol to use for delivering messages
17/// sent to an [ActorPath](ActorPath)
18///
19/// # Note
20///
21/// Dispatcher implementations are not required to implement all protocols.
22/// Check your concrete implementation, before selecting an arbitrary protocol.
23#[derive(Copy, Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
24#[repr(u8)]
25pub enum Transport {
26    /// Local reflection only, no network messages involved
27    Local = 0b00,
28    /// Send messages over TCP
29    Tcp = 0b01,
30    /// Send messages as UDP datagrams
31    Udp = 0b10,
32}
33
34impl Transport {
35    /// Returns `true` if this is an instance of [Transport::Local](Transport::Local)
36    pub fn is_local(&self) -> bool {
37        matches!(*self, Transport::Local)
38    }
39
40    /// Returns `true` if this is *not* an instance of [Transport::Local](Transport::Local)
41    pub fn is_remote(&self) -> bool {
42        !self.is_local()
43    }
44}
45
46impl fmt::Display for Transport {
47    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
48        match self {
49            &Transport::Local => write!(fmt, "local"),
50            &Transport::Tcp => write!(fmt, "tcp"),
51            &Transport::Udp => write!(fmt, "udp"),
52        }
53    }
54}
55
56impl FromStr for Transport {
57    type Err = TransportParseError;
58
59    fn from_str(s: &str) -> Result<Transport, TransportParseError> {
60        match s {
61            "local" => Ok(Transport::Local),
62            "tcp" => Ok(Transport::Tcp),
63            "udp" => Ok(Transport::Udp),
64            _ => Err(TransportParseError),
65        }
66    }
67}
68
69/// Error type for parsing the [Transport](Transport) from a string
70#[derive(Debug, PartialEq, Eq, Clone, Hash)]
71pub struct TransportParseError;
72
73impl fmt::Display for TransportParseError {
74    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
75        fmt.write_str("TransportParseError")
76    }
77}
78
79impl Error for TransportParseError {
80    fn description(&self) -> &str {
81        "Transport must be one of [local,tcp,udp]"
82    }
83}
84
85/// Error type for parsing [paths](ActorPath) from a string
86#[derive(Debug, PartialEq, Eq, Clone)]
87pub enum PathParseError {
88    /// The format is wrong
89    Form(String),
90    /// The transport protocol was invalid
91    Transport(TransportParseError),
92    /// The network address was invalid
93    Addr(AddrParseError),
94    /// The path contains a disallowed character
95    IllegalCharacter(char),
96}
97
98impl fmt::Display for PathParseError {
99    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
100        match self {
101            PathParseError::Form(s) => write!(fmt, "Invalid formatting: {}", s),
102            PathParseError::Transport(e) => write!(fmt, "Could not parse transport: {}", e),
103            PathParseError::Addr(e) => write!(fmt, "Could not parse address: {}", e),
104            PathParseError::IllegalCharacter(c) => {
105                write!(fmt, "The path contains an illegal character: {}", c)
106            }
107        }
108    }
109}
110
111impl Error for PathParseError {
112    fn description(&self) -> &str {
113        "Path could not be parsed"
114    }
115
116    fn cause(&self) -> Option<&dyn Error> {
117        match self {
118            PathParseError::Form(_) => None,
119            PathParseError::Transport(e) => Some(e),
120            PathParseError::Addr(e) => Some(e),
121            PathParseError::IllegalCharacter(_) => None,
122        }
123    }
124}
125
126impl From<TransportParseError> for PathParseError {
127    fn from(e: TransportParseError) -> PathParseError {
128        PathParseError::Transport(e)
129    }
130}
131
132impl From<AddrParseError> for PathParseError {
133    fn from(e: AddrParseError) -> PathParseError {
134        PathParseError::Addr(e)
135    }
136}
137
138/// The part of an [ActorPath](ActorPath) that refers to the [KompactSystem](KompactSystem)
139///
140/// As a URI, a `SystemPath` looks like `"tcp://127.0.0.1:8080"`, for example.
141#[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
142pub struct SystemPath {
143    protocol: Transport,
144    // TODO address could be IPv4, IPv6, or a domain name (not supported yet)
145    address: IpAddr,
146    port: u16,
147}
148
149impl SystemPath {
150    /// Construct a new system path from individual parts
151    pub fn new(protocol: Transport, address: IpAddr, port: u16) -> SystemPath {
152        SystemPath {
153            protocol,
154            address,
155            port,
156        }
157    }
158
159    /// Construct a new system path from individual parts using a [SocketAddr](std::net::SocketAddr)
160    pub fn with_socket(protocol: Transport, socket: SocketAddr) -> SystemPath {
161        SystemPath {
162            protocol,
163            address: socket.ip(),
164            port: socket.port(),
165        }
166    }
167
168    /// Returns a reference to the [Transport](Transport) protocol associated with with this system path
169    pub fn protocol(&self) -> Transport {
170        self.protocol
171    }
172
173    /// Returns a reference to the IP address associated with with this system path
174    pub fn address(&self) -> &IpAddr {
175        &self.address
176    }
177
178    /// Returns the port associated with with this system path
179    pub fn port(&self) -> u16 {
180        self.port
181    }
182
183    /// Create a named path starting with this system path and ending with the given string
184    ///
185    /// Paths created with this function will be validated to be a valid lookup path,
186    /// and an error will be returned if they are not.
187    pub fn into_named_with_string(self, path: &str) -> Result<NamedPath, PathParseError> {
188        let parsed = parse_path(path);
189        self.into_named_with_vec(parsed)
190    }
191
192    /// Returns the SocketAddr corresponding to the SystemPath
193    pub fn socket_address(&self) -> SocketAddr {
194        SocketAddr::new(self.address, self.port)
195    }
196
197    /// Create a named path starting with this system path and ending with the sequence of path segments
198    ///
199    /// Paths created with this function will be validated to be a valid lookup path,
200    /// and an error will be returned if they are not.
201    pub fn into_named_with_vec(self, path: Vec<String>) -> Result<NamedPath, PathParseError> {
202        validate_lookup_path(&path)?;
203        let named = NamedPath::with_system(self, path);
204        Ok(named)
205    }
206
207    /// Create a unique path starting with this system path and ending with the given `id`
208    pub fn into_unique(self, id: Uuid) -> UniquePath {
209        UniquePath::with_system(self, id)
210    }
211}
212
213impl fmt::Display for SystemPath {
214    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
215        write!(fmt, "{}://{}:{}", self.protocol, self.address, self.port)
216    }
217}
218
219/// Methods for things that contain a [SystemPath](SystemPath)
220pub trait SystemField {
221    /// Returns a reference to the system path
222    fn system(&self) -> &SystemPath;
223
224    /// Returns the transport protocol used in the system path
225    fn protocol(&self) -> Transport {
226        self.system().protocol()
227    }
228
229    /// Returns the address used in the system path
230    fn address(&self) -> &IpAddr {
231        self.system().address()
232    }
233
234    /// Returns the port used in the system path
235    fn port(&self) -> u16 {
236        self.system().port()
237    }
238}
239
240impl SystemField for SystemPath {
241    fn system(&self) -> &SystemPath {
242        self
243    }
244}
245
246/// A factory trait for things that have associated [actor paths](ActorPath)
247pub trait ActorPathFactory {
248    /// Returns the associated actor path
249    fn actor_path(&self) -> ActorPath;
250}
251
252/// A temporary combination of an [ActorPath](ActorPath)
253/// and something that can [dispatch](Dispatching) stuff
254///
255/// This can be used when you want to use a different actor path
256/// than the one of the current component for a [tell](ActorPath::tell)
257/// but still need to use the component's dispatcher for the message.
258/// This is useful for forwarding components, for example, when trying to
259/// preserve the original sender.
260///
261/// See also [using_dispatcher](ActorPath::using_dispatcher).
262pub struct DispatchingPath<'a, 'b> {
263    path: &'a ActorPath,
264    ctx: &'b dyn Dispatching,
265}
266
267impl Dispatching for DispatchingPath<'_, '_> {
268    fn dispatcher_ref(&self) -> DispatcherRef {
269        self.ctx.dispatcher_ref()
270    }
271}
272
273impl ActorPathFactory for DispatchingPath<'_, '_> {
274    fn actor_path(&self) -> ActorPath {
275        self.path.clone()
276    }
277}
278
279/// An actor path is a serialisable, possibly remote reference to
280/// an actor
281///
282/// Any message sent via an actor path *might* go over the network,
283/// and must be treated as fallible.
284/// It must also be [serialisable](Serialisable).
285#[derive(Clone, Debug)]
286#[repr(u8)]
287#[derive(PartialEq, Eq, Hash, PartialOrd, Ord)]
288pub enum ActorPath {
289    /// A unique actor path identifies a concrete instance of an actor
290    ///
291    /// Unique actor paths use the component's unique id internally.
292    ///
293    /// Unique actor paths become invalid when a component is
294    /// replaced with a new instance of the same type.
295    ///
296    /// A unique path may look something like `"tcp://127.0.0.1:8080#1e555f40-de1d-4aee-8202-64fdc27edfa8"`, for example.
297    Unique(UniquePath),
298    /// A named actor path identifies a service, rather than a concrete actor instance
299    ///
300    /// Named paths must be [registered](KompactSystem::register_by_alias) to a particular actor,
301    /// and their registration can be changed over time,
302    /// as actors fail and are replaced, for example.
303    ///
304    /// Named paths may be described hierarchically, similar to URLs.
305    ///
306    /// A named path may look something like `"tcp://127.0.0.1:8080/my-actor-group/my-actor"`, for example.
307    Named(NamedPath),
308}
309
310impl ActorPath {
311    /// Send message `m` to the actor designated by this path
312    ///
313    /// The `from` field is used as a source,
314    /// and the `ActorPath` it resolved to will be supplied at the destination.
315    ///
316    /// This function ensures that the protocol of the source path matches the
317    /// protocol of this path. If you would like the sender path to have a different
318    /// transport protocol, use [tell_with_sender](ActorPath::tell_with_sender) instead.
319    ///
320    /// Serialisation of `m` happens lazily in the dispatcher,
321    /// and only if it really goes over the network. If this actor path
322    /// turns out to be local, `m` will be moved to the heap and send locally instead.
323    ///
324    /// In fact, this method always moves `m` onto the heap (unless it already is allocated there),
325    /// to facilitate this lazy serialisation.
326    /// As this method has some overhead in the case where it is certain
327    /// that `m` will definitely go over the network, you can use
328    /// [tell_serialised](ActorPath::tell_serialised) to force eager serialisation instead.
329    pub fn tell<S, B>(&self, m: B, from: &S) -> ()
330    where
331        S: ActorPathFactory + Dispatching,
332        B: Into<Box<dyn Serialisable>>,
333    {
334        let mut src = from.actor_path();
335        src.set_protocol(self.protocol());
336        self.tell_with_sender(m, from, src)
337    }
338
339    /// Send message `m` to the actor designated by this path
340    ///
341    /// The `from` field is used as a source and will be supplied at the destination.
342    ///
343    /// Serialisation of `m` happens lazily in the dispatcher,
344    /// and only if it really goes over the network. If this actor path
345    /// turns out to be local, `m` will be moved to the heap and send locally instead.
346    ///
347    /// In fact, this method always moves `m` onto the heap (unless it already is allocated there),
348    /// to facilitate this lazy serialisation.
349    /// As this method has some overhead in the case where it is certain
350    /// that `m` will definitely go over the network, you can use
351    /// [tell_serialised](ActorPath::tell_serialised) to force eager serialisation instead.
352    pub fn tell_with_sender<B, D>(&self, m: B, dispatch: &D, from: ActorPath) -> ()
353    where
354        B: Into<Box<dyn Serialisable>>,
355        D: Dispatching,
356    {
357        let msg: Box<dyn Serialisable> = m.into();
358        let dst = self.clone();
359        let env = DispatchEnvelope::Msg {
360            src: from.clone(),
361            dst: dst.clone(),
362            msg: DispatchData::Lazy(msg, from, dst),
363        };
364        dispatch.dispatcher_ref().enqueue(MsgEnvelope::Typed(env))
365    }
366
367    /// Send message `m` to the actor designated by this path
368    ///
369    /// This function has the same effect as [tell](ActorPath::tell),
370    /// but serialises eagerly into a Pooled buffer (pre-allocated and bounded).
371    pub fn tell_serialised<CD, B>(&self, m: B, from: &CD) -> Result<(), SerError>
372    where
373        CD: ComponentTraits + ComponentLifecycle,
374        B: Serialisable + 'static,
375    {
376        let mut src = from.actor_path();
377        src.set_protocol(self.protocol());
378        self.tell_serialised_with_sender(m, from, src)
379    }
380
381    /// Send message `m` to the actor designated by this path
382    ///
383    /// This function has the same effect as [tell](ActorPath::tell),
384    /// but serialises eagerly into a Pooled buffer (pre-allocated and bounded).
385    pub fn tell_serialised_with_sender<CD, B>(
386        &self,
387        m: B,
388        dispatch: &CD,
389        from: ActorPath,
390    ) -> Result<(), SerError>
391    where
392        CD: ComponentTraits + ComponentLifecycle,
393        B: Serialisable + 'static,
394    {
395        if self.protocol() == Transport::Local {
396            // No need to serialize!
397            self.tell_with_sender(m, dispatch, from);
398            Ok(())
399        } else {
400            dispatch.ctx().with_buffer(|buffer| {
401                let mut buf = buffer.get_buffer_encoder()?;
402                let msg =
403                    crate::serialisation::ser_helpers::serialise_msg(&from, self, &m, &mut buf)?;
404                let env = DispatchEnvelope::Msg {
405                    src: from,
406                    dst: self.clone(),
407                    msg: DispatchData::Serialised(SerialisedFrame::ChunkLease(msg)),
408                };
409                dispatch.dispatcher_ref().enqueue(MsgEnvelope::Typed(env));
410                Ok(())
411            })
412        }
413    }
414
415    /// Send prepared message `content` to the actor designated by this path
416    ///
417    /// This function has the same effect as [tell_serialised](ActorPath::tell_serialised), but
418    /// requires the content to be pre-serialised into a [ChunkRef](net::buffers::ChunkRef).
419    ///
420    /// Use [self.ctx.preserialise(m)](component::ComponentContext#preserialise)
421    /// to preserialise data into a `ChunkRef`
422    pub fn tell_preserialised<CD>(&self, content: ChunkRef, from: &CD) -> Result<(), SerError>
423    where
424        CD: ComponentTraits + ComponentLifecycle,
425    {
426        let mut src = from.actor_path();
427        src.set_protocol(self.protocol());
428        self.tell_preserialised_with_sender(content, from, src)
429    }
430
431    /// Send preserialised message `content` to the actor designated by this path with specified `from`
432    ///
433    /// This function has the same effect as [tell_serialised](ActorPath::tell_serialised), but
434    /// requires the content to be pre-serialised into a [ChunkRef](net::buffers::ChunkRef).
435    ///
436    /// Use [self.ctx.preserialise(m)](component::ComponentContext#preserialise) to preserialise data into a `ChunkRef`
437    pub fn tell_preserialised_with_sender<CD: ComponentTraits + ComponentLifecycle>(
438        &self,
439        content: ChunkRef,
440        dispatch: &CD,
441        from: ActorPath,
442    ) -> Result<(), SerError> {
443        dispatch.ctx().with_buffer(|buffer| {
444            let mut buf = buffer.get_buffer_encoder()?;
445            let msg = crate::serialisation::ser_helpers::serialise_msg_with_preserialised(
446                &from, self, content, &mut buf,
447            )?;
448            let env = DispatchEnvelope::Msg {
449                src: from,
450                dst: self.clone(),
451                msg: DispatchData::Serialised(SerialisedFrame::ChunkRef(msg)),
452            };
453            dispatch.dispatcher_ref().enqueue(MsgEnvelope::Typed(env));
454            Ok(())
455        })
456    }
457
458    /// Forwards the still serialised message to this path without changing the sender
459    ///
460    /// This can be used for routing protocls where the final recipient is supposed to reply
461    /// to the original sender, not the intermediaries.
462    pub fn forward_with_original_sender<D>(
463        &self,
464        mut serialised_message: NetMessage,
465        dispatcher: &D,
466    ) -> ()
467    where
468        D: Dispatching,
469    {
470        serialised_message.receiver = self.clone();
471        let env = DispatchEnvelope::ForwardedMsg {
472            msg: serialised_message,
473        };
474        dispatcher.dispatcher_ref().enqueue(MsgEnvelope::Typed(env));
475    }
476
477    /// Forwards the still serialised message to this path replacing the sender with the given one
478    pub fn forward_with_sender<D>(
479        &self,
480        mut serialised_message: NetMessage,
481        dispatch: &D,
482        from: ActorPath,
483    ) -> ()
484    where
485        D: Dispatching,
486    {
487        serialised_message.receiver = self.clone();
488        serialised_message.sender = from;
489        let env = DispatchEnvelope::ForwardedMsg {
490            msg: serialised_message,
491        };
492        dispatch.dispatcher_ref().enqueue(MsgEnvelope::Typed(env));
493    }
494
495    /// Returns a temporary combination of an [ActorPath](ActorPath)
496    /// and something that can [dispatch](Dispatching) stuff
497    ///
498    /// This can be used when you want to use a different actor path
499    /// than the one of the current component for a [tell](ActorPath::tell)
500    /// but still need to use the component's dispatcher for the message.
501    /// This is useful for forwarding components, for example, when trying to
502    /// preserve the original sender.
503    pub fn using_dispatcher<'a, 'b>(
504        &'a self,
505        disp: &'b dyn Dispatching,
506    ) -> DispatchingPath<'a, 'b> {
507        DispatchingPath {
508            path: self,
509            ctx: disp,
510        }
511    }
512
513    fn system_mut(&mut self) -> &mut SystemPath {
514        match self {
515            ActorPath::Unique(ref mut up) => up.system_mut(),
516            ActorPath::Named(ref mut np) => np.system_mut(),
517        }
518    }
519
520    /// Change the transport protocol for this actor path
521    pub fn set_protocol(&mut self, proto: Transport) {
522        self.system_mut().protocol = proto;
523    }
524
525    /// Sets the transport protocol for this actor path to UDP
526    pub fn via_udp(&mut self) {
527        self.set_protocol(Transport::Udp);
528    }
529
530    /// Sets the transport protocol for this actor path to TCP
531    pub fn via_tcp(&mut self) {
532        self.set_protocol(Transport::Tcp);
533    }
534
535    /// Sets the transport protocol for this actor path to LOCAL
536    pub fn via_local(&mut self) {
537        self.set_protocol(Transport::Local);
538    }
539
540    /// If this path is a named path, return the underlying named path
541    ///
542    /// Otherwise return `None`.
543    pub fn named(self) -> Option<NamedPath> {
544        match self {
545            ActorPath::Named(p) => Some(p),
546            _ => None,
547        }
548    }
549
550    /// Return the underlying named path
551    ///
552    /// Panics if this is not actually a named path variant.
553    pub fn unwrap_named(self) -> NamedPath {
554        self.named().unwrap()
555    }
556
557    /// If this path is a unique path, return the underlying unique path
558    ///
559    /// Otherwise return `None`.
560    pub fn unique(self) -> Option<UniquePath> {
561        match self {
562            ActorPath::Unique(p) => Some(p),
563            _ => None,
564        }
565    }
566
567    /// Return the underlying unique path
568    ///
569    /// Panics if this is not actually a uniqe path variant.
570    pub fn unwrap_unique(self) -> UniquePath {
571        self.unique().unwrap()
572    }
573}
574
575impl SystemField for ActorPath {
576    fn system(&self) -> &SystemPath {
577        match self {
578            ActorPath::Unique(up) => up.system(),
579            ActorPath::Named(np) => np.system(),
580        }
581    }
582}
583
584impl From<(SystemPath, Uuid)> for ActorPath {
585    fn from(t: (SystemPath, Uuid)) -> ActorPath {
586        ActorPath::Unique(UniquePath {
587            system: t.0,
588            id: t.1,
589        })
590    }
591}
592
593impl From<UniquePath> for ActorPath {
594    fn from(p: UniquePath) -> ActorPath {
595        ActorPath::Unique(p)
596    }
597}
598
599impl From<NamedPath> for ActorPath {
600    fn from(p: NamedPath) -> ActorPath {
601        ActorPath::Named(p)
602    }
603}
604
605/// Separator for segments in named actor paths
606///
607/// # Example
608///
609/// `tcp://127.0.0.1:8080/segment1/segment2`
610pub const PATH_SEP: char = '/';
611/// Separator for the id in unique actor paths
612///
613/// # Example
614///
615/// `tcp://127.0.0.1:8080#1e555f40-de1d-4aee-8202-64fdc27edfa8`
616pub const UNIQUE_PATH_SEP: char = '#';
617/// Marker for broadcast routing
618///
619/// # Example
620///
621/// `tcp://127.0.0.1:8080/segment1/*`
622/// will route to all actors whose path starts with `segment1`.
623pub const BROADCAST_MARKER: char = '*';
624/// Marker for select routing
625///
626/// # Example
627///
628/// `tcp://127.0.0.1:8080/segment1/?`
629/// will route to some actor whose path starts with `segment1`.
630pub const SELECT_MARKER: char = '?';
631
632impl fmt::Display for ActorPath {
633    fn fmt(&self, fmt: &mut fmt::Formatter<'_>) -> fmt::Result {
634        match self {
635            ActorPath::Named(np) => {
636                let path = np.path.iter().fold(String::new(), |mut acc, arg| {
637                    acc.push(PATH_SEP);
638                    acc.push_str(arg);
639                    acc
640                });
641                write!(fmt, "{}{}", np.system, path)
642            }
643            ActorPath::Unique(up) => write!(fmt, "{}{}{}", up.system, UNIQUE_PATH_SEP, up.id),
644        }
645    }
646}
647
648impl TryFrom<String> for ActorPath {
649    type Error = PathParseError;
650
651    fn try_from(s: String) -> Result<Self, Self::Error> {
652        let p = ActorPath::from_str(&s)?;
653        Ok(p)
654    }
655}
656
657impl FromStr for ActorPath {
658    type Err = PathParseError;
659
660    fn from_str(s: &str) -> Result<Self, Self::Err> {
661        if s.contains(UNIQUE_PATH_SEP) {
662            let p = UniquePath::from_str(s)?;
663            Ok(ActorPath::Unique(p))
664        } else {
665            let p = NamedPath::from_str(s)?;
666            Ok(ActorPath::Named(p))
667        }
668    }
669}
670
671/// A unique actor path identifies a concrete instance of an actor
672///
673/// Unique actor paths use the component's unique id internally.
674///
675/// Unique actor paths become invalid when a component is
676/// replaced with a new instance of the same type.
677///
678/// A unique path may look something like `"tcp://127.0.0.1:8080#1e555f40-de1d-4aee-8202-64fdc27edfa8"`, for example.
679#[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
680pub struct UniquePath {
681    system: SystemPath,
682    id: Uuid,
683}
684
685impl UniquePath {
686    /// Construct a new unique path from parts
687    pub fn new(protocol: Transport, address: IpAddr, port: u16, id: Uuid) -> UniquePath {
688        UniquePath {
689            system: SystemPath::new(protocol, address, port),
690            id,
691        }
692    }
693
694    /// Construct a new unique path from a system path and an id
695    pub fn with_system(system: SystemPath, id: Uuid) -> UniquePath {
696        UniquePath { system, id }
697    }
698
699    /// Construct a new unique path with a [socket](std::net::SocketAddr) and an id
700    pub fn with_socket(protocol: Transport, socket: SocketAddr, id: Uuid) -> UniquePath {
701        UniquePath {
702            system: SystemPath::with_socket(protocol, socket),
703            id,
704        }
705    }
706
707    /// Returns a reference to this path's unique id
708    pub fn id(&self) -> Uuid {
709        self.id
710    }
711
712    /// Returns a mutable reference to this path's system path part
713    pub fn system_mut(&mut self) -> &mut SystemPath {
714        &mut self.system
715    }
716}
717
718impl TryFrom<String> for UniquePath {
719    type Error = PathParseError;
720
721    fn try_from(s: String) -> Result<Self, Self::Error> {
722        let p = UniquePath::from_str(&s)?;
723        Ok(p)
724    }
725}
726
727impl FromStr for UniquePath {
728    type Err = PathParseError;
729
730    fn from_str(s: &str) -> Result<Self, Self::Err> {
731        let parts: Vec<&str> = s.split("://").collect();
732        // parts: [tcp]://[IP:port#id]
733        if parts.len() != 2 {
734            return Err(PathParseError::Form(s.to_string()));
735        }
736        let proto: Transport = parts[0].parse()?;
737        let parts: Vec<&str> = parts[1].split(UNIQUE_PATH_SEP).collect();
738        // parts: [IP:port]#[UUID]
739        if parts.len() != 2 {
740            return Err(PathParseError::Form(s.to_string()));
741        }
742        let socket = SocketAddr::from_str(parts[0])?;
743        let uuid =
744            Uuid::from_str(parts[1]).map_err(|_parse_err| PathParseError::Form(s.to_string()))?;
745
746        Ok(UniquePath::with_socket(proto, socket, uuid))
747    }
748}
749
750impl SystemField for UniquePath {
751    fn system(&self) -> &SystemPath {
752        &self.system
753    }
754}
755
756/// A named actor path identifies a service, rather than a concrete actor instance
757///
758/// Named paths must be [registered](KompactSystem::register_by_alias) to a particular actor,
759/// and their registration can be changed over time,
760/// as actors fail and are replaced, for example.
761///
762/// Named paths may be described hierarchically, similar to URLs.
763///
764/// A named path may look something like `"tcp://127.0.0.1:8080/my-actor-group/my-actor"`, for example.
765#[derive(Clone, Debug, PartialEq, Eq, Hash, PartialOrd, Ord)]
766pub struct NamedPath {
767    system: SystemPath,
768    path: Vec<String>,
769}
770
771impl NamedPath {
772    /// Construct a new named path from parts
773    ///
774    /// Make sure that none of the elements of the `path` vector contain slashes (i.e., `/`),
775    /// as those will be the path separators for the individual elements later.
776    pub fn new(protocol: Transport, address: IpAddr, port: u16, path: Vec<String>) -> NamedPath {
777        debug_assert!(
778            crate::actors::validate_lookup_path(&path).is_ok(),
779            "Path contains illegal characters: {:?}",
780            path
781        );
782        NamedPath {
783            system: SystemPath::new(protocol, address, port),
784            path,
785        }
786    }
787
788    /// Construct a new named path from a [socket](std::net::SocketAddr) and vector of path elements
789    ///
790    /// Make sure that none of the elements of the `path` vector contain slashes (i.e., `/`),
791    /// as those will be the path separators for the individual elements later.
792    pub fn with_socket(protocol: Transport, socket: SocketAddr, path: Vec<String>) -> NamedPath {
793        debug_assert!(
794            crate::actors::validate_lookup_path(&path).is_ok(),
795            "Path contains illegal characters: {:?}",
796            path
797        );
798        NamedPath {
799            system: SystemPath::with_socket(protocol, socket),
800            path,
801        }
802    }
803
804    /// Construct a new named path from a system path and vector of path elements
805    ///
806    /// Make sure that none of the elements of the `path` vector contain slashes (i.e., `/`),
807    /// as those will be the path separators for the individual elements later.
808    pub fn with_system(system: SystemPath, path: Vec<String>) -> NamedPath {
809        debug_assert!(
810            crate::actors::validate_lookup_path(&path).is_ok(),
811            "Path contains illegal characters: {:?}",
812            path
813        );
814        NamedPath { system, path }
815    }
816
817    /// Returns a reference to the path vector
818    pub fn path_ref(&self) -> &[String] {
819        &self.path
820    }
821
822    /// Returns a copy to the path vector
823    pub fn clone_path(&self) -> Vec<String> {
824        self.path.clone()
825    }
826
827    /// Return just the path vector, discarding the SystemPath
828    pub fn into_path(self) -> Vec<String> {
829        self.path
830    }
831
832    /// Returns a mutable reference to this path's system path part
833    pub fn system_mut(&mut self) -> &mut SystemPath {
834        &mut self.system
835    }
836
837    /// Add the given `segment` to the path, if it is valid to do so
838    pub fn push(&mut self, segment: String) -> Result<(), PathParseError> {
839        if let Some(last_segment) = self.path.last() {
840            validate_lookup_path_segment(last_segment, false)?;
841        }
842        validate_lookup_path_segment(&segment, true)?;
843        self.path.push(segment);
844        Ok(())
845    }
846
847    /// Add the given relative `path` to this path, if it is valid to do so
848    pub fn append(mut self, path: &str) -> Result<Self, PathParseError> {
849        if let Some(last_segment) = self.path.last() {
850            validate_lookup_path_segment(last_segment, false)?;
851        }
852        let mut segments = parse_path(path);
853        self.path.append(&mut segments);
854        validate_lookup_path(&self.path)?;
855        Ok(self)
856    }
857}
858
859impl SystemField for NamedPath {
860    fn system(&self) -> &SystemPath {
861        &self.system
862    }
863}
864
865impl TryFrom<String> for NamedPath {
866    type Error = PathParseError;
867
868    fn try_from(s: String) -> Result<Self, Self::Error> {
869        NamedPath::from_str(&s)
870    }
871}
872
873impl FromStr for NamedPath {
874    type Err = PathParseError;
875
876    fn from_str(s: &str) -> Result<Self, Self::Err> {
877        let s1: Vec<&str> = s.split("://").collect();
878        if s1.len() != 2 {
879            return Err(PathParseError::Form(s.to_string()));
880        }
881        let proto: Transport = s1[0].parse()?;
882        let mut s2: Vec<&str> = s1[1].split(PATH_SEP).collect();
883        if s2.is_empty() {
884            return Err(PathParseError::Form(s.to_string()));
885        }
886        let socket = SocketAddr::from_str(s2[0])?;
887        let path: Vec<String> = if s2.len() > 1 {
888            s2.split_off(1).into_iter().map(|v| v.to_string()).collect()
889        } else {
890            Vec::default()
891        };
892        validate_lookup_path(&path)?;
893        Ok(NamedPath::with_socket(proto, socket, path))
894    }
895}
896
897/// Some syntactic sugar for [append](NamedPath::append)
898///
899/// Allows the pretty `path / "segment" / "*"` syntax
900/// to create sub-paths of `path`.
901impl Div<&str> for NamedPath {
902    type Output = Self;
903
904    fn div(self, rhs: &str) -> Self::Output {
905        self.append(rhs).expect("illegal path")
906    }
907}
908/// Some syntactic sugar for [push](NamedPath::push)
909///
910/// Allows the pretty `path / '*'` syntax
911/// to create a broadcast variant of `path`.
912impl Div<char> for NamedPath {
913    type Output = Self;
914
915    fn div(mut self, rhs: char) -> Self::Output {
916        self.push(rhs.to_string()).expect("illegal path");
917        self
918    }
919}
920
921/// Split the `&str` into segments using [PATH_SEP](crate::constants::PATH_SEP)
922pub fn parse_path(s: &str) -> Vec<String> {
923    s.split(PATH_SEP).map(|v| v.to_string()).collect()
924}
925
926/// Check that the given path is valid as a lookup path
927///
928/// Lookup paths must not contain the following characters:
929/// - [PATH_SEP](crate::constants::PATH_SEP)
930/// - [UNIQUE_PATH_SEP](crate::constants::UNIQUE_PATH_SEP)
931/// - [BROADCAST_MARKER](crate::constants::BROADCAST_MARKER) unless as the only item in the last string
932/// - [SELECT_MARKER](crate::constants::SELECT_MARKER) unless as the only item in the last string
933pub fn validate_lookup_path(path: &[String]) -> Result<(), PathParseError> {
934    let len = path.len();
935    for (index, segment) in path.iter().enumerate() {
936        validate_lookup_path_segment(segment, (index + 1) == len)?;
937    }
938    Ok(())
939}
940
941/// Check that the given path is valid as an insert path
942///
943/// Insert paths must not contain the following characters:
944/// - [PATH_SEP](crate::constants::PATH_SEP)
945/// - [UNIQUE_PATH_SEP](crate::constants::UNIQUE_PATH_SEP)
946/// - [BROADCAST_MARKER](crate::constants::BROADCAST_MARKER)
947/// - [SELECT_MARKER](crate::constants::SELECT_MARKER)
948pub fn validate_insert_path(path: &[String]) -> Result<(), PathParseError> {
949    for segment in path.iter() {
950        validate_insert_path_segment(segment)?;
951    }
952    Ok(())
953}
954
955pub(crate) fn validate_lookup_path_segment(
956    segment: &str,
957    is_last: bool,
958) -> Result<(), PathParseError> {
959    if segment.is_empty() {
960        return Err(PathParseError::Form(
961            "Path segments may not be empty!".to_string(),
962        ));
963    }
964    let len = segment.len();
965    for c in segment.chars() {
966        match c {
967            PATH_SEP => return Err(PathParseError::IllegalCharacter(PATH_SEP)),
968            BROADCAST_MARKER if !is_last && len == 1 => {
969                return Err(PathParseError::IllegalCharacter(BROADCAST_MARKER))
970            }
971            SELECT_MARKER if !is_last && len == 1 => {
972                return Err(PathParseError::IllegalCharacter(SELECT_MARKER))
973            }
974            UNIQUE_PATH_SEP => return Err(PathParseError::IllegalCharacter(UNIQUE_PATH_SEP)),
975            _ => (), // ok
976        }
977    }
978    Ok(())
979}
980pub(crate) fn validate_insert_path_segment(segment: &str) -> Result<(), PathParseError> {
981    if segment.is_empty() {
982        return Err(PathParseError::Form(
983            "Path segments may not be empty!".to_string(),
984        ));
985    }
986    for c in segment.chars() {
987        match c {
988            PATH_SEP => return Err(PathParseError::IllegalCharacter(PATH_SEP)),
989            BROADCAST_MARKER => return Err(PathParseError::IllegalCharacter(BROADCAST_MARKER)),
990            SELECT_MARKER => return Err(PathParseError::IllegalCharacter(SELECT_MARKER)),
991            UNIQUE_PATH_SEP => return Err(PathParseError::IllegalCharacter(UNIQUE_PATH_SEP)),
992            _ => (), // ok
993        }
994    }
995    Ok(())
996}
997
998#[cfg(test)]
999mod tests {
1000    use super::*;
1001
1002    const PATH: &str = "local://127.0.0.1:0/test_actor";
1003
1004    #[test]
1005    fn actor_path_strings() {
1006        let ap = ActorPath::from_str(PATH).expect("a proper path");
1007        println!("Got ActorPath={}", ap);
1008
1009        let s = ap.to_string();
1010        assert_eq!(PATH, &s);
1011        let ap2: ActorPath = s.parse().expect("a proper path");
1012        assert_eq!(ap, ap2);
1013    }
1014
1015    #[test]
1016    fn actor_path_unique_strings() {
1017        let ref1 = ActorPath::Unique(UniquePath::new(
1018            Transport::Local,
1019            "127.0.0.1".parse().expect("hardcoded IP"),
1020            8080,
1021            Uuid::new_v4(),
1022        ));
1023        let ref1_string = ref1.to_string();
1024        let ref1_deser = ActorPath::from_str(&ref1_string).expect("a proper path");
1025        let ref1_deser2: ActorPath = ref1_string.parse().expect("a proper path");
1026        assert_eq!(ref1, ref1_deser);
1027        assert_eq!(ref1, ref1_deser2);
1028    }
1029
1030    #[test]
1031    fn actor_path_named_strings() {
1032        let ref1 = ActorPath::Named(NamedPath::new(
1033            Transport::Local,
1034            "127.0.0.1".parse().expect("hardcoded IP"),
1035            8080,
1036            vec!["test".to_string(), "path".to_string()],
1037        ));
1038        let ref1_string = ref1.to_string();
1039        let ref1_deser = ActorPath::from_str(&ref1_string).expect("a proper path");
1040        let ref1_deser2: ActorPath = ref1_string.parse().expect("a proper path");
1041        assert_eq!(ref1, ref1_deser);
1042        assert_eq!(ref1, ref1_deser2);
1043    }
1044}
kompact/actors/paths.rs

kompact/actors/
paths.rs
