async fn handle_incoming(
    mut socket: TcpStream,
    sender: Sender<Message>,
    state: SharedState,
    settings: Settings,
) -> Result<()> {
    loop {
        // Receive the secret and check, whether the client is allowed to connect
        let payload_bytes = receive_bytes(&mut socket).await?;
        let secret = String::from_utf8(payload_bytes)?;
        if secret != settings.daemon.secret {
            warn!("Received invalid secret: {}", secret);
            return Ok(());
        }

        // Receive the actual instruction from the client
        let message = receive_message(&mut socket).await?;
        info!("Received instruction: {:?}", message);

        let response = if let Message::StreamRequest(message) = message {
            // The client requested the output of a task
            // Since we allow streaming, this needs to be handled seperately
            handle_show(&settings, &mut socket, message).await?
        } else if let Message::DaemonShutdown = message {
            // Simply shut down the daemon right after sending a success response
            let response = create_success_message("Daemon is shutting down");
            send_message(&response, &mut socket).await?;
            std::process::exit(0);
        } else {
            // Process a normal message
            handle_message(message, &sender, &state)
        };

        // Respond to the client
        send_message(&response, &mut socket).await?;
    }
}
/// Continuously poll the existing incoming futures.
/// In case we received an instruction, handle it and create a response future.
/// The response future is added to unix_responses and handled in a separate function.
Continuously poll the existing incoming futures.
In case we received an instruction, handle it and create a response future.
The response future is added to unix_responses and handled in a separate function.
Continuously poll the existing incoming futures . In case we received an instruction handle it and create a response future . The response future is added to unix_responses and handled in a separate function .
async fn handle_incoming(
    mut socket: TcpStream,
    sender: Sender<Message>,
    state: SharedState,
    settings: Settings,
) -> Result<()> {
    loop {
        // Receive the secret and check, whether the client is allowed to connect
        let payload_bytes = receive_bytes(&mut socket).await?;
        let secret = String::from_utf8(payload_bytes)?;
        if secret != settings.daemon.secret {
            warn!("Received invalid secret: {}", secret);
            return Ok(());
        }
        // Receive the actual instruction from the client
        let message = receive_message(&mut socket).await?;
        info!("Received instruction: {:?}", message);
        let response = if let Message::StreamRequest(message) = message {
            // The client requested the output of a task
            // Since we allow streaming, this needs to be handled seperately
            handle_show(&settings, &mut socket, message).await?
        } else if let Message::DaemonShutdown = message {
            // Simply shut down the daemon right after sending a success response
            let response = create_success_message("Daemon is shutting down");
            send_message(&response, &mut socket).await?;
            std::process::exit(0);
        } else {
            // Process a normal message
            handle_message(message, &sender, &state)
        };
        // Respond to the client
        send_message(&response, &mut socket).await?;
    }
}
async fn handle_incoming ( mut socket : TcpStream , sender : Sender < Message > , state : SharedState , settings : Settings , ) -> Result < ( ) > { loop { // Receive the secret and check, whether the client is allowed to connect let payload_bytes = receive_bytes ( & mut socket ) . await ? ; let secret = String :: from_utf8 ( payload_bytes ) ? ; if secret != settings . daemon . secret { warn ! ( " " secret ) ; return Ok ( ( ) ) ; } // Receive the actual instruction from the client let message = receive_message ( & mut socket ) . await ? ; info ! ( " " message ) ; let response = if let Message :: StreamRequest ( message ) = message { // The client requested the output of a task // Since we allow streaming, this needs to be handled seperately handle_show ( & settings , & mut socket , message ) . await ? } else if let Message :: DaemonShutdown = message { // Simply shut down the daemon right after sending a success response let response = create_success_message ( " " ) ; send_message ( & response , & mut socket ) . await ? ; std :: process :: exit ( 0 ) ; } else { // Process a normal message handle_message ( message , & sender , & state ) } ; // Respond to the client send_message ( & response , & mut socket ) . await ? ; } }
Continuously poll the existing incoming futures.
Continuously poll the existing incoming futures .
socket
TcpStream
sender
Sender<Message>
state
SharedState
settings
Settings
socket
TcpStream
sender
Sender<Message>
state
SharedState
settings
Settings