Add support for unix sockets

src/lib.rs

@@ -35,6 +35,7 @@ pub use messages::merkle::*;
 pub use messages::types::*;
 
 use server::serve;
+use server::serve_unix;
 
 /// Main Trait for an ABCI application. Provides generic responses for all callbacks
 /// Override desired callbacks as needed.  Tendermint makes 3 TCP connections to the
@@ -114,3 +115,11 @@ where
 {
     serve(app, listen_addr).unwrap();
 }
+
+/// Setup the application and start the server using unix domain socket.
+pub fn run_unix<A>(listen_addr: &str, app: A)
+where
+    A: Application + 'static + Send + Sync,
+{
+    serve_unix(app, listen_addr).unwrap();
+}

src/server.rs

@@ -7,8 +7,36 @@ use std::sync::{Arc, Mutex};
 use std::thread;
 
 use messages::abci::*;
+use std::os::unix::net::UnixListener;
 use stream::AbciStream;
 
+/// Creates the unix server and listens for connections from Tendermint
+pub fn serve_unix<A>(app: A, addr: &str) -> io::Result<()>
+where
+    A: Application + 'static + Send + Sync,
+{
+    let listener = UnixListener::bind(addr).unwrap();
+
+    // Wrap the app atomically and clone for each connection.
+    let app = Arc::new(Mutex::new(app));
+
+    for new_connection in listener.incoming() {
+        let app_instance = Arc::clone(&app);
+        match new_connection {
+            Ok(stream) => {
+                println!("Got connection! {:?}", stream);
+                thread::spawn(move || handle_stream(AbciStream::from(stream), &app_instance));
+            }
+            Err(err) => {
+                // We need all 3 connections...
+                panic!("Connection failed: {}", err);
+            }
+        }
+    }
+    drop(listener);
+    Ok(())
+}
+
 /// Creates the TCP server and listens for connections from Tendermint
 pub fn serve<A>(app: A, addr: SocketAddr) -> io::Result<()>
 where

src/stream.rs

@@ -7,18 +7,21 @@ use std::fmt::{self, Debug, Formatter};
 use std::io;
 use std::io::{Read, Write};
 use std::net::TcpStream;
+use std::os::unix::net::UnixStream;
 
 /// Wraps a stream for easier testing.  Original code from tomtau
 pub enum StreamWrapper {
     Mocked(SharedMockStream),
     Tcp(TcpStream),
+    Unix(UnixStream),
 }
 
 impl io::Read for StreamWrapper {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         match *self {
             StreamWrapper::Mocked(ref mut s) => s.read(buf),
             StreamWrapper::Tcp(ref mut s) => s.read(buf),
+            StreamWrapper::Unix(ref mut s) => s.read(buf),
         }
     }
 }
@@ -28,13 +31,15 @@ impl io::Write for StreamWrapper {
         match *self {
             StreamWrapper::Mocked(ref mut s) => s.write(buf),
             StreamWrapper::Tcp(ref mut s) => s.write(buf),
+            StreamWrapper::Unix(ref mut s) => s.write(buf),
         }
     }
 
     fn flush(&mut self) -> io::Result<()> {
         match *self {
             StreamWrapper::Mocked(ref mut s) => s.flush(),
             StreamWrapper::Tcp(ref mut s) => s.flush(),
+            StreamWrapper::Unix(ref mut s) => s.flush(),
         }
     }
 }
@@ -49,6 +54,7 @@ impl Debug for AbciStream {
         match self.stream {
             StreamWrapper::Mocked(_) => Ok(f.debug_struct("SharedMockStream").finish()?),
             StreamWrapper::Tcp(ref s) => s.fmt(f),
+            StreamWrapper::Unix(ref s) => s.fmt(f),
         }
     }
 }
@@ -99,6 +105,14 @@ impl From<TcpStream> for AbciStream {
     }
 }
 
+impl From<UnixStream> for AbciStream {
+    fn from(stream: UnixStream) -> AbciStream {
+        AbciStream {
+            stream: StreamWrapper::Unix(stream),
+        }
+    }
+}
+
 /// Parse out the varint. This code was adapted from the excellent integer-encoding crate
 fn read_varint(stream: &mut Read) -> Result<i64, io::Error> {
     const BUFLEN: usize = 10;