Add a channel_from_std API

This is more obvious to use from e.g. systemd socket activation, where we're passed a socket allocated outside the current process.

typed_channel: Make channel asymmetric, add symmetric_channel

To better handle the general case of asymmetric protocols.

I originally tried to just change channel() but it caused a lot of pain in type inference in the test suite. Change most of the tests to use a symmetric_channel() as a convenient shorthand, with one new test for an asymmetric channel().

https://github.com/mitsuhiko/tokio-unix-ipc/issues/2

Attempt to make Bootstrapper thread safe.

1. Bootstrapper.sender change from RefCell to tokio::sync::Mutex.
2. Add an example (async_trait_example.rs).

I am not familiar with unix domain socket programming, maybe IPC_FDS and PANIC_INFO should change to tokio::sync::Mutex?

I have several projects that might make use of this crate, and one of them today uses SO_PEERCRED to validate the sender's identity.

Add basic support for this.

I chose to make a small new struct for credentials instead of exposing nix in our public types (perhaps we want to use rustix in the future, etc.)

Closes: https://github.com/mitsuhiko/tokio-unix-ipc/issues/4

This method is just a wrapper around already public APIs, and it's convenient to use directly; in my case I have a socket already passed via systemd socket activation.

If I try to do the code snippet below, to have a background task wait for a message over IPC I get a compiler error:

`Cell<isize>` cannot be shared between threads safely

Is there any reason the sender field of Bootstrapper can't be changed from a RefCell to an Arc which would make this possible? Or is there a different way to go about what I am trying to achieve?

tokio::spawn(async move {
        let bootstrapper = {
            let bootstrapper = tokio_unix_ipc::Bootstrapper::bind("/tmp/test-uds").unwrap();
            bootstrapper.send(Message::Sender(tx)).await.unwrap();
            bootstrapper
        };

        tokio::select! {
            res = rx.recv() => {
                if let Ok(msg) = res {
                    match msg {
                        _ => {}
                    }
                }
            }
        };
});

I tried to send the file descriptor using Bootstrapper.send() method but it doesn't work. Bootstrapper.send() method seems to transmit the file descriptor as an ordinary integer, without conversion by the operating system kernel. Similar to the functionality of this crate: https://github.com/polachok/passfd.

There is another question I wonder: Why is the Bootstrapper#sender wrapped in RefCell instead of Arc, so that Bootstrapper cannot be used in asynchronous traits.

Looking forward to your reply!

In one project I'm using SO_PEERCRED: https://github.com/coreos/bootupd/blob/main/src/ipc.rs - looking at using this crate instead. What do you think about exposing that via an API?

The usual pattern here is to pass credentials on the first message. In the above code I represented this as a state machine encoded via structs; e.g. perhaps we'd add RawAuthenticatingSender and RawUnauthenticatedReceiver, where e.g.

impl RawUnauthenticatedReceiver {
   pub async fn recv(self) -> Result<(RawReceiver, Credentials, Vec<u8>, Option<Vec<RawFd>>)>

Note this consumes the unauthenticated type and returns credentials. Would need to debate exposing e.g. https://docs.rs/nix/latest/nix/sys/socket/struct.UnixCredentials.html or (probably better for semver) a new struct here.

Alternatively we could add recv_with_creds() and send_with_creds() to the raw types.

I want to wrap Bootstrapper in a async trait, which is consistent of RefCell<Option<RawSender>> and is not thread safe.

Here is a complete demo, TaskService is an asynchronous trait of Task abstraction. IpcSender is wrap of Bootstrapper, and IpcReceiver is wrap of Receiver<Task>. Both of them implement TaskService to send or receive ipc request.

use std::{env, process};
use tokio_unix_ipc::{Bootstrapper, channel, Receiver, Sender};
use serde::Serialize;
use serde::Deserialize;

const ENV_VAR: &str = "PROC_CONNECT_TO";

#[derive(Serialize, Deserialize)]
pub enum Task {
    Sum(Vec<i64>, Sender<i64>),
    Shutdown
}

#[async_trait::async_trait]
pub trait TaskService {
    async fn sum(&self, vec: Vec<i64>) -> i64;
    async fn shutdown(&self);
}

#[tokio::main]
async fn main() {
    if let Ok(path) = env::var(ENV_VAR) {
        let server = IpcReceiver::new(path).await;
        server.start().await;
    } else {
        let client = IpcSender::new();
        let mut child = process::Command::new(env::current_exe().unwrap())
            .env(ENV_VAR, client.bootstrap.path())
            .spawn()
            .unwrap();

        let sum = client.sum(vec![1, 2, 3]).await;
        println!("sum = {}", sum);

        child.kill().ok();
        child.wait().ok();
    }
}


/// ipc_receiver
pub struct IpcReceiver {
    pub recv: Receiver<Task>,
}
impl IpcReceiver {
    pub async fn new(path: String) -> Self {
        let receiver = Receiver::<Task>::connect(path).await.unwrap();
        IpcReceiver {
            recv: receiver,
        }
    }
    pub async fn start(&self) {
        loop {
            let task = self.recv.recv().await.unwrap();
            match task {
                Task::Sum(vec, tx) => {
                    let sum = self.sum(vec).await;
                    tx.send(sum).await.unwrap();
                },
                Task::Shutdown => {
                    self.shutdown().await;
                    break
                }
            }
        }
    }
}
#[async_trait::async_trait]
impl TaskService for IpcReceiver {
    async fn sum(&self, vec: Vec<i64>) -> i64 {
        vec.into_iter().sum::<i64>()
    }

    async fn shutdown(&self) {
    }
}

/// ipc_sender
pub struct IpcSender {
    pub bootstrap: Bootstrapper,
}
impl IpcSender {
    pub fn new() -> Self {
        let bootstrapper = Bootstrapper::new().unwrap();
        IpcSender {
            bootstrap: bootstrapper
        }
    }
}
#[async_trait::async_trait]
impl TaskService for IpcSender {
    async fn sum(&self, vec: Vec<i64>) -> i64 {
        let (tx, rx) = channel().unwrap();
        self.bootstrap.send(Task::Sum(vec, tx));
        rx.recv().await.unwrap()
    }

    async fn shutdown(&self) {
        self.bootstrap.send(Task::Shutdown).await.unwrap()
    }
}

I wonder if Bootstrapper#sender can change to Mutex. Maybe I can modify it later.

https://docs.rs/unix-ipc/latest/unix-ipc/struct.Handle.html is wrong, should be https://docs.rs/unix-ipc/0.2.0/unix_ipc/struct.Handle.html

see also https://github.com/mitsuhiko/unix-ipc/pull/2

Tokio's UnixStream::from_std does not set the stream to non-blocking, it's up to the caller to do so: https://docs.rs/tokio/latest/tokio/net/struct.UnixStream.html#method.from_std

They seem unwilling to fix this particular footgun because of backwards compatibility, but tokio_unix_ipc at least could set the stream to non-blocking so users don't have to remember to do so.

The protocol isn't documented today as far as I can tell. But I think this crate is implicitly committing to its stability - because any time one is doing IPC, the possibility of version drift between the binaries (and hence versions of this crate) comes into play. I'm sure the protocol wouldn't change across Rust semvers of this crate - but do you see it changing ever?

In one use case I have that's similar to this crate, I have the client send the (device, inode) pair of /proc/self/exe over to the server, and the server verifies they match. (This also of course covers the higher-level protocol that applications would use on top of this crate; a related but distinct issue)

One reason I ask this is I have a use case for doing IPC between Rust and Go, and I'm today doing JSON over SOCK_SEQPACKET but...while the great thing about SEQPACKET is the kernel takes care of the framing, it has less support in the userspace ecosystem, and a simple "[header with length][payload]" protocol has well-understood tooling because it's how a vast array of TCP protocols work.

TL;DR: Would you say we could document the "raw" protocol used here and commit to its stability approximately ~forever?