flowr 1.0.0

Runners for compiled 'flow' programs
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

use std::fmt::Display;

use flowcore::errors::{Result, ResultExt};
pub use flowrlib::discovery::discover_service;
/// This is the message-queue implementation of the Client<-->[Coordinator][flowrlib::coordinator::Coordinator]
/// communications
use log::{info, trace};
use zmq::Socket;

use crate::gui::coordinator_connection::WAIT;

/// `ClientConnection` stores information related to the connection from a client
/// to the [Coordinator][flowrlib::coordinator::Coordinator] and is used each time a message is to
/// be sent or received.
pub struct ClientConnection {
    requester: Socket,
}

impl ClientConnection {
    /// Create a new connection between client and [Coordinator][flowrlib::coordinator::Coordinator]
    pub fn new(coordinator_address: &str) -> Result<Self> {
        info!("Client will attempt to connect to coordinator at: '{coordinator_address}'");

        let context = zmq::Context::new();

        let requester = context
            .socket(zmq::REQ)
            .chain_err(|| "Client could not connect to coordinator service")?;

        requester
            .connect(&format!("tcp://{coordinator_address}"))
            .chain_err(|| {
                format!("Client Connection - Could not connect to socket at: {coordinator_address}")
            })?;

        info!("Client connected to coordinator at '{coordinator_address}'");

        Ok(ClientConnection { requester })
    }

    /// Receive a [`CoordinatorMessage`][crate::gui::coordinator_message::CoordinatorMessage] from the
    /// [Coordinator][flowrlib::coordinator::Coordinator]
    pub fn receive<CM>(&self) -> Result<CM>
    where
        CM: From<String> + Display,
    {
        trace!("Client waiting for message from coordinator");

        let msg = self
            .requester
            .recv_msg(WAIT)
            .map_err(|e| format!("Error receiving from coordinator: {e}"))?;

        let message_string = msg
            .as_str()
            .ok_or("Could not get Message as String")?
            .to_owned();
        trace!("Client Received <--- {message_string}");
        Ok(message_string.into())
    }

    /// Send a [`ClientMessage`][crate::gui::client_message::ClientMessage] to the
    /// [Coordinator][flowrlib::coordinator::Coordinator]
    pub fn send<CM>(&self, message: CM) -> Result<()>
    where
        CM: Into<String> + Display,
    {
        trace!("Client Sending     ---> {message}");
        self.requester
            .send(&message.into(), WAIT)
            .chain_err(|| "Error sending to coordinator")
    }
}