pub fn subscribe(
        &self,
        cmd: SubCommand,
    ) -> impl Future<Item = impl Stream<Item = Message, Error = NatsError> + Send + Sync, Error = NatsError> + Send + Sync
    {
        let inner_rx = self.rx.clone();
        let sid = cmd.sid.clone();
        self.tx.send(Op::SUB(cmd)).and_then(move |_| {
            let stream = inner_rx.for_sid(sid.clone()).and_then(move |msg| {
                {
                    let mut stx = inner_rx.subs_tx.write();
                    let mut delete = None;
                    debug!(target: "nitox", "Retrieving sink for sid {:?}", sid);
                    if let Some(s) = stx.get_mut(&sid) {
                        debug!(target: "nitox", "Checking if count exists");
                        if let Some(max_count) = s.max_count {
                            s.count += 1;
                            debug!(target: "nitox", "Max: {} / current: {}", max_count, s.count);
                            if s.count >= max_count {
                                debug!(target: "nitox", "Starting deletion");
                                delete = Some(max_count);
                            }
                        }
                    }

                    if let Some(count) = delete.take() {
                        debug!(target: "nitox", "Deleted stream for sid {} at count {}", sid, count);
                        stx.remove(&sid);
                        return Err(NatsError::SubscriptionReachedMaxMsgs(count));
                    }
                }

                Ok(msg)
            });

            future::ok(stream)
        })
    }
/// Send a SUB command and register subscription stream in the multiplexer and return that `Stream` in a future
///
/// Returns `impl Future<Item = impl Stream<Item = Message, Error = NatsError>>`
Send a SUB command and register subscription stream in the multiplexer and return that `Stream` in a future
Returns `impl Future>`
Send a SUB command and register subscription stream in the multiplexer and return that ` Stream ` in a future Returns ` impl Future > `
pub fn subscribe(
        &self,
        cmd: SubCommand,
    ) -> impl Future<Item = impl Stream<Item = Message, Error = NatsError> + Send + Sync, Error = NatsError> + Send + Sync
    {
        let inner_rx = self.rx.clone();
        let sid = cmd.sid.clone();
        self.tx.send(Op::SUB(cmd)).and_then(move |_| {
            let stream = inner_rx.for_sid(sid.clone()).and_then(move |msg| {
                {
                    let mut stx = inner_rx.subs_tx.write();
                    let mut delete = None;
                    debug!(target: "nitox", "Retrieving sink for sid {:?}", sid);
                    if let Some(s) = stx.get_mut(&sid) {
                        debug!(target: "nitox", "Checking if count exists");
                        if let Some(max_count) = s.max_count {
                            s.count += 1;
                            debug!(target: "nitox", "Max: {} / current: {}", max_count, s.count);
                            if s.count >= max_count {
                                debug!(target: "nitox", "Starting deletion");
                                delete = Some(max_count);
                            }
                        }
                    }
                    if let Some(count) = delete.take() {
                        debug!(target: "nitox", "Deleted stream for sid {} at count {}", sid, count);
                        stx.remove(&sid);
                        return Err(NatsError::SubscriptionReachedMaxMsgs(count));
                    }
                }
                Ok(msg)
            });
            future::ok(stream)
        })
    }
pub fn subscribe ( & self , cmd : SubCommand , ) -> impl Future < Item = impl Stream < Item = Message , Error = NatsError > + Send + Sync , Error = NatsError > + Send + Sync { let inner_rx = self . rx . clone ( ) ; let sid = cmd . sid . clone ( ) ; self . tx . send ( Op :: SUB ( cmd ) ) . and_then ( move | _ | { let stream = inner_rx . for_sid ( sid . clone ( ) ) . and_then ( move | msg | { { let mut stx = inner_rx . subs_tx . write ( ) ; let mut delete = None ; debug ! ( target " " " " sid ) ; if let Some ( s ) = stx . get_mut ( & sid ) { debug ! ( target " " " " ) ; if let Some ( max_count ) = s . max_count { s . count += 1 ; debug ! ( target " " " " max_count s count ) ; if s . count >= max_count { debug ! ( target " " " " ) ; delete = Some ( max_count ) ; } } } if let Some ( count ) = delete . take ( ) { debug ! ( target " " " " sid count ) ; stx . remove ( & sid ) ; return Err ( NatsError :: SubscriptionReachedMaxMsgs ( count ) ) ; } } Ok ( msg ) } ) ; future :: ok ( stream ) } ) }
Send a SUB command and register subscription stream in the multiplexer and return that `Stream` in a future
Returns `impl Future<Item = impl Stream<Item = Message, Error = NatsError>>`
Send a SUB command and register subscription stream in the multiplexer and return that ` Stream ` in a future Returns ` impl Future<Item = impl Stream<Item = Message Error = NatsError >> `
&self
cmd
SubCommand
&self
cmd
SubCommand