Function client_rpc_loop

crates/hyperqueue/src/server/client/mod.rs:192–368

(
    mut tx: Tx,
    mut rx: Rx,
    server_dir: ServerDir,
    state_ref: StateRef,
    senders: &Senders,
    end_flag: Arc<Notify>,
)

}

pub async fn client_rpc_loop<
    Tx: Sink<ToClientMessage, Error = tako::Error> + Unpin + 'static,
    Rx: Stream<Item = tako::Result<FromClientMessage>> + Unpin,
>(
    mut tx: Tx,
    mut rx: Rx,
    server_dir: ServerDir,
    state_ref: StateRef,
    senders: &Senders,
    end_flag: Arc<Notify>,
) where
    Tx::Error: Debug,
{
    while let Some(message_result) = rx.next().await {
        match message_result {
            Ok(message) => {
                let response = match message {
                    FromClientMessage::Submit(msg, stream_opts) => {
                        let response = submit::handle_submit(&state_ref, senders, msg);
                        if !response.is_error() {
                            senders.events.flush_journal().await;
                        };
                        if let Some(mut stream_opts) = stream_opts
                            && let ToClientMessage::SubmitResponse(SubmitResponse::Ok {
                                job, ..
                            }) = &response
                        {
                            if !stream_opts.filter.is_filtering_jobs() {
                                let mut s = Set::new();
                                s.insert(job.info.id);
                                stream_opts.filter.set_jobs(s);
                            }
                            start_streaming(
                                tx,
                                rx,
                                state_ref,
                                senders,
                                stream_opts,
                                Some(response),
                            )
                            .await;
                            break;
                        }
                        response
                    }
                    FromClientMessage::JobInfo(msg, stream_opts) => {
                        let response =
                            compute_job_info(&state_ref, &msg.selector, msg.include_running_tasks);
                        if let Some(mut stream_opts) = stream_opts
                            && let ToClientMessage::JobInfoResponse(JobInfoResponse { jobs }) =
                                &response
                        {
                            if !stream_opts.filter.is_filtering_jobs() {
                                stream_opts
                                    .filter
                                    .set_jobs(jobs.iter().map(|j| j.id).collect());
                            }
                            start_streaming(