zeromq 0.6.0-pre.1

A native Rust implementation of ZeroMQ
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

#[cfg(feature = "tokio-runtime")]
use tokio::net::{UnixListener, UnixStream};

#[cfg(any(feature = "async-std-runtime", feature = "async-dispatcher-runtime"))]
use async_std::os::unix::net::{UnixListener, UnixStream};

use super::make_framed;
use super::AcceptStopHandle;
use crate::async_rt;
use crate::codec::FramedIo;
use crate::endpoint::Endpoint;
use crate::task_handle::TaskHandle;
use crate::ZmqResult;

use futures::channel::oneshot;
use futures::{select, FutureExt};

use std::path::Path;

pub(crate) async fn connect(path: &Path) -> ZmqResult<(FramedIo, Endpoint)> {
    let raw_socket = UnixStream::connect(path).await?;
    let peer_addr = raw_socket.peer_addr()?;
    let peer_addr = peer_addr.as_pathname().map(|a| a.to_owned());

    Ok((make_framed(raw_socket), Endpoint::Ipc(peer_addr)))
}

pub(crate) async fn begin_accept<T>(
    path: &Path,
    cback: impl Fn(ZmqResult<(FramedIo, Endpoint)>) -> T + Send + 'static,
) -> ZmqResult<(Endpoint, AcceptStopHandle)>
where
    T: std::future::Future<Output = ()> + Send + 'static,
{
    let wildcard: &Path = "*".as_ref();
    if path == wildcard {
        todo!("Need to implement support for wildcard paths!");
    }

    #[cfg(feature = "tokio-runtime")]
    let listener = UnixListener::bind(path)?;
    #[cfg(any(feature = "async-std-runtime", feature = "async-dispatcher-runtime"))]
    let listener = UnixListener::bind(path).await?;

    let resolved_addr = listener.local_addr()?;
    let resolved_addr = resolved_addr.as_pathname().map(|a| a.to_owned());
    let listener_addr = resolved_addr.clone();
    let (stop_channel, stop_callback) = oneshot::channel::<()>();
    let task_handle = async_rt::task::spawn(async move {
        let mut stop_callback = stop_callback.fuse();
        loop {
            select! {
                incoming = listener.accept().fuse() => {
                    let maybe_accepted: Result<_, _> = incoming.map(|(raw_socket, peer_addr)| {
                        let peer_addr = peer_addr.as_pathname().map(|a| a.to_owned());
                        (make_framed(raw_socket), Endpoint::Ipc(peer_addr))
                    }).map_err(|err| err.into());
                    async_rt::task::spawn(cback(maybe_accepted));
                },
                _ = stop_callback => {
                    log::debug!("Accept task received stop signal. {:?}", listener_addr);
                    break
                }
            }
        }
        drop(listener);
        if let Some(listener_addr) = listener_addr {
            #[cfg(any(feature = "async-std-runtime", feature = "async-dispatcher-runtime"))]
            use async_std::fs::remove_file;
            #[cfg(feature = "tokio-runtime")]
            use tokio::fs::remove_file;

            if let Err(err) = remove_file(&listener_addr).await {
                log::warn!(
                    "Could not delete unix socket at {}: {}",
                    listener_addr.display(),
                    err
                );
            }
        }
        Ok(())
    });
    Ok((
        Endpoint::Ipc(resolved_addr),
        AcceptStopHandle(TaskHandle::new(stop_channel, task_handle)),
    ))
}