async fn run_wormhole(
    config: Config,
    introspect_web_addr: SocketAddr,
    mut restart_tx: UnboundedSender<Option<Error>>,
) -> Result<(), Error> {
    let interface = CliInterface::start(config.clone(), introspect_web_addr);
    tokio::time::sleep(std::time::Duration::from_millis(500)).await;
    let Wormhole {
        websocket,
        sub_domain,
        hostname,
    } = connect_to_wormhole(&config).await?;

    interface.did_connect(&sub_domain, &hostname);

    // split reading and writing
    let (mut ws_sink, mut ws_stream) = websocket.split();

    // tunnel channel
    let (tunnel_tx, mut tunnel_rx) = unbounded::<ControlPacket>();

    // continuously write to websocket tunnel
    let mut restart = restart_tx.clone();
    tokio::spawn(async move {
        loop {
            let packet = match tunnel_rx.next().await {
                Some(data) => data,
                None => {
                    warn!("control flow didn't send anything!");
                    let _ = restart.send(Some(Error::Timeout)).await;
                    return;
                }
            };

            if let Err(e) = ws_sink.send(Message::binary(packet.serialize())).await {
                warn!("failed to write message to tunnel websocket: {:?}", e);
                let _ = restart.send(Some(Error::WebSocketError(e))).await;
                return;
            }
        }
    });

    // continuously read from websocket tunnel

    loop {
        match ws_stream.next().await {
            Some(Ok(message)) if message.is_close() => {
                debug!("got close message");
                let _ = restart_tx.send(None).await;
                return Ok(());
            }
            Some(Ok(message)) => {
                let packet = process_control_flow_message(
                    config.clone(),
                    tunnel_tx.clone(),
                    message.into_data(),
                )
                .await
                .map_err(|e| {
                    error!("Malformed protocol control packet: {:?}", e);
                    Error::MalformedMessageFromServer
                })?;
                debug!("Processed packet: {:?}", packet.packet_type());
            }
            Some(Err(e)) => {
                warn!("websocket read error: {:?}", e);
                return Err(Error::Timeout);
            }
            None => {
                warn!("websocket sent none");
                return Err(Error::Timeout);
            }
        }
    }
}
/// Setup the tunnel to our control server
Setup the tunnel to our control server
Setup the tunnel to our control server
async fn run_wormhole(
    config: Config,
    introspect_web_addr: SocketAddr,
    mut restart_tx: UnboundedSender<Option<Error>>,
) -> Result<(), Error> {
    let interface = CliInterface::start(config.clone(), introspect_web_addr);
    tokio::time::sleep(std::time::Duration::from_millis(500)).await;
    let Wormhole {
        websocket,
        sub_domain,
        hostname,
    } = connect_to_wormhole(&config).await?;
    interface.did_connect(&sub_domain, &hostname);
    // split reading and writing
    let (mut ws_sink, mut ws_stream) = websocket.split();
    // tunnel channel
    let (tunnel_tx, mut tunnel_rx) = unbounded::<ControlPacket>();
    // continuously write to websocket tunnel
    let mut restart = restart_tx.clone();
    tokio::spawn(async move {
        loop {
            let packet = match tunnel_rx.next().await {
                Some(data) => data,
                None => {
                    warn!("control flow didn't send anything!");
                    let _ = restart.send(Some(Error::Timeout)).await;
                    return;
                }
            };
            if let Err(e) = ws_sink.send(Message::binary(packet.serialize())).await {
                warn!("failed to write message to tunnel websocket: {:?}", e);
                let _ = restart.send(Some(Error::WebSocketError(e))).await;
                return;
            }
        }
    });
    // continuously read from websocket tunnel
    loop {
        match ws_stream.next().await {
            Some(Ok(message)) if message.is_close() => {
                debug!("got close message");
                let _ = restart_tx.send(None).await;
                return Ok(());
            }
            Some(Ok(message)) => {
                let packet = process_control_flow_message(
                    config.clone(),
                    tunnel_tx.clone(),
                    message.into_data(),
                )
                .await
                .map_err(|e| {
                    error!("Malformed protocol control packet: {:?}", e);
                    Error::MalformedMessageFromServer
                })?;
                debug!("Processed packet: {:?}", packet.packet_type());
            }
            Some(Err(e)) => {
                warn!("websocket read error: {:?}", e);
                return Err(Error::Timeout);
            }
            None => {
                warn!("websocket sent none");
                return Err(Error::Timeout);
            }
        }
    }
}
async fn run_wormhole ( config : Config , introspect_web_addr : SocketAddr , mut restart_tx : UnboundedSender < Option < Error > > , ) -> Result < ( ) , Error > { let interface = CliInterface :: start ( config . clone ( ) , introspect_web_addr ) ; tokio :: time :: sleep ( std :: time :: Duration :: from_millis ( 500 ) ) . await ; let Wormhole { websocket , sub_domain , hostname , } = connect_to_wormhole ( & config ) . await ? ; interface . did_connect ( & sub_domain , & hostname ) ; // split reading and writing let ( mut ws_sink , mut ws_stream ) = websocket . split ( ) ; // tunnel channel let ( tunnel_tx , mut tunnel_rx ) = unbounded :: < ControlPacket > ( ) ; // continuously write to websocket tunnel let mut restart = restart_tx . clone ( ) ; tokio :: spawn ( async move { loop { let packet = match tunnel_rx . next ( ) . await { Some ( data ) => data , None => { warn ! ( " " ) ; let _ = restart . send ( Some ( Error :: Timeout ) ) . await ; return ; } } ; if let Err ( e ) = ws_sink . send ( Message :: binary ( packet . serialize ( ) ) ) . await { warn ! ( " " e ) ; let _ = restart . send ( Some ( Error :: WebSocketError ( e ) ) ) . await ; return ; } } } ) ; // continuously read from websocket tunnel loop { match ws_stream . next ( ) . await { Some ( Ok ( message ) ) if message . is_close ( ) => { debug ! ( " " ) ; let _ = restart_tx . send ( None ) . await ; return Ok ( ( ) ) ; } Some ( Ok ( message ) ) => { let packet = process_control_flow_message ( config . clone ( ) , tunnel_tx . clone ( ) , message . into_data ( ) , ) . await . map_err ( | e | { error ! ( " " e ) ; Error :: MalformedMessageFromServer } ) ? ; debug ! ( " " packet packet_type ( ) ) ; } Some ( Err ( e ) ) => { warn ! ( " " e ) ; return Err ( Error :: Timeout ) ; } None => { warn ! ( " " ) ; return Err ( Error :: Timeout ) ; } } } }
Setup the tunnel to our control server
Setup the tunnel to our control server
config
Config
introspect_web_addr
SocketAddr
restart_tx
UnboundedSender<Option<Error>>