pub fn subscribe_to_topic<T>(
        &mut self,
        topic: &str,
        callback: PubSubCallback<T>,
    ) -> Result<(), IoError>
    where
        T: Debug + DeserializeOwned + Serialize + 'static,
    {
        if self.subscribers.contains_key(topic) {
            return Err(IoError::new(
                ErrorKind::AlreadyExists,
                format!("Already subscribed to topic: {}", topic),
            ));
        }

        let topic_config = match self.config.get_pubsub_endpoint(topic, &self.name) {
            Some(config) => config,
            None => {
                return Err(IoError::new(
                    ErrorKind::NotFound,
                    format!("PubSub topic not in config: {}", topic),
                ))
            }
        };

        let topic_name = format!("{}.{}", topic_config.name(), topic);
        match topic_config.config() {
            PubSubKind::Udp(addr) => {
                if addr.ip() != IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)) {
                    let msg = format!(
                        "PubSub topic {} refers to remote endpoint: {}",
                        Process::config_name(&self.name, topic),
                        addr
                    );
                    return Err(IoError::new(ErrorKind::AddrNotAvailable, msg));
                }
                assert!(self
                    .subscribers
                    .insert(
                        String::from(topic),
                        Subscriber::with_udp_port(&topic_name, addr.port(), callback),
                    )
                    .is_none());
            }
            PubSubKind::Unix => {
                let socket_path = self.unix_socket_base()?.with_extension(topic);
                assert!(self
                    .subscribers
                    .insert(
                        String::from(topic),
                        Subscriber::with_unix_datagram(&topic_name, &socket_path, callback),
                    )
                    .is_none());
            }
        }

        log::trace!("Subscribed to topic: {}", topic_name);
        Ok(())
    }
/// Subscribe to a topic.
///
/// The topic must be defined in the `Process` config and there must be an endpoint listed for
/// the `Process` name. If the endpoint is a UDP endpoint, it must be a local address.
/// Additionally, if the topic has already been subscribed to, an error will be returned.
///
/// Args:
/// * `topic`: The PubSub topic to subscribe to.
/// * `callback`: The function that is called when the PubSub subscriber receives data.
Subscribe to a topic.
The topic must be defined in the `Process` config and there must be an endpoint listed for
the `Process` name. If the endpoint is a UDP endpoint, it must be a local address.
Additionally, if the topic has already been subscribed to, an error will be returned.

`topic`: The PubSub topic to subscribe to.
`callback`: The function that is called when the PubSub subscriber receives data.
Subscribe to a topic . The topic must be defined in the ` Process ` config and there must be an endpoint listed for the ` Process ` name . If the endpoint is a UDP endpoint it must be a local address . Additionally if the topic has already been subscribed to an error will be returned . ` topic ` : The PubSub topic to subscribe to . ` callback ` : The function that is called when the PubSub subscriber receives data .
pub fn subscribe_to_topic<T>(
        &mut self,
        topic: &str,
        callback: PubSubCallback<T>,
    ) -> Result<(), IoError>
    where
        T: Debug + DeserializeOwned + Serialize + 'static,
    {
        if self.subscribers.contains_key(topic) {
            return Err(IoError::new(
                ErrorKind::AlreadyExists,
                format!("Already subscribed to topic: {}", topic),
            ));
        }
        let topic_config = match self.config.get_pubsub_endpoint(topic, &self.name) {
            Some(config) => config,
            None => {
                return Err(IoError::new(
                    ErrorKind::NotFound,
                    format!("PubSub topic not in config: {}", topic),
                ))
            }
        };
        let topic_name = format!("{}.{}", topic_config.name(), topic);
        match topic_config.config() {
            PubSubKind::Udp(addr) => {
                if addr.ip() != IpAddr::V4(Ipv4Addr::new(127, 0, 0, 1)) {
                    let msg = format!(
                        "PubSub topic {} refers to remote endpoint: {}",
                        Process::config_name(&self.name, topic),
                        addr
                    );
                    return Err(IoError::new(ErrorKind::AddrNotAvailable, msg));
                }
                assert!(self
                    .subscribers
                    .insert(
                        String::from(topic),
                        Subscriber::with_udp_port(&topic_name, addr.port(), callback),
                    )
                    .is_none());
            }
            PubSubKind::Unix => {
                let socket_path = self.unix_socket_base()?.with_extension(topic);
                assert!(self
                    .subscribers
                    .insert(
                        String::from(topic),
                        Subscriber::with_unix_datagram(&topic_name, &socket_path, callback),
                    )
                    .is_none());
            }
        }
        log::trace!("Subscribed to topic: {}", topic_name);
        Ok(())
    }
pub fn subscribe_to_topic < T > ( & mut self , topic : & str , callback : PubSubCallback < T > , ) -> Result < ( ) , IoError > where T : Debug + DeserializeOwned + Serialize + ' static , { if self . subscribers . contains_key ( topic ) { return Err ( IoError :: new ( ErrorKind :: AlreadyExists , format ! ( " " topic ) , ) ) ; } let topic_config = match self . config . get_pubsub_endpoint ( topic , & self . name ) { Some ( config ) => config , None => { return Err ( IoError :: new ( ErrorKind :: NotFound , format ! ( " " topic ) , ) ) } } ; let topic_name = format ! ( " " topic_config name ( ) topic ) ; match topic_config . config ( ) { PubSubKind :: Udp ( addr ) => { if addr . ip ( ) != IpAddr :: V4 ( Ipv4Addr :: new ( 127 , 0 , 0 , 1 ) ) { let msg = format ! ( " " Process config_name ( self name topic ) addr ) ; return Err ( IoError :: new ( ErrorKind :: AddrNotAvailable , msg ) ) ; } assert ! ( self subscribers insert ( String from ( topic ) Subscriber with_udp_port ( topic_name addr port ( ) callback ) ) is_none ( ) ) ; } PubSubKind :: Unix => { let socket_path = self . unix_socket_base ( ) ? . with_extension ( topic ) ; assert ! ( self subscribers insert ( String from ( topic ) Subscriber with_unix_datagram ( topic_name socket_path callback ) ) is_none ( ) ) ; } } log :: trace ! ( " " topic_name ) ; Ok ( ( ) ) }
Subscribe to a topic.
Subscribe to a topic .
&mut self
topic
&str
callback
PubSubCallback<T>
&mut self
topic
&str
callback
PubSubCallback<T>