interprocess_helpers

Wrapper around interprocess::local_socket::tokio for convenient communication between multiple Rust processes.

Use Cases

Whenever you need two processes to communicate with each other but you can't use stdin and stdout or other methods. An example would be communication between an elevated app and a non elevated one on Windows.

This crate also includes a helper function to easily bundle a different rust project with your main app. In comparison to interprocess, this crate is a little simpler to use as it abstracts more logic away (at the cost of flexibility). In contrast to local sockets in interprocess, this crate also allows the user to have multiple, clonable and sendable writers that send information to another process as opposed to just one.

Example

A main app (controller) communicates with another app (connector).

Main App (controller):

use interprocess::local_socket::traits::tokio::Stream;
use std::time::Duration;
use tokio::time::sleep;

use interprocess_helpers::{
    ManagedSender, Receiver,
    controller::{connect, get_socket_name},
    end_stream,
};

#[tokio::main(flavor = "current_thread")]
async fn main() -> anyhow::Result<()> {
    // Common name between controller and the connector app.
    let socket_name = &get_socket_name("connector");

    // Configure and launch the connector app.
    let mut cmd = get_connector_cmd();
    cmd.arg(socket_name);
    let mut child = cmd.spawn()?;

    // If the connection is not established in time, cancel the
    // connection. Note: in a real-world scenario you might want
     // to add a third branch that cancels the connection if the
     // connector app crashes
     tokio::select! {
         // Aquire a interprocess local socket stream.
         result = connect(socket_name) => {
            let (rx, tx) = result?.split();

            // ManagedSender lets you use a clonable sender in contrast
            // to the non-clonable one from interprocess. The interprocess
            // sender is managed within the manager variable.
            let (sender, manager) = ManagedSender::channel(tx, 1);
            let mut receiver = Receiver::from(rx);

            sender.send(b"Ping!".into()).await?;

            // This loop runs until the connector closes the connection
            // or it is cancelled
            while let Some(result) = receiver.next().await {
                assert_eq!(result?, "Pong!");
            }

            end_stream(receiver, manager).await?;

            Ok(())
        }
        _ = sleep(Duration::from_secs(20)) => {
            eprintln!("Could not connect: Connection timed out.");
            anyhow::Result::<()>::Ok(())
        }
    }?;

    // Wait for the connector to exit.
    let status = child.wait()?;
    println!("Connector exit code: {}", status.code().unwrap());

    Ok(())
}

Other App (connector):

use interprocess::local_socket::traits::tokio::Stream;
use std::{env, time::Duration};
use tokio::time::sleep;

// Careful: The controller app uses the `controller` module and connector app (this one) uses the `connector` module!
use interprocess_helpers::{ManagedSender, Receiver, connector::connect, end_stream};

#[tokio::main(flavor = "current_thread")]
async fn main() -> anyhow::Result<()> {
    let mut args = env::args();
    args.next();

    // Common name between controller app and connector app.
    let socket_name = args.next().unwrap();

    // If the connection is not established in time, cancel the
    // connection.
    tokio::select! {
        // Aquire a interprocess local socket stream.
        result = connect(&socket_name) => {
            let (rx, tx) = result?.split();

            // ManagedSender lets you use a clonable sender in contrast
            // to the non-clonable one from interprocess. The interprocess
            // sender is managed within the manager variable.
            let (sender, manager) = ManagedSender::channel(tx, 1);
            let mut receiver = Receiver::from(rx);

            // Read the first message that is received.
            // In a real.world scenario, it is recommended to use a
            // while loop in a new thread spawned thorugh tokio::spawn
            // to continuously receive messages without the rest of the
            // program interrupting.
            // Additional blocking tasks should be executed within a
            // tokio::spawn_blocking thread.
            let message = receiver.next().await.unwrap();
            assert_eq!(message?, "Ping!".to_string());

            sender.send(b"Pong!".into()).await?;

            end_stream(receiver, manager).await?;

            Ok(())
        }
        _ = sleep(Duration::from_secs(20)) => {
            panic!("Could not connect: Connection timed out.");
        }
    }
}

License: MIT OR Apache-2.0