oreganoli/zhur - Change AGX35J5GL7C33UWYITQZEP6QRYBUTFJV3ANW6YMEMZ6WVWG7ZLXAC

work on communicating with core

Created by oreganoli on January 11, 2021

AGX35J5GL7C33UWYITQZEP6QRYBUTFJV3ANW6YMEMZ6WVWG7ZLXAC

Dependencies

In channels

main

Change contents

Replacement in zhur_invk/src/lib.rs at line 9 [2.29]
B:BD[2.209] → [2.209:250]
```
#[derive(Deserialize, Serialize, Debug)]
```
[2.209]
[2.250]
```
#[derive(Clone, Deserialize, Serialize, Debug)]
```
Replacement in zhur_invk/src/http.rs at line 21 [2.485]
B:BD[2.1212] → [2.1212:1253]
```
#[derive(Debug, Deserialize, Serialize)]
```
[2.1212]
[2.1253]
```
#[derive(Clone, Debug, Deserialize, Serialize)]
```
Replacement in zhur_invk/src/err.rs at line 5 [2.1938]
B:BD[2.2075] → [2.2075:2116]
```
#[derive(Deserialize, Serialize, Debug)]
```
[2.2075]
[2.2116]
```
#[derive(Clone, Deserialize, Serialize, Debug)]
```

Insertion in zhur_invk/src/err.rs at line 20 [2.1938]

[2.2614]

    /// The invocation could not be serialized for transport.
    SerializeErr,

Replacement in zhur_invk/src/err.rs at line 37 [2.1938]

B:BD[2.3377] → [2.3377:3518]

            Self::MalformedReply => "The reply from the Zhur core server could not be properly deserialized as an app's output.".to_owned(),

[2.3377]

[2.3518]

            Self::MalformedReply => "The reply from the Zhur core server could not be properly deserialized as an app's HTTP output.".to_owned(),

Insertion in zhur_invk/src/err.rs at line 39 [2.1938]

[2.3627]

            Self::SerializeErr => "The invocation could not be serialized for transport.".to_owned(),

Replacement in zhur_gate/src/main.rs at line 1 [4.30]

B:BD[4.30] → [5.0:30]

use zhur_common::init_logger;

[4.30]

[5.30]

use zhur_common::{init_logger, zmq,
    msg::chan::{
        client_server,
    }
};
use zhur_gate::comms::Gate2CoreServer;

Insertion in zhur_gate/src/main.rs at line 10 [4.30]

[4.435]

[5.60]

    let zmq_ctx = zmq::Context::new();
    let gate_core_server = Gate2CoreServer::new(&zmq_ctx);
    let (client, server) = client_server(gate_core_server);
    std::thread::spawn(move || {
        let mut server = server;
        loop {
            server.handle()
        }
    });

Replacement in zhur_gate/src/main.rs at line 20 [4.30]
B:BD[5.79] → [5.79:105]
```
    start_server().await;
```
[5.79]
[4.1170]
```
    start_server(client).await;
```

Insertion in zhur_gate/src/lib.rs at line 13 [5.115]

[6.150]

[5.455]

/// Communication with the core module.
pub mod comms;
use zhur_common::msg::chan::*;
use zhur_invk::{HttpRes, Invocation, InvocationError};

Replacement in zhur_gate/src/lib.rs at line 19 [5.115]

B:BD[5.486] → [5.486:516]

pub async fn start_server() {

[5.486]

[5.516]

pub async fn start_server(client: ChannelClient<Invocation, Result<HttpRes, InvocationError>>) {

Insertion in zhur_gate/src/lib.rs at line 36 [5.115]
[6.266]
[6.266]
```
            let client = client.clone();
```

Replacement in zhur_gate/src/lib.rs at line 38 [5.115]

B:BD[6.291] → [6.291:398]

                Ok::<_, Infallible>(service_fn(move |req| handle_req(FullRequest{ req, ip: ip.clone() })))

[6.291]

[6.398]

                Ok::<_, Infallible>(service_fn(move |req| handle_req(
                    FullRequest{ req, ip: ip.clone() },
                    client.clone() // TODO: remove this horrific hack
                )))

Replacement in zhur_gate/src/handle.rs at line 8 [6.442]
B:BD[2.4409] → [2.4409:4434]
```
use zhur_common::log::*;
```
[2.4409]
[2.4434]
```
use zhur_common::{log::*, msg::chan::ChannelClient};
```

Replacement in zhur_gate/src/handle.rs at line 19 [6.442]

B:BD[6.835] → [6.835:917]

B:BD[6.917] → [2.4526:4759]

pub async fn handle_req(req: FullRequest) -> Result<Response<Body>, Infallible> {
    let invocation = req.into_invoc().await;
    // TODO: actually send the invocation to the core and try to get something back.
    // Also TODO: extract all this stuff into its own function so we can ? it.
    match &invocation {

[6.835]

[2.4759]

pub async fn handle_req(req: FullRequest, mut client: ChannelClient<Invocation, Result<HttpRes, InvocationError>>) -> Result<Response<Body>, Infallible> {
    let invocation = match req.into_invoc().await {

Replacement in zhur_gate/src/handle.rs at line 24 [6.442]
B:BD[2.4930] → [2.4930:4973]
```
            Ok(Response::new(text.into()))
```
[2.4930]
[2.4973]
```
            i
```

Replacement in zhur_gate/src/handle.rs at line 29 [6.442]

B:BD[2.5102] → [2.5102:5145]

            Ok(Response::new(text.into()))

[2.5102]

[2.5145]

            return Ok(Response::new(text.into())) // TODO: Error pages
        }
    };
    // TODO: actually send the invocation to the core and try to get something back.
    let reply = client.request(invocation);
    match reply {
        Ok(res) => {
            info!("Got a well-formed HttpRes as an invocation result!");
            return Ok(Response::new(
                format!("Response with body length {}", res.body.len())
                .into()
            )) // TODO actually handle the damn HttpRes
        },
        Err(e) => {
            let text = format!("Got an invocation error: {}", e);
            warn!("{}", &text);
            return Ok(Response::new(text.into())) // TODO: Error pages
            // TODO: Not brazen copypaste

File addition: comms.rs (-rw-r--r--)

[4.19]

use bincode::deserialize;
use zhur_common::{bincode, log::warn, msg::chan::*, zmq};
use zhur_invk::{HttpRes, Invocation, InvocationError};
use zmq::{Context, Socket, SocketType};
/// Struct responsible for relaying requests from the gateway to the core.
pub struct Gate2CoreServer {
    /// ZMQ REQ socket.
    req_socket: Socket,
}
impl Gate2CoreServer {
    pub fn new(zmq_ctx: &Context) -> Self {
        Self {
            req_socket: {
                let sck = zmq_ctx.socket(SocketType::REQ)
                .expect("Expected to be able to construct a socket.");
                let endpoint = match std::env::var("ZHUR_CORE_REP_URI") {
                    Ok(s) => s,
                    Err(_) => {
                        warn!("ZHUR_CORE_REP_URI not set - assuming default value of tcp://127.0.0.1:8081!");
                        "tcp://127.0.0.1:8081".to_owned()
                    }
                };
                sck.connect(&endpoint)
                .expect("Expected to be able to connect a REQ socket from the gateway to the core.");
                sck
            }
        }
    }
}
impl HandleRequest<Invocation, Result<HttpRes, InvocationError>> for Gate2CoreServer {
    fn handle(&mut self, msg: Invocation) -> Result<HttpRes, InvocationError> {
        let invoc_bytes = match bincode::serialize(&msg) {
            Ok(b) => b,
            Err(_) => return Err(InvocationError::SerializeErr)
        };
        match self.req_socket.send(invoc_bytes, 0) {
            Ok(_) => (),
            Err(_) => return Err(InvocationError::NoCore)
        };
        let response_bytes = match self.req_socket.recv_bytes(0) {
            Ok(b) => b,
            Err(_) => return Err(InvocationError::MalformedReply)
        };
        match deserialize(&response_bytes) {
            Ok(r) => r,
            Err(_) => Err(InvocationError::MalformedReply)
        }
    }
}

Insertion in zhur_common/src/msg/chan.rs at line 10 [3.84]

[3.371]

impl<T: Send, U: Send, V: HandleRequest<T, U>> ChannelServer<T, U, V> {
    pub fn handle(&mut self) {
        // TODO: handle timeouts
        let (req, reply_tx) = match self.request_rx.recv() {
            Ok(t) => {
                trace!("A ChannelServer got a request!");
                t
            },
            Err(_) => {
                panic!("A ChannelServer could not receive a request!");
            }
        };
        let reply = self.handler.handle(req);
        match reply_tx.send(reply) {
            Ok(_) => {
                trace!("A ChannelServer replied to a request!");
            },
            Err(_) => {
                panic!("A ChannelServer could not respond to a request!");
            }
        }
    }
}

work on communicating with core

Dependencies

In channels

Change contents

Replacement in zhur_invk/src/lib.rs at line 9 [2.29]

Replacement in zhur_invk/src/http.rs at line 21 [2.485]

Replacement in zhur_invk/src/err.rs at line 5 [2.1938]