Skip to main content

proxy_steerable

Function proxy_steerable 

Source
pub async fn proxy_steerable<F, B, C, Ctrl>(
    frontend: &mut F,
    backend: &mut B,
    capture: Option<&mut C>,
    control: &mut Ctrl,
) -> Result<(), Error>
Expand description

Run a steerable bidirectional message proxy with control socket.

Like proxy() but can be controlled via a control socket that receives commands:

  • PAUSE - Stop forwarding messages (buffering continues)
  • RESUME - Resume forwarding messages
  • TERMINATE - Stop the proxy and return
  • STATISTICS - Future: report proxy statistics

§Parameters

  • frontend: Socket facing clients/publishers
  • backend: Socket facing workers/subscribers
  • capture: Optional socket to receive message copies
  • control: Socket that receives control commands

§Control Socket Protocol

Send single-frame messages with command text:

PAUSE       - Pause forwarding
RESUME      - Resume forwarding
TERMINATE   - Stop proxy
STATISTICS  - Get stats (future)

§Example

use monocoque_zmtp::proxy::{proxy_steerable, ProxySocket, ProxyCommand};
use monocoque_zmtp::router::RouterSocket;
use monocoque_zmtp::dealer::DealerSocket;
use monocoque_zmtp::pair::PairSocket;

#[compio::main]
async fn main() -> std::io::Result<()> {
    // Broker sockets
    let (_, mut frontend) = RouterSocket::bind("127.0.0.1:5555").await?;
    let (_, mut backend) = DealerSocket::bind("127.0.0.1:5556").await?;

    // Control socket
    let (_, mut control) = PairSocket::bind("127.0.0.1:5557").await?;

    // Run steerable proxy
    proxy_steerable(&mut frontend, &mut backend, None, &mut control).await?;
    Ok(())
}

Send control commands from another socket:

use monocoque_zmtp::pair::PairSocket;
use bytes::Bytes;

let mut control_client = PairSocket::connect("127.0.0.1:5557").await?;

// Pause proxy
control_client.send(vec![Bytes::from("PAUSE")]).await?;

// Resume proxy
control_client.send(vec![Bytes::from("RESUME")]).await?;

// Terminate proxy
control_client.send(vec![Bytes::from("TERMINATE")]).await?;