pub fn from_options(opts: NatsClientOptions) -> impl Future<Item = Self, Error = NatsError> + Send + Sync {
        let tls_required = opts.connect_command.tls_required;

        let cluster_uri = opts.cluster_uri.clone();
        let cluster_sa = if let Ok(sockaddr) = SocketAddr::from_str(&cluster_uri) {
            Ok(sockaddr)
        } else {
            match cluster_uri.to_socket_addrs() {
                Ok(mut ips_iter) => ips_iter.next().ok_or(NatsError::UriDNSResolveError(None)),
                Err(e) => Err(NatsError::UriDNSResolveError(Some(e))),
            }
        };

        future::result(cluster_sa)
            .from_err()
            .and_then(move |cluster_sa| {
                if tls_required {
                    match Url::parse(&cluster_uri) {
                        Ok(url) => match url.host_str() {
                            Some(host) => future::ok(Either::B(connect_tls(host.to_string(), cluster_sa))),
                            None => future::err(NatsError::TlsHostMissingError),
                        },
                        Err(e) => future::err(e.into()),
                    }
                } else {
                    future::ok(Either::A(connect(cluster_sa)))
                }
            }).and_then(|either| either)
            .and_then(move |connection| {
                let (sink, stream): (NatsSink, NatsStream) = connection.split();
                let (rx, other_rx) = NatsClientMultiplexer::new(stream);
                let tx = NatsClientSender::new(sink);

                let (tmp_other_tx, tmp_other_rx) = mpsc::unbounded();
                let tx_inner = tx.clone();
                let client = NatsClient {
                    tx,
                    server_info: Arc::new(RwLock::new(None)),
                    other_rx: Box::new(tmp_other_rx.map_err(|_| NatsError::InnerBrokenChain)),
                    rx: Arc::new(rx),
                    opts,
                };

                let server_info_arc = Arc::clone(&client.server_info);

                tokio_executor::spawn(
                    other_rx
                        .for_each(move |op| {
                            match op {
                                Op::PING => {
                                    tokio_executor::spawn(tx_inner.send(Op::PONG).map_err(|_| ()));
                                    let _ = tmp_other_tx.unbounded_send(op);
                                }
                                Op::INFO(server_info) => {
                                    *server_info_arc.write() = Some(server_info);
                                }
                                op => {
                                    let _ = tmp_other_tx.unbounded_send(op);
                                }
                            }

                            future::ok(())
                        }).into_future()
                        .map_err(|_| ()),
                );

                future::ok(client)
            })
    }
/// Creates a client and initiates a connection to the server
///
/// Returns `impl Future<Item = Self, Error = NatsError>`
Creates a client and initiates a connection to the server
Returns `impl Future`
Creates a client and initiates a connection to the server Returns ` impl Future `
pub fn from_options(opts: NatsClientOptions) -> impl Future<Item = Self, Error = NatsError> + Send + Sync {
        let tls_required = opts.connect_command.tls_required;
        let cluster_uri = opts.cluster_uri.clone();
        let cluster_sa = if let Ok(sockaddr) = SocketAddr::from_str(&cluster_uri) {
            Ok(sockaddr)
        } else {
            match cluster_uri.to_socket_addrs() {
                Ok(mut ips_iter) => ips_iter.next().ok_or(NatsError::UriDNSResolveError(None)),
                Err(e) => Err(NatsError::UriDNSResolveError(Some(e))),
            }
        };
        future::result(cluster_sa)
            .from_err()
            .and_then(move |cluster_sa| {
                if tls_required {
                    match Url::parse(&cluster_uri) {
                        Ok(url) => match url.host_str() {
                            Some(host) => future::ok(Either::B(connect_tls(host.to_string(), cluster_sa))),
                            None => future::err(NatsError::TlsHostMissingError),
                        },
                        Err(e) => future::err(e.into()),
                    }
                } else {
                    future::ok(Either::A(connect(cluster_sa)))
                }
            }).and_then(|either| either)
            .and_then(move |connection| {
                let (sink, stream): (NatsSink, NatsStream) = connection.split();
                let (rx, other_rx) = NatsClientMultiplexer::new(stream);
                let tx = NatsClientSender::new(sink);
                let (tmp_other_tx, tmp_other_rx) = mpsc::unbounded();
                let tx_inner = tx.clone();
                let client = NatsClient {
                    tx,
                    server_info: Arc::new(RwLock::new(None)),
                    other_rx: Box::new(tmp_other_rx.map_err(|_| NatsError::InnerBrokenChain)),
                    rx: Arc::new(rx),
                    opts,
                };
                let server_info_arc = Arc::clone(&client.server_info);
                tokio_executor::spawn(
                    other_rx
                        .for_each(move |op| {
                            match op {
                                Op::PING => {
                                    tokio_executor::spawn(tx_inner.send(Op::PONG).map_err(|_| ()));
                                    let _ = tmp_other_tx.unbounded_send(op);
                                }
                                Op::INFO(server_info) => {
                                    *server_info_arc.write() = Some(server_info);
                                }
                                op => {
                                    let _ = tmp_other_tx.unbounded_send(op);
                                }
                            }
                            future::ok(())
                        }).into_future()
                        .map_err(|_| ()),
                );
                future::ok(client)
            })
    }
pub fn from_options ( opts : NatsClientOptions ) -> impl Future < Item = Self , Error = NatsError > + Send + Sync { let tls_required = opts . connect_command . tls_required ; let cluster_uri = opts . cluster_uri . clone ( ) ; let cluster_sa = if let Ok ( sockaddr ) = SocketAddr :: from_str ( & cluster_uri ) { Ok ( sockaddr ) } else { match cluster_uri . to_socket_addrs ( ) { Ok ( mut ips_iter ) => ips_iter . next ( ) . ok_or ( NatsError :: UriDNSResolveError ( None ) ) , Err ( e ) => Err ( NatsError :: UriDNSResolveError ( Some ( e ) ) ) , } } ; future :: result ( cluster_sa ) . from_err ( ) . and_then ( move | cluster_sa | { if tls_required { match Url :: parse ( & cluster_uri ) { Ok ( url ) => match url . host_str ( ) { Some ( host ) => future :: ok ( Either :: B ( connect_tls ( host . to_string ( ) , cluster_sa ) ) ) , None => future :: err ( NatsError :: TlsHostMissingError ) , } , Err ( e ) => future :: err ( e . into ( ) ) , } } else { future :: ok ( Either :: A ( connect ( cluster_sa ) ) ) } } ) . and_then ( | either | either ) . and_then ( move | connection | { let ( sink , stream ) : ( NatsSink , NatsStream ) = connection . split ( ) ; let ( rx , other_rx ) = NatsClientMultiplexer :: new ( stream ) ; let tx = NatsClientSender :: new ( sink ) ; let ( tmp_other_tx , tmp_other_rx ) = mpsc :: unbounded ( ) ; let tx_inner = tx . clone ( ) ; let client = NatsClient { tx , server_info : Arc :: new ( RwLock :: new ( None ) ) , other_rx : Box :: new ( tmp_other_rx . map_err ( | _ | NatsError :: InnerBrokenChain ) ) , rx : Arc :: new ( rx ) , opts , } ; let server_info_arc = Arc :: clone ( & client . server_info ) ; tokio_executor :: spawn ( other_rx . for_each ( move | op | { match op { Op :: PING => { tokio_executor :: spawn ( tx_inner . send ( Op :: PONG ) . map_err ( | _ | ( ) ) ) ; let _ = tmp_other_tx . unbounded_send ( op ) ; } Op :: INFO ( server_info ) => { * server_info_arc . write ( ) = Some ( server_info ) ; } op => { let _ = tmp_other_tx . unbounded_send ( op ) ; } } future :: ok ( ( ) ) } ) . into_future ( ) . map_err ( | _ | ( ) ) , ) ; future :: ok ( client ) } ) }
Creates a client and initiates a connection to the server
Returns `impl Future<Item = Self, Error = NatsError>`
Creates a client and initiates a connection to the server Returns ` impl Future<Item = Self Error = NatsError > `
opts
NatsClientOptions
opts
NatsClientOptions