pub fn request(
        &self,
        subject: String,
        payload: Bytes,
    ) -> impl Future<Item = Message, Error = NatsError> + Send + Sync {
        if let Some(ref server_info) = *self.server_info.read() {
            if payload.len() > server_info.max_payload as usize {
                return Either::A(future::err(NatsError::MaxPayloadOverflow(server_info.max_payload)));
            }
        }

        let inbox = PubCommand::generate_reply_to();
        let pub_cmd = PubCommand {
            subject,
            payload,
            reply_to: Some(inbox.clone()),
        };

        let sub_cmd = SubCommand {
            queue_group: None,
            sid: SubCommand::generate_sid(),
            subject: inbox,
        };

        let sid = sub_cmd.sid.clone();

        let unsub_cmd = UnsubCommand {
            sid: sub_cmd.sid.clone(),
            max_msgs: Some(1),
        };

        let tx1 = self.tx.clone();
        let tx2 = self.tx.clone();
        let rx_arc = Arc::clone(&self.rx);

        let stream = self
            .rx
            .for_sid(sid.clone())
            .inspect(|msg| debug!(target: "nitox", "Request saw msg in multiplexed stream {:#?}", msg))
            .take(1)
            .into_future()
            .map(|(surely_message, _)| surely_message.unwrap())
            .map_err(|(e, _)| e)
            .and_then(move |msg| {
                rx_arc.remove_sid(&sid);
                future::ok(msg)
            });

        Either::B(
            self.tx
                .send(Op::SUB(sub_cmd))
                .and_then(move |_| tx1.send(Op::UNSUB(unsub_cmd)))
                .and_then(move |_| tx2.send(Op::PUB(pub_cmd)))
                .and_then(move |_| stream),
        )
    }
/// Performs a request to the server following the Request/Reply pattern. Returns a future containing the MSG that will be replied at some point by a third party
///
/// Returns `impl Future<Item = Message, Error = NatsError>`
Performs a request to the server following the Request/Reply pattern. Returns a future containing the MSG that will be replied at some point by a third party
Returns `impl Future`
Performs a request to the server following the Request / Reply pattern . Returns a future containing the MSG that will be replied at some point by a third party Returns ` impl Future `
pub fn request(
        &self,
        subject: String,
        payload: Bytes,
    ) -> impl Future<Item = Message, Error = NatsError> + Send + Sync {
        if let Some(ref server_info) = *self.server_info.read() {
            if payload.len() > server_info.max_payload as usize {
                return Either::A(future::err(NatsError::MaxPayloadOverflow(server_info.max_payload)));
            }
        }
        let inbox = PubCommand::generate_reply_to();
        let pub_cmd = PubCommand {
            subject,
            payload,
            reply_to: Some(inbox.clone()),
        };
        let sub_cmd = SubCommand {
            queue_group: None,
            sid: SubCommand::generate_sid(),
            subject: inbox,
        };
        let sid = sub_cmd.sid.clone();
        let unsub_cmd = UnsubCommand {
            sid: sub_cmd.sid.clone(),
            max_msgs: Some(1),
        };
        let tx1 = self.tx.clone();
        let tx2 = self.tx.clone();
        let rx_arc = Arc::clone(&self.rx);
        let stream = self
            .rx
            .for_sid(sid.clone())
            .inspect(|msg| debug!(target: "nitox", "Request saw msg in multiplexed stream {:#?}", msg))
            .take(1)
            .into_future()
            .map(|(surely_message, _)| surely_message.unwrap())
            .map_err(|(e, _)| e)
            .and_then(move |msg| {
                rx_arc.remove_sid(&sid);
                future::ok(msg)
            });
        Either::B(
            self.tx
                .send(Op::SUB(sub_cmd))
                .and_then(move |_| tx1.send(Op::UNSUB(unsub_cmd)))
                .and_then(move |_| tx2.send(Op::PUB(pub_cmd)))
                .and_then(move |_| stream),
        )
    }
pub fn request ( & self , subject : String , payload : Bytes , ) -> impl Future < Item = Message , Error = NatsError > + Send + Sync { if let Some ( ref server_info ) = * self . server_info . read ( ) { if payload . len ( ) > server_info . max_payload as usize { return Either :: A ( future :: err ( NatsError :: MaxPayloadOverflow ( server_info . max_payload ) ) ) ; } } let inbox = PubCommand :: generate_reply_to ( ) ; let pub_cmd = PubCommand { subject , payload , reply_to : Some ( inbox . clone ( ) ) , } ; let sub_cmd = SubCommand { queue_group : None , sid : SubCommand :: generate_sid ( ) , subject : inbox , } ; let sid = sub_cmd . sid . clone ( ) ; let unsub_cmd = UnsubCommand { sid : sub_cmd . sid . clone ( ) , max_msgs : Some ( 1 ) , } ; let tx1 = self . tx . clone ( ) ; let tx2 = self . tx . clone ( ) ; let rx_arc = Arc :: clone ( & self . rx ) ; let stream = self . rx . for_sid ( sid . clone ( ) ) . inspect ( | msg | debug ! ( target " " " " msg ) ) . take ( 1 ) . into_future ( ) . map ( | ( surely_message , _ ) | surely_message . unwrap ( ) ) . map_err ( | ( e , _ ) | e ) . and_then ( move | msg | { rx_arc . remove_sid ( & sid ) ; future :: ok ( msg ) } ) ; Either :: B ( self . tx . send ( Op :: SUB ( sub_cmd ) ) . and_then ( move | _ | tx1 . send ( Op :: UNSUB ( unsub_cmd ) ) ) . and_then ( move | _ | tx2 . send ( Op :: PUB ( pub_cmd ) ) ) . and_then ( move | _ | stream ) , ) }
Performs a request to the server following the Request/Reply pattern.
Performs a request to the server following the Request / Reply pattern .
&self
subject
String
payload
Bytes
&self
subject
String
payload
Bytes