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
89
90

pub(crate) mod errors;
mod http_gateway;
pub(crate) mod web_handling;

pub use http_gateway::HttpGateway;
use locutus_core::{
    locutus_runtime::ContractKey, ClientError, ClientId, ClientRequest, HostResponse, HostResult,
};

type DynError = Box<dyn std::error::Error + Send + Sync + 'static>;

#[derive(Debug)]
enum ClientConnection {
    NewConnection(tokio::sync::mpsc::UnboundedSender<HostCallbackResult>),
    Request {
        client_id: ClientId,
        req: ClientRequest,
    },
}

#[derive(Debug)]
enum HostCallbackResult {
    NewId(ClientId),
    Result {
        id: ClientId,
        result: Result<HostResponse, ClientError>,
    },
    SubscriptionChannel {
        key: ContractKey,
        id: ClientId,
        callback: tokio::sync::mpsc::UnboundedReceiver<HostResult>,
    },
}

#[cfg(feature = "local")]
pub mod local_node {
    use std::net::{Ipv4Addr, SocketAddr};

    use locutus_core::{
        either, ClientError, ClientEventsProxy, ContractExecutor, ErrorKind, OpenRequest,
        RequestError, WebSocketProxy,
    };

    use crate::{DynError, HttpGateway};

    pub async fn run_local_node(mut executor: ContractExecutor) -> Result<(), DynError> {
        let (mut http_handle, filter) = HttpGateway::as_filter();
        let socket: SocketAddr = (Ipv4Addr::LOCALHOST, 50509).into();
        let _ws_handle = WebSocketProxy::as_upgrade(socket, filter).await?;
        // FIXME: use combinator
        // let mut all_clients =
        //    ClientEventsCombinator::new([Box::new(ws_handle), Box::new(http_handle)]);
        loop {
            let OpenRequest {
                id,
                request,
                notification_channel,
                ..
            } = http_handle.recv().await?;
            tracing::debug!("client {id}, req -> {request}");
            match executor
                .handle_request(id, request, notification_channel)
                .await
            {
                Ok(res) => {
                    http_handle.send(id, Ok(res)).await?;
                }
                Err(either::Left(RequestError::Disconnect)) => {}
                Err(either::Left(err)) => {
                    log::error!("{err}");
                    http_handle
                        .send(id, Err(ClientError::from(ErrorKind::from(err))))
                        .await?;
                }
                Err(either::Right(err)) => {
                    log::error!("{err}");
                    http_handle
                        .send(
                            id,
                            Err(ErrorKind::Unhandled {
                                cause: format!("{err}"),
                            }
                            .into()),
                        )
                        .await?;
                }
            }
        }
    }
}