malkuth 0.1.0

A composable service-supervision toolkit for Rust
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

//! Local IPC transport via [`interprocess`] (Unix domain sockets on Unix, named
//! pipes on Windows), tokio async. Address forms: `ipc:/full/path` (filesystem
//! socket path) or `ipc:name` (a short / namespaced name).

use std::io;

use async_trait::async_trait;
use interprocess::local_socket::{
    tokio::{Listener as LocalSocketListener, Stream as LocalSocketStream},
    traits::tokio::{Listener as _, Stream as _},
    {GenericFilePath, GenericNamespaced, ListenerOptions, Name, ToFsName, ToNsName},
};

use crate::{Transport, WireConn, WireListener, codec::FramedConn};

fn name_err<E: std::fmt::Display>(e: E) -> io::Error {
    io::Error::new(
        io::ErrorKind::InvalidInput,
        format!("invalid local-socket name: {e}"),
    )
}

fn to_name(addr: &str) -> io::Result<Name<'_>> {
    let s = addr.strip_prefix("ipc:").unwrap_or(addr);
    if s.starts_with('/') {
        s.to_fs_name::<GenericFilePath>().map_err(name_err)
    } else {
        s.to_ns_name::<GenericNamespaced>().map_err(name_err)
    }
}

/// Local IPC transport.
pub struct IpcTransport;

#[async_trait]
impl Transport for IpcTransport {
    async fn listen(&self, addr: &str) -> io::Result<Box<dyn WireListener>> {
        let name = to_name(addr)?;
        #[cfg(unix)]
        if let Some(path) = addr.strip_prefix("ipc:").filter(|p| p.starts_with('/')) {
            let _ = std::fs::remove_file(path);
        }
        let listener = ListenerOptions::new().name(name).create_tokio()?;
        Ok(Box::new(IpcWireListener { listener }))
    }

    async fn connect(&self, addr: &str) -> io::Result<Box<dyn WireConn>> {
        let name = to_name(addr)?;
        let stream = LocalSocketStream::connect(name).await?;
        Ok(Box::new(FramedConn::new(stream)))
    }

    fn name(&self) -> &'static str {
        "ipc"
    }
}

pub struct IpcWireListener {
    listener: LocalSocketListener,
}

#[async_trait]
impl WireListener for IpcWireListener {
    async fn accept(&self) -> io::Result<Box<dyn WireConn>> {
        let stream = self.listener.accept().await?;
        Ok(Box::new(FramedConn::new(stream)))
    }

    fn local_addr(&self) -> io::Result<String> {
        Ok("ipc:local".to_string())
    }
}