koi-net 0.4.0

Local network toolkit: service discovery, DNS, health monitoring, TLS proxy, and certificate mesh
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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137

use std::sync::Arc;
use std::time::Duration;

use tokio::io::{AsyncBufReadExt, BufReader};

use tokio_util::sync::CancellationToken;

use koi_mdns::MdnsCore;

use super::dispatch;

/// IPC session grace period: 30 seconds.
const SESSION_GRACE: Duration = Duration::from_secs(30);

/// Start the IPC adapter with graceful shutdown support.
/// Windows: Named Pipe at `\\.\pipe\koi`
/// Unix: Unix Domain Socket at the given path
pub async fn start(
    core: Arc<MdnsCore>,
    path: std::path::PathBuf,
    cancel: CancellationToken,
) -> anyhow::Result<()> {
    #[cfg(unix)]
    {
        start_unix(core, path, cancel).await
    }
    #[cfg(windows)]
    {
        start_windows(core, path, cancel).await
    }
}

#[cfg(unix)]
async fn start_unix(
    core: Arc<MdnsCore>,
    path: std::path::PathBuf,
    cancel: CancellationToken,
) -> anyhow::Result<()> {
    // Remove stale socket file
    let _ = std::fs::remove_file(&path);

    let listener = tokio::net::UnixListener::bind(&path)?;
    tracing::info!(path = %path.display(), "IPC adapter listening (Unix socket)");

    loop {
        tokio::select! {
            result = listener.accept() => {
                let (stream, _addr) = result?;
                let core = core.clone();
                tokio::spawn(async move {
                    if let Err(e) = handle_connection(core, stream).await {
                        tracing::warn!(error = %e, "IPC connection error");
                    }
                });
            }
            _ = cancel.cancelled() => break,
        }
    }
    tracing::debug!("IPC adapter stopped (Unix)");
    Ok(())
}

#[cfg(unix)]
async fn handle_connection(
    core: Arc<MdnsCore>,
    stream: tokio::net::UnixStream,
) -> anyhow::Result<()> {
    let session_id = dispatch::new_session_id();
    let (reader, mut writer) = stream.into_split();
    let reader = BufReader::new(reader);
    let mut lines = reader.lines();

    while let Some(line) = lines.next_line().await? {
        let line = line.trim().to_string();
        if line.is_empty() {
            continue;
        }
        dispatch::handle_line(&core, &session_id, &line, SESSION_GRACE, &mut writer).await?;
    }
    core.session_disconnected(&session_id);
    Ok(())
}

#[cfg(windows)]
async fn start_windows(
    core: Arc<MdnsCore>,
    path: std::path::PathBuf,
    cancel: CancellationToken,
) -> anyhow::Result<()> {
    use tokio::net::windows::named_pipe::ServerOptions;

    let pipe_name = path.to_string_lossy();

    tracing::info!(pipe = %pipe_name, "IPC adapter listening (Named Pipe)");

    loop {
        let server = ServerOptions::new()
            .first_pipe_instance(false)
            .create(pipe_name.as_ref())?;

        tokio::select! {
            result = server.connect() => {
                result?;
                let core = core.clone();
                tokio::spawn(async move {
                    if let Err(e) = handle_pipe_connection(core, server).await {
                        tracing::warn!(error = %e, "IPC pipe connection error");
                    }
                });
            }
            _ = cancel.cancelled() => break,
        }
    }
    tracing::debug!("IPC adapter stopped (Named Pipe)");
    Ok(())
}

#[cfg(windows)]
async fn handle_pipe_connection(
    core: Arc<MdnsCore>,
    pipe: tokio::net::windows::named_pipe::NamedPipeServer,
) -> anyhow::Result<()> {
    let session_id = dispatch::new_session_id();
    let (reader, mut writer) = tokio::io::split(pipe);
    let reader = BufReader::new(reader);
    let mut lines = reader.lines();

    while let Some(line) = lines.next_line().await? {
        let line = line.trim().to_string();
        if line.is_empty() {
            continue;
        }
        dispatch::handle_line(&core, &session_id, &line, SESSION_GRACE, &mut writer).await?;
    }
    core.session_disconnected(&session_id);
    Ok(())
}