gitgrip 0.19.0

Multi-repo workflow tool - manage multiple git repositories as one
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
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
138
139
140
141
142
143
144
145
146
147
148
149

//! Cross-platform transport for IPC.
//!
//! Ported from codi-rs with minimal changes. Abstracts Unix domain sockets
//! and Windows named pipes behind a common async IO trait.

use std::io;
use std::path::Path;

use tokio::io::{AsyncRead, AsyncWrite};

/// Trait alias for async read+write streams.
pub trait IpcIo: AsyncRead + AsyncWrite + Unpin + Send {}

impl<T> IpcIo for T where T: AsyncRead + AsyncWrite + Unpin + Send {}

/// Type-erased IPC stream.
pub type IpcStream = Box<dyn IpcIo>;

#[cfg(unix)]
use tokio::net::UnixListener;
#[cfg(unix)]
use tokio::net::UnixStream;

/// IPC listener that accepts incoming connections.
pub struct IpcListener {
    #[cfg(unix)]
    inner: UnixListener,
    #[cfg(windows)]
    name: String,
    #[cfg(windows)]
    first: std::sync::Mutex<Option<tokio::net::windows::named_pipe::NamedPipeServer>>,
}

/// Bind a listener to a socket/pipe path.
pub async fn bind(path: &Path) -> io::Result<IpcListener> {
    #[cfg(unix)]
    {
        if path.exists() {
            let _ = std::fs::remove_file(path);
        }

        if let Some(parent) = path.parent() {
            std::fs::create_dir_all(parent)?;
        }

        let inner = UnixListener::bind(path)?;
        Ok(IpcListener { inner })
    }

    #[cfg(windows)]
    {
        use tokio::net::windows::named_pipe::ServerOptions;
        let name = pipe_name_from_path(path);
        // Create the first pipe instance — must use first_pipe_instance(true).
        let first = ServerOptions::new()
            .first_pipe_instance(true)
            .create(&name)?;
        Ok(IpcListener {
            name,
            first: std::sync::Mutex::new(Some(first)),
        })
    }
}

/// Connect to a socket/pipe as a client.
pub async fn connect(path: &Path) -> io::Result<IpcStream> {
    #[cfg(unix)]
    {
        let stream = UnixStream::connect(path).await?;
        Ok(Box::new(stream))
    }

    #[cfg(windows)]
    {
        use tokio::net::windows::named_pipe::ClientOptions;
        let name = pipe_name_from_path(path);
        let pipe = ClientOptions::new().open(&name)?;
        Ok(Box::new(pipe))
    }
}

impl IpcListener {
    /// Accept a new connection.
    pub async fn accept(&self) -> io::Result<IpcStream> {
        #[cfg(unix)]
        {
            let (stream, _addr) = self.inner.accept().await?;
            Ok(Box::new(stream))
        }

        #[cfg(windows)]
        {
            use tokio::net::windows::named_pipe::ServerOptions;
            // Use the pre-created first instance if available, otherwise create a new one.
            let server = if let Some(first) = self.first.lock().unwrap().take() {
                first
            } else {
                ServerOptions::new()
                    .first_pipe_instance(false)
                    .create(&self.name)?
            };
            server.connect().await?;
            Ok(Box::new(server))
        }
    }
}

#[cfg(windows)]
fn pipe_name_from_path(path: &Path) -> String {
    let name = path.to_string_lossy().replace('/', "-").replace('\\', "-");
    format!(r"\\.\pipe\gitgrip-{}", name)
}

#[cfg(test)]
mod tests {
    use super::*;
    use tokio::io::{AsyncReadExt, AsyncWriteExt};

    #[cfg(unix)]
    #[tokio::test]
    async fn test_unix_socket_roundtrip() {
        let dir = tempfile::tempdir().unwrap();
        let sock = dir.path().join("test.sock");

        let listener = bind(&sock).await.unwrap();

        let client_task = tokio::spawn({
            let sock = sock.clone();
            async move {
                let mut stream = connect(&sock).await.unwrap();
                stream.write_all(b"hello\n").await.unwrap();

                let mut buf = vec![0u8; 64];
                let n = stream.read(&mut buf).await.unwrap();
                String::from_utf8_lossy(&buf[..n]).to_string()
            }
        });

        let mut server_stream = listener.accept().await.unwrap();
        let mut buf = vec![0u8; 64];
        let n = server_stream.read(&mut buf).await.unwrap();
        assert_eq!(&buf[..n], b"hello\n");

        server_stream.write_all(b"world\n").await.unwrap();

        let response = client_task.await.unwrap();
        assert_eq!(response, "world\n");
    }
}