pub(in crate::nats_client) async fn start(self_arc: Arc<Self>, version: u128, mut stream: SplitStream<NatsTcpStream>) -> Result<(), RatsioError> {
        let opts = self_arc.opts.clone();
        //Register for NATS incoming messages
        //let mut executor = futures::executor::LocalPool::new();
        let stream_self = self_arc.clone();
        let _ = tokio::spawn(async move {
            while let Some(item) = stream.next().await {
                let current_version = stream_self.reconnect_version.read().await;
                if *current_version != version {
                    break;
                }
                stream_self.process_nats_event(item).await
            }
        });
        //executor.run();
        let connect = Op::CONNECT(Connect {
            verbose: opts.verbose,
            pedantic: opts.pedantic,
            tls_required: opts.tls_required,
            auth_token: Some(opts.auth_token).filter(|a| !a.is_empty()),
            user: Some(opts.username).filter(|a| !a.is_empty()),
            pass: Some(opts.password).filter(|a| !a.is_empty()),
            name: Some(opts.name).filter(|a| !a.is_empty()),
            lang: "rust".to_string(),
            version: "0.3.0".to_string(),
            protocol: 1,
            echo: false,
            sig: None,
            jwt: None,
            nkey: None,
        });
        self_arc.send_command(connect).await?;
        let mut state_guard = self_arc.state.write().await;
        *state_guard = NatsClientState::Connected;
        Ok(())
    }
// Issue a connect command to NATS
Issue a connect command to NATS
Issue a connect command to NATS
pub(in crate::nats_client) async fn start(self_arc: Arc<Self>, version: u128, mut stream: SplitStream<NatsTcpStream>) -> Result<(), RatsioError> {
        let opts = self_arc.opts.clone();
        //Register for NATS incoming messages
        //let mut executor = futures::executor::LocalPool::new();
        let stream_self = self_arc.clone();
        let _ = tokio::spawn(async move {
            while let Some(item) = stream.next().await {
                let current_version = stream_self.reconnect_version.read().await;
                if *current_version != version {
                    break;
                }
                stream_self.process_nats_event(item).await
            }
        });
        //executor.run();
        let connect = Op::CONNECT(Connect {
            verbose: opts.verbose,
            pedantic: opts.pedantic,
            tls_required: opts.tls_required,
            auth_token: Some(opts.auth_token).filter(|a| !a.is_empty()),
            user: Some(opts.username).filter(|a| !a.is_empty()),
            pass: Some(opts.password).filter(|a| !a.is_empty()),
            name: Some(opts.name).filter(|a| !a.is_empty()),
            lang: "rust".to_string(),
            version: "0.3.0".to_string(),
            protocol: 1,
            echo: false,
            sig: None,
            jwt: None,
            nkey: None,
        });
        self_arc.send_command(connect).await?;
        let mut state_guard = self_arc.state.write().await;
        *state_guard = NatsClientState::Connected;
        Ok(())
    }
pub ( in crate :: nats_client ) async fn start ( self_arc : Arc < Self > , version : u128 , mut stream : SplitStream < NatsTcpStream > ) -> Result < ( ) , RatsioError > { let opts = self_arc . opts . clone ( ) ; //Register for NATS incoming messages //let mut executor = futures::executor::LocalPool::new(); let stream_self = self_arc . clone ( ) ; let _ = tokio :: spawn ( async move { while let Some ( item ) = stream . next ( ) . await { let current_version = stream_self . reconnect_version . read ( ) . await ; if * current_version != version { break ; } stream_self . process_nats_event ( item ) . await } } ) ; //executor.run(); let connect = Op :: CONNECT ( Connect { verbose : opts . verbose , pedantic : opts . pedantic , tls_required : opts . tls_required , auth_token : Some ( opts . auth_token ) . filter ( | a | ! a . is_empty ( ) ) , user : Some ( opts . username ) . filter ( | a | ! a . is_empty ( ) ) , pass : Some ( opts . password ) . filter ( | a | ! a . is_empty ( ) ) , name : Some ( opts . name ) . filter ( | a | ! a . is_empty ( ) ) , lang : " " . to_string ( ) , version : " " . to_string ( ) , protocol : 1 , echo : false , sig : None , jwt : None , nkey : None , } ) ; self_arc . send_command ( connect ) . await ? ; let mut state_guard = self_arc . state . write ( ) . await ; * state_guard = NatsClientState :: Connected ; Ok ( ( ) ) }
Issue a connect command to NATS
Issue a connect command to NATS
self_arc
Arc<Self>
version
u128
stream
SplitStream<NatsTcpStream>
self_arc
Arc<Self>
version
u128
stream
SplitStream<NatsTcpStream>