pub fn socket_read_messages<
    Message: DeserializeOwned + EnumerateInnerFileDescriptors + Clone,
    Handler: Fn(Message),
>(
    handler: Handler,
    socket: Arc<UnixStream>,
) {
    const MIN_READ_SIZE: usize = 16384;
    const INITIAL_SIZE: usize = 65536;
    let fd = socket.as_raw_fd();
    let mut decoder = FrameDecoder::<Message>::new();
    let mut buf = BytesMut::with_capacity(INITIAL_SIZE);
    let mut fds = Vec::<RawFd>::new();
    loop {
        while let Some(mut frame) = decoder.decode(&mut buf) {
            install_file_descriptors(&mut frame, &mut fds);
            handler(frame);
        }
        buf.reserve(MIN_READ_SIZE);
        let p = buf.as_mut_ptr();
        #[cfg(not(target_os = "linux"))]
        let flags = 0;
        #[cfg(target_os = "linux")]
        let flags = libc::MSG_NOSIGNAL;
        // The unsafe section is required due to raw handling of buffer contents,
        // calling directly into libc send/recv function. This is done because
        // at the time of writing there was no better way known to the author
        // to perform I/O without either silly thread-bouncing or superfluous
        // locking of read side against write side.
        let num_bytes_received = unsafe {
            let mut iov = libc::iovec {
                iov_base: p.add(buf.len()) as *mut std::ffi::c_void,
                iov_len: buf.capacity() - buf.len(),
            };
            let mut cmsgbuf: [u8; 4096] = [0; 4096];
            let mut hdr = libc::msghdr {
                msg_name: std::ptr::null_mut(),
                msg_namelen: 0,
                msg_iov: &mut iov,
                msg_iovlen: 1,
                msg_control: cmsgbuf.as_mut_ptr() as *mut std::ffi::c_void,
                msg_controllen: 4096,
                msg_flags: flags,
            };
            let num_bytes_received = libc::recvmsg(fd, &mut hdr, flags);
            if num_bytes_received > 0 && hdr.msg_controllen > 0 {
                let mut cmsg = libc::CMSG_FIRSTHDR(&hdr);
                while !cmsg.is_null() {
                    if (*cmsg).cmsg_level == libc::SOL_SOCKET
                        && (*cmsg).cmsg_type == libc::SCM_RIGHTS
                    {
                        let data = libc::CMSG_DATA(cmsg);
                        let len = (*cmsg).cmsg_len - libc::CMSG_LEN(0) as MsgControlLenType;
                        let mut pos = 0;
                        while pos + 4 <= len {
                            let src = std::slice::from_raw_parts(data.add(pos as usize), 4);
                            let mut raw: [libc::c_uchar; 4] = [0, 0, 0, 0];
                            raw.copy_from_slice(src);
                            let fd = RawFd::from_ne_bytes(raw);
                            fds.push(fd);
                            pos += 4;
                        }
                    }
                    cmsg = libc::CMSG_NXTHDR(&hdr, cmsg);
                }
            }
            num_bytes_received
        };
        if num_bytes_received > 0 {
            // The unsafe section is required because data was read into the buffer
            // (in above unsafe section). This is actually known to be correctly
            // initialized by system call already.
            unsafe {
                buf.set_len(buf.len() + (num_bytes_received as usize));
            }
        } else {
            break;
        }
    }
}
/// Reads from a unix stream socket and passes individual messages
/// to given handler.
Reads from a unix stream socket and passes individual messages
to given handler.
Reads from a unix stream socket and passes individual messages to given handler .
pub fn socket_read_messages<
    Message: DeserializeOwned + EnumerateInnerFileDescriptors + Clone,
    Handler: Fn(Message),
>(
    handler: Handler,
    socket: Arc<UnixStream>,
) {
    const MIN_READ_SIZE: usize = 16384;
    const INITIAL_SIZE: usize = 65536;
    let fd = socket.as_raw_fd();
    let mut decoder = FrameDecoder::<Message>::new();
    let mut buf = BytesMut::with_capacity(INITIAL_SIZE);
    let mut fds = Vec::<RawFd>::new();
    loop {
        while let Some(mut frame) = decoder.decode(&mut buf) {
            install_file_descriptors(&mut frame, &mut fds);
            handler(frame);
        }
        buf.reserve(MIN_READ_SIZE);
        let p = buf.as_mut_ptr();
        #[cfg(not(target_os = "linux"))]
        let flags = 0;
        #[cfg(target_os = "linux")]
        let flags = libc::MSG_NOSIGNAL;
        // The unsafe section is required due to raw handling of buffer contents,
        // calling directly into libc send/recv function. This is done because
        // at the time of writing there was no better way known to the author
        // to perform I/O without either silly thread-bouncing or superfluous
        // locking of read side against write side.
        let num_bytes_received = unsafe {
            let mut iov = libc::iovec {
                iov_base: p.add(buf.len()) as *mut std::ffi::c_void,
                iov_len: buf.capacity() - buf.len(),
            };
            let mut cmsgbuf: [u8; 4096] = [0; 4096];
            let mut hdr = libc::msghdr {
                msg_name: std::ptr::null_mut(),
                msg_namelen: 0,
                msg_iov: &mut iov,
                msg_iovlen: 1,
                msg_control: cmsgbuf.as_mut_ptr() as *mut std::ffi::c_void,
                msg_controllen: 4096,
                msg_flags: flags,
            };
            let num_bytes_received = libc::recvmsg(fd, &mut hdr, flags);
            if num_bytes_received > 0 && hdr.msg_controllen > 0 {
                let mut cmsg = libc::CMSG_FIRSTHDR(&hdr);
                while !cmsg.is_null() {
                    if (*cmsg).cmsg_level == libc::SOL_SOCKET
                        && (*cmsg).cmsg_type == libc::SCM_RIGHTS
                    {
                        let data = libc::CMSG_DATA(cmsg);
                        let len = (*cmsg).cmsg_len - libc::CMSG_LEN(0) as MsgControlLenType;
                        let mut pos = 0;
                        while pos + 4 <= len {
                            let src = std::slice::from_raw_parts(data.add(pos as usize), 4);
                            let mut raw: [libc::c_uchar; 4] = [0, 0, 0, 0];
                            raw.copy_from_slice(src);
                            let fd = RawFd::from_ne_bytes(raw);
                            fds.push(fd);
                            pos += 4;
                        }
                    }
                    cmsg = libc::CMSG_NXTHDR(&hdr, cmsg);
                }
            }
            num_bytes_received
        };
        if num_bytes_received > 0 {
            // The unsafe section is required because data was read into the buffer
            // (in above unsafe section). This is actually known to be correctly
            // initialized by system call already.
            unsafe {
                buf.set_len(buf.len() + (num_bytes_received as usize));
            }
        } else {
            break;
        }
    }
}
pub fn socket_read_messages < Message : DeserializeOwned + EnumerateInnerFileDescriptors + Clone , Handler : Fn ( Message ) , > ( handler : Handler , socket : Arc < UnixStream > , ) { const MIN_READ_SIZE : usize = 16384 ; const INITIAL_SIZE : usize = 65536 ; let fd = socket . as_raw_fd ( ) ; let mut decoder = FrameDecoder :: < Message > :: new ( ) ; let mut buf = BytesMut :: with_capacity ( INITIAL_SIZE ) ; let mut fds = Vec :: < RawFd > :: new ( ) ; loop { while let Some ( mut frame ) = decoder . decode ( & mut buf ) { install_file_descriptors ( & mut frame , & mut fds ) ; handler ( frame ) ; } buf . reserve ( MIN_READ_SIZE ) ; let p = buf . as_mut_ptr ( ) ; # [ cfg ( not ( target_os " " ) ) ] let flags = 0 ; # [ cfg ( target_os " " ) ] let flags = libc :: MSG_NOSIGNAL ; // The unsafe section is required due to raw handling of buffer contents, // calling directly into libc send/recv function. This is done because // at the time of writing there was no better way known to the author // to perform I/O without either silly thread-bouncing or superfluous // locking of read side against write side. let num_bytes_received = unsafe { let mut iov = libc :: iovec { iov_base : p . add ( buf . len ( ) ) as * mut std :: ffi :: c_void , iov_len : buf . capacity ( ) - buf . len ( ) , } ; let mut cmsgbuf : [ u8 ; 4096 ] = [ 0 ; 4096 ] ; let mut hdr = libc :: msghdr { msg_name : std :: ptr :: null_mut ( ) , msg_namelen : 0 , msg_iov : & mut iov , msg_iovlen : 1 , msg_control : cmsgbuf . as_mut_ptr ( ) as * mut std :: ffi :: c_void , msg_controllen : 4096 , msg_flags : flags , } ; let num_bytes_received = libc :: recvmsg ( fd , & mut hdr , flags ) ; if num_bytes_received > 0 && hdr . msg_controllen > 0 { let mut cmsg = libc :: CMSG_FIRSTHDR ( & hdr ) ; while ! cmsg . is_null ( ) { if ( * cmsg ) . cmsg_level == libc :: SOL_SOCKET && ( * cmsg ) . cmsg_type == libc :: SCM_RIGHTS { let data = libc :: CMSG_DATA ( cmsg ) ; let len = ( * cmsg ) . cmsg_len - libc :: CMSG_LEN ( 0 ) as MsgControlLenType ; let mut pos = 0 ; while pos + 4 <= len { let src = std :: slice :: from_raw_parts ( data . add ( pos as usize ) , 4 ) ; let mut raw : [ libc :: c_uchar ; 4 ] = [ 0 , 0 , 0 , 0 ] ; raw . copy_from_slice ( src ) ; let fd = RawFd :: from_ne_bytes ( raw ) ; fds . push ( fd ) ; pos += 4 ; } } cmsg = libc :: CMSG_NXTHDR ( & hdr , cmsg ) ; } } num_bytes_received } ; if num_bytes_received > 0 { // The unsafe section is required because data was read into the buffer // (in above unsafe section). This is actually known to be correctly // initialized by system call already. unsafe { buf . set_len ( buf . len ( ) + ( num_bytes_received as usize ) ) ; } } else { break ; } } }
Reads from a unix stream socket and passes individual messages
to given handler.
Reads from a unix stream socket and passes individual messages to given handler .
handler
Handler
socket
Arc<UnixStream>
handler
Handler
socket
Arc<UnixStream>