ora 0.12.7

Part of the Ora scheduler framework.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

use flume::{Receiver, Sender};
use futures::{Stream, StreamExt};

use crate::proto::executors::v1::{
    ExecutorConnectionRequest, ExecutorMessage, execution_service_client::ExecutionServiceClient,
    executor_message::ExecutorMessageKind, server_message::ServerMessageKind,
};

#[tracing::instrument(skip_all, fields(executor_id))]
pub(super) async fn connection_loop<T>(
    client: &mut ExecutionServiceClient<T>,
    incoming: Sender<ServerMessageKind>,
    outgoing: Receiver<ExecutorMessageKind>,
) where
    T: tonic::client::GrpcService<tonic::body::Body>,
    T::Error: Into<tonic::codegen::StdError>,
    T::ResponseBody:
        tonic::codegen::Body<Data = tonic::codegen::Bytes> + std::marker::Send + 'static,
    <T::ResponseBody as tonic::codegen::Body>::Error:
        Into<tonic::codegen::StdError> + std::marker::Send,
{
    let stream = client
        .executor_connection(
            hack_to_stream(outgoing).map(|msg| ExecutorConnectionRequest {
                message: Some(ExecutorMessage {
                    executor_message_kind: Some(msg),
                }),
            }),
        )
        .await;

    let mut stream = match stream {
        Ok(s) => s.into_inner(),
        Err(error) => {
            tracing::error!(%error, "failed to establish executor connection");
            return;
        }
    };

    tracing::info!("executor connected");

    loop {
        let res = match stream.message().await {
            Ok(Some(res)) => res,
            Ok(None) => {
                break;
            }
            Err(error) => {
                tracing::error!(%error, "failed to receive server message");
                break;
            }
        };

        let Some(msg) = res.message else {
            tracing::error!("received server response with no message");
            break;
        };

        let Some(msg) = msg.server_message_kind else {
            tracing::error!("received server message with no message kind");
            break;
        };

        if let ServerMessageKind::Properties(props) = &msg {
            tracing::Span::current().record("executor_id", &props.executor_id);
        }

        if incoming.send_async(msg).await.is_err() {
            break;
        }
    }

    tracing::info!("executor disconnected");
}

// Needed due to the lifetime of `RecvStream<'_, T>` not being generic enough,
// when the compiler tries to infer lifetimes.
// With this function we assert that
// it is infact a stream implementation with a `'static` lifetime.
//
// It's either a compiler bug or a limitation.
#[inline]
fn hack_to_stream<T>(recv: Receiver<T>) -> impl Stream<Item = T> + Unpin + 'static
where
    T: 'static,
{
    recv.into_stream()
}