/// Reads from a unix stream socket and demultiplex messages to handlers.
Reads from a unix stream socket and demultiplex messages to handlers.
Reads from a unix stream socket and demultiplex messages to handlers .
pub fn socket_read_demux<
    Request,
    Reply,
    Message: MessageDemux<Request, Reply> + DeserializeOwned + Clone,
>(
    hdl: Demux<Request, Reply, Message>,
    socket: Arc<UnixStream>,
) {
    let fd = socket.as_raw_fd();
    let mut decoder = FrameDecoder::<Message>::new();
    let mut buf = BytesMut::new();
    loop {
        while let Some(frame) = decoder.decode(&mut buf) {
            hdl.handle(frame);
        }
        if buf.capacity() < 4096 {
            buf.reserve(4096);
        }
        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 count = unsafe {
            libc::recv(
                fd,
                p.add(buf.len()) as *mut std::ffi::c_void,
                buf.capacity(),
                flags,
            )
        };
        if count > 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() + (count as usize));
            }
        } else {
            break;
        }
    }
}
pub fn socket_read_demux < Request , Reply , Message : MessageDemux < Request , Reply > + DeserializeOwned + Clone , > ( hdl : Demux < Request , Reply , Message > , socket : Arc < UnixStream > , ) { let fd = socket . as_raw_fd ( ) ; let mut decoder = FrameDecoder :: < Message > :: new ( ) ; let mut buf = BytesMut :: new ( ) ; loop { while let Some ( frame ) = decoder . decode ( & mut buf ) { hdl . handle ( frame ) ; } if buf . capacity ( ) < 4096 { buf . reserve ( 4096 ) ; } 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 count = unsafe { libc :: recv ( fd , p . add ( buf . len ( ) ) as * mut std :: ffi :: c_void , buf . capacity ( ) , flags , ) } ; if count > 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 ( ) + ( count as usize ) ) ; } } else { break ; } } }
Reads from a unix stream socket and demultiplex messages to handlers.
Reads from a unix stream socket and demultiplex messages to handlers .
hdl
Demux<Request, Reply, Message>
socket
Arc<UnixStream>