Struct SharedMemoryFrameTransport 

pub struct SharedMemoryFrameTransport { /* private fields */ }

Shared memory frame transport with ring buffer (Layer 1).

Implementations

impl SharedMemoryFrameTransport

pub fn create_server( name: impl Into<String>, config: SharedMemoryConfig, ) -> TransportResult<Self>

Create a server-side transport.

Examples found in repository

examples/rpc_client_server.rs (line 55)

async fn run_server() -> Result<(), Box<dyn std::error::Error>> {
    println!("[Server] Starting RPC server");

    let config = SharedMemoryConfig::default();
    let transport = SharedMemoryFrameTransport::create_server(SERVICE_NAME, config)?;
    let channel = Arc::new(MessageChannelAdapter::new(transport));

    let server = RpcServer::new();

    server.register_typed("add", |req: AddRequest| async move {
        println!("[Server] add({}, {})", req.a, req.b);
        Ok(AddResponse {
            result: req.a + req.b,
        })
    });

    server.register_typed("echo", |req: EchoRequest| async move {
        println!("[Server] echo(\"{}\")", req.message);
        let len = req.message.len();
        Ok(EchoResponse {
            message: req.message,
            length: len,
        })
    });

    println!("[Server] Registered {} handlers", server.handler_count());
    println!("[Server] Waiting for requests\n");

    server.serve(channel).await?;

    Ok(())
}

examples/message_transport_shm.rs (line 50)

async fn run_server() -> Result<(), Box<dyn std::error::Error>> {
    println!("[Server] Creating shared memory transport");

    let config = SharedMemoryConfig::default();
    let transport = SharedMemoryFrameTransport::create_server(SERVICE_NAME, config)?;
    let channel = MessageChannelAdapter::new(transport);

    println!("[Server] Waiting for messages");

    loop {
        let message = channel.recv().await?;

        match message.msg_type {
            MessageType::Call => {
                println!(
                    "[Server] Received Call: method={}, id={}",
                    message.method, message.id
                );

                match message.method.as_str() {
                    "add" => {
                        let req: AddRequest = message.deserialize_payload()?;
                        println!("[Server] AddRequest: {} + {}", req.a, req.b);

                        let resp = AddResponse {
                            result: req.a + req.b,
                        };
                        let reply = Message::reply(message.id, resp)?;
                        channel.send(&reply).await?;
                        println!("[Server] Sent reply");
                    }
                    "divide" => {
                        let req: AddRequest = message.deserialize_payload()?;
                        if req.b == 0 {
                            let error = Message::error(message.id, "Division by zero");
                            channel.send(&error).await?;
                            println!("[Server] Sent error: division by zero");
                        } else {
                            let resp = AddResponse {
                                result: req.a / req.b,
                            };
                            let reply = Message::reply(message.id, resp)?;
                            channel.send(&reply).await?;
                        }
                    }
                    "shutdown" => {
                        println!("[Server] Shutdown requested");
                        let reply = Message::reply(message.id, "ok")?;
                        channel.send(&reply).await?;
                        break;
                    }
                    _ => {
                        let error = Message::error(message.id, "Unknown method");
                        channel.send(&error).await?;
                    }
                }
            }
            MessageType::Notification => {
                println!("[Server] Received Notification: method={}", message.method);
                if message.method == "log" {
                    let event: LogEvent = message.deserialize_payload()?;
                    println!("[Server] Log [{}]: {}", event.level, event.message);
                }
            }
            _ => {
                println!("[Server] Unknown message type: {:?}", message.msg_type);
            }
        }
    }

    println!("[Server] Shutting down");
    Ok(())
}

pub fn connect_client_with_config( name: impl Into<String>, config: SharedMemoryConfig, ) -> TransportResult<Self>

Connect as a client with custom config.

pub fn connect_client(name: impl Into<String>) -> TransportResult<Self>

Connect as a client with default config.

Examples found in repository

examples/rpc_client_server.rs (line 87)

async fn run_client() -> Result<(), Box<dyn std::error::Error>> {
    println!("[Client] Connecting to RPC server");

    let transport = SharedMemoryFrameTransport::connect_client(SERVICE_NAME)?;
    let channel = MessageChannelAdapter::new(transport);

    let client = RpcClient::new(channel);
    let _handle = client.start();

    println!("[Client] Connected!\n");

    println!("[Client] Calling add(10, 32)");
    let resp: AddResponse = client.call("add", &AddRequest { a: 10, b: 32 }).await?;
    println!("[Client] Result: {}\n", resp.result);

    println!("[Client] Calling add(100, 200)");
    let resp: AddResponse = client.call("add", &AddRequest { a: 100, b: 200 }).await?;
    println!("[Client] Result: {}\n", resp.result);

    println!("[Client] Calling echo(\"Hello, RPC!\")");
    let resp: EchoResponse = client
        .call(
            "echo",
            &EchoRequest {
                message: "Hello, RPC!".to_string(),
            },
        )
        .await?;
    println!(
        "[Client] Result: message=\"{}\", length={}\n",
        resp.message, resp.length
    );

    println!("[Client] Calling unknown method");
    let result: Result<(), _> = client.call("unknown", &()).await;
    match result {
        Ok(_) => println!("[Client] Unexpected success"),
        Err(e) => println!("[Client] Got expected error: {}\n", e),
    }

    client.close().await?;
    println!("[Client] Done!");

    Ok(())
}

examples/message_transport_shm.rs (line 123)

async fn run_client() -> Result<(), Box<dyn std::error::Error>> {
    println!("[Client] Connecting to shared memory transport");

    let transport = SharedMemoryFrameTransport::connect_client(SERVICE_NAME)?;
    let channel = MessageChannelAdapter::new(transport);

    println!("[Client] Connected!");

    // Call: add (no compression)
    println!("\n[Client] Calling add (10 + 32)");
    let call = Message::call("add", AddRequest { a: 10, b: 32 })?;
    channel.send(&call).await?;

    let reply = channel.recv().await?;
    let resp: AddResponse = reply.deserialize_payload()?;
    println!("[Client] Result: {}", resp.result);

    // Call: add with LZ4 compression
    println!("\n[Client] Calling add with LZ4 compression (100 + 200)");
    let mut call = Message::call("add", AddRequest { a: 100, b: 200 })?;
    call.metadata.compression = CompressionType::Lz4;
    channel.send(&call).await?;

    let reply = channel.recv().await?;
    let resp: AddResponse = reply.deserialize_payload()?;
    println!(
        "[Client] Result: {} (compression: {:?})",
        resp.result, reply.metadata.compression
    );

    // Call: divide by zero (expect error)
    println!("\n[Client] Calling divide (10 / 0)");
    let call = Message::call("divide", AddRequest { a: 10, b: 0 })?;
    channel.send(&call).await?;

    let reply = channel.recv().await?;
    if reply.msg_type == MessageType::Error {
        let error_msg: String = reply.deserialize_payload()?;
        println!("[Client] Got error: {}", error_msg);
    }

    // Notification (fire-and-forget)
    println!("\n[Client] Sending notification log");
    let notification = Message::notification(
        "log",
        LogEvent {
            level: "INFO".to_string(),
            message: "Client started successfully".to_string(),
        },
    )?;
    channel.send(&notification).await?;
    println!("[Client] Notification sent (no response expected)");

    // Shutdown
    println!("\n[Client] Sending shutdown...");
    let call = Message::call("shutdown", ())?;
    channel.send(&call).await?;
    let _ = channel.recv().await?;
    println!("[Client] Done!");

    Ok(())
}

pub fn shm_is_healthy(&self) -> bool

pub fn shm_stats(&self) -> TransportStats

Trait Implementations

impl Debug for SharedMemoryFrameTransport

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl FrameTransport for SharedMemoryFrameTransport

fn is_healthy(&self) -> bool

Check if the transport is healthy.

fn send_frame<'life0, 'life1, 'async_trait>( &'life0 self, data: &'life1 [u8], ) -> Pin<Box<dyn Future<Output = TransportResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait,

Send a frame of bytes.

fn recv_frame<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = TransportResult<Bytes>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Receive a frame of bytes.

fn is_connected(&self) -> bool

Check if the transport is connected.

fn close<'life0, 'async_trait>( &'life0 self, ) -> Pin<Box<dyn Future<Output = TransportResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Close the transport connection.

fn stats(&self) -> Option<TransportStats>

Get transport statistics.

fn name(&self) -> &str

Get transport name/identifier.