Method execute_deployment

app/src/deployment_manager/mod.rs:273–361

We have a new deployment to execute, but no task yet. Contact the gateway to claim the deployment and get a task to execute

(
        &mut self,
        mut deployment: DeploymentContext,
    )

    /// We have a new deployment to execute, but no task yet. Contact the gateway to claim the deployment and get a task to execute
    async fn execute_deployment(
        &mut self,
        mut deployment: DeploymentContext,
    ) -> (DeploymentManagerState, Option<Duration>) {
        info!("Starting to execute deployment");
        let (mut msg_stream, close_upstream_tx, logger, metrics_registry, log_file_writer) =
            match deployment.execute_deployment(&mut self.engine_client).await {
                Ok(upstream_msg) => upstream_msg,
                Err(err) => {
                    return match err.code() {
                        Code::NotFound | Code::DeadlineExceeded => {
                            error!(
                                "Deployment cannot be executed due to {:?} for {:?}",
                                err, &deployment.deployment_info
                            );
                            (DeploymentManagerState::SeekingNewDeployment {}, None)
                        }
                        _ => {
                            error!("Error while getting new deployment: {}", err);
                            let next_step = DeploymentManagerState::ExecutingDeployment { deployment };
                            (next_step, Some(self.default_wait_time))
                        }
                    };
                }
            };
        info!(
            "Connected to gateway, executing deployment task for: {:?}",
            deployment.deployment_info
        );

        tokio::select! {
            biased;

            // We lost the connection with gateway to forward engine message, trying to reconnect
            _ = close_upstream_tx.closed() => {
                info!("EngineEvent forwarder to gateway has been close, trying to resume connection");
                let next_step = DeploymentManagerState::ExecutingDeployment { deployment };
                (next_step, None)
            }
            msg = msg_stream.next() => match msg {
                Some(Ok(msg)) => {
                    // We record the last message we received, so in case of cnx loss
                    // we can resume the deployment and restart from the last message
                    deployment.set_last_message_id(msg.message_id.clone());

                    match msg.request {
                        Some(engine_message_rx::Request::DeploymentRequest(payload)) => {
                            info!("Received new deployment task: {}", payload);
                            let task = (self.mk_engine_task)(payload, &deployment.deployment_info, &self.engine_client, logger.clone(), metrics_registry.clone(), log_file_writer.clone());
                            let upstream = UpstreamGatewayContext::new(msg_stream, close_upstream_tx, logger, metrics_registry, log_file_writer);
                            match task {
                                Ok(task) => {
                                    let next_step = DeploymentManagerState::ExecutingDeploymentTask {
                                        deployment,
                                        task: TaskContext::spawn_new_task(task),
                                        upstream_gtw: Box::new(upstream),
                                    };
                                    (next_step, None)