How to use TCP-Inlet with user-defined Message type in Rust

[epi052]

Good evening!

I've been trying to use a TCP-Inlet to send commands from a user-interface through a worker Node on to another worker Node.

My primary components are:

• User Interface (UI)
• Server
• Agent

I'm open to suggestions on how to better facilitate these communications using Ockam, but here's what I've planned/done so far.

• Agent connects in to Server's TCPTransport listener on port 4001
• UI connects in to Server's TCP-Inlet listener on port 4000
• UI sends JSON-encoded command over TCP from a python program (no ockam) to the Server's TCP-Inlet
• Server receives command on TCP-inlet and routes to the Server's Agent connection via its worker. The Message type that the Server worker expects is an Event
• Return trips flow back from the Agent, through the Server, to the UI

Currently, when I connect the UI to the TCP-Inlet, I get an empty message. I've tried inserting an additional Worker as a translation layer between the incoming String (? assumed) and the Event, so I've seen .body() is empty, even though the encoded json traverses the wire. Even without the additional translator, routing directly to the Server<->Agent Worker, the initial connection triggers an unexpected enum variant from Event.

All that said, I'm not entirely sure what my issue is. I suspect it's due to the String vs Event Message types? What's the best way for me to approach this?

Thank you!

[epi052]

Some code to reproduce / clarify

Cargo.toml

[package]
name = "ockam-error-repro"
version = "0.1.0"
edition = "2021"

[dependencies]
ockam = { version = "0.83.0" }
serde = { version = "1.0.130", features = ["derive"] }
serde_json = { version = "1.0.70" }

lib.rs

use ockam::Message;
use serde::{Deserialize, Serialize};

#[derive(Message, Debug, Clone, Serialize, Deserialize)]
pub enum Command {
    RegisterUI,
    RegisterClient,
}

#[derive(Message, Debug, Clone, Serialize, Deserialize)]
pub enum Event {
    Inbound(Command),
    Outbound(Command),
}

bin/server.rs

use ockam::access_control::AllowAll;
use ockam::{
    Context, Result, Routed, TcpInletTrustOptions, TcpListenerTrustOptions, TcpTransport, Worker,
};

use ockam_error_repro::{Command, Event};

#[derive(Default, Debug)]
pub struct ServerWorker;

#[ockam::worker]
impl Worker for ServerWorker {
    type Message = Event;
    type Context = Context;

    async fn handle_message(
        &mut self,
        context: &mut Self::Context,
        msg: Routed<Self::Message>,
    ) -> Result<()> {
        let route = msg.return_route();

        println!(
            "Address: {}, Received: {:?} from {}",
            context.address(),
            msg,
            route
        );

        let body = msg.body();

        match body {
            Event::Outbound(Command::RegisterClient) | Event::Inbound(Command::RegisterClient) => {
                println!("Received client registration: {:?}", body);
            }
            Event::Outbound(Command::RegisterUI) | Event::Inbound(Command::RegisterUI) => {
                println!("Received UI registration: {:?}", body);
            }
        }

        Ok(())
    }
}

#[derive(Debug, Default)]
pub struct Translator;

#[ockam::worker]
impl Worker for Translator {
    type Message = String;
    type Context = Context;

    async fn handle_message(
        &mut self,
        context: &mut Self::Context,
        msg: Routed<Self::Message>,
    ) -> Result<()> {
        println!("Received message for translation: {:?}", msg);

        let body = msg.body();

        let Ok(command) = serde_json::from_str::<Event>(&body) else {
            println!("Failed to parse message: {:?}", body);
            return Ok(());
        };

        context.send("server", command).await?;

        Ok(())
    }
}

#[ockam::node]
fn main(context: &Context) -> Result<()> {
    let tcp = TcpTransport::create(&context).await?;

    tcp.listen("127.0.0.1:4000", TcpListenerTrustOptions::new())
        .await?;

    context
        .start_worker("server", ServerWorker, AllowAll, AllowAll)
        .await?;

    tcp.create_inlet("127.0.0.1:4001", "inlet", TcpInletTrustOptions::new())
        .await?;

    context
        .start_worker("inlet", Translator {}, AllowAll, AllowAll)
        .await?;

    Ok(())
}

bin/client.rs

use ockam::access_control::AllowAll;
use ockam::{route, Context, Result, Routed, TcpConnectionTrustOptions, TcpTransport, Worker};

use ockam_error_repro::{Command, Event};

#[derive(Debug)]
pub struct ClientWorker {
    route: ockam::Route,
}

#[ockam::worker]
impl Worker for ClientWorker {
    type Message = Event;
    type Context = Context;

    async fn handle_message(
        &mut self,
        context: &mut Self::Context,
        msg: Routed<Self::Message>,
    ) -> Result<()> {
        let route = msg.return_route();

        println!(
            "Address: {}, Received: {:?} from {}",
            context.address(),
            msg,
            route
        );

        Ok(())
    }

    async fn initialize(&mut self, context: &mut Self::Context) -> Result<()> {
        context
            .send(self.route.clone(), Event::Outbound(Command::RegisterClient))
            .await?;

        Ok(())
    }
}

#[ockam::node]
fn main(context: &Context) -> Result<()> {
    let tcp = TcpTransport::create(&context).await?;

    let server_addr = tcp
        .connect("localhost:4000", TcpConnectionTrustOptions::new())
        .await?;
    let route = route![server_addr, "server"];

    context
        .start_worker("client", ClientWorker { route }, AllowAll, AllowAll)
        .await?;

    Ok(())
}

bin/ui.py

import asyncio
import json

async def main():
    """connect to port 4001 and send a message"""
    reader, writer = await asyncio.open_connection("localhost", 4001)
    # writer.write(json.dumps({"command": "RegisterUI", "event": "Inbound"}).encode())
    # writer.write(json.dumps({"command": "RegisterClient", "event": "Inbound"}).encode())
    writer.write(b"hello")
    await writer.drain()
    data = await reader.read(100)

if __name__ == '__main__':
    loop = asyncio.new_event_loop()
    loop.run_until_complete(main())

[epi052]

I've experimented with the python message, with/without the translator, with a tcp-outlet pointed at the server Worker, and other configurations. It doesn't seem to matter, the message received by the server Worker is always empty when coming from the UI (client works fine)

[etorreborre]

Hi @epi052 you are not seeing your expected message because a TcpInlet is in general supposed to work with a corresponding outlet. So, there is a small protocol in place to make sure that both the inlet and the outlet are up using the PortalMessage struct.
This means that the messages you are expecting will eventually come from PortalMessage::Payload.

I took your code and created a small project here that you can run. You will see that the registration command eventually ends-up on the Server. Note that I had to also modify the ui.py code in order to send the exact JSON representation for an Event. I hope that helps!

[epi052]

woooo! awesome, tyvm 🙏

I played around with an explicit outlet as well, but definitely didn't do anything with a PortalMessage. Appreciate the help!

[epi052]

Just reporting back. I was able to successfully work this into the real project. Thank you again for your help!