Add ZMQ_STREAM socket type

This is basically pull request #109, with the following tweaks added:

- Add example to Cargo.toml and remove "zmq" from directory name to more
  closely match current practice.

- Use ephemeral port bound to localhost, using `get_last_endpoint()` in
  test suite, to avoid potential collisions with test environment.

Fixes #95.

Author: karec

Cargo.toml

@@ -23,6 +23,10 @@ path = "examples/zguide/msreader/main.rs"
 name = "msgsend"
 path = "examples/msgsend/main.rs"
 
+[[example]]
+name = "stream-logserver"
+path = "examples/stream/logserver.rs"
+
 [[example]]
 name = "tasksink"
 path = "examples/zguide/tasksink/main.rs"

examples/stream/logserver.rs

@@ -0,0 +1,20 @@
+// Very basic example to listen tcp socket from zmq using STREAM sockets
+// You can use telnet to send messages and they will be output to console
+// ZMQ_STREAM socket will prepend socket identity on message, that's why we use recv_multipart here
+
+use std::str;
+extern crate zmq;
+
+
+fn main() {
+    println!("Hello, world!");
+
+    let ctx = zmq::Context::new();
+
+    let socket = ctx.socket(zmq::STREAM).unwrap();
+    socket.bind("tcp://*:8888").unwrap();
+    loop {
+        let data = socket.recv_multipart(0).unwrap();
+        println!("Identity: {:?} Message : {}", data[0], str::from_utf8(&data[1]).unwrap());
+    }
+}

src/lib.rs

@@ -54,6 +54,7 @@ pub enum SocketType {
     PUSH   = 8,
     XPUB   = 9,
     XSUB   = 10,
+    STREAM = 11,
 }
 
 impl Copy for SocketType {}
@@ -707,6 +708,7 @@ impl Socket {
                 8 => SocketType::PUSH,
                 9 => SocketType::XPUB,
                 10 => SocketType::XSUB,
+                11 => SocketType::STREAM,
                 _ => panic!("socket type is out of range!")
             }
         })

tests/test.rs

@@ -1,5 +1,6 @@
 extern crate zmq;
 use zmq::*;
+use std::net::TcpStream;
 
 fn create_socketpair() -> (Socket, Socket) {
     let ctx = Context::default();
@@ -151,14 +152,26 @@ fn test_get_socket_type() {
         SocketType::PULL,
         SocketType::PUSH,
         SocketType::XPUB,
-        SocketType::XSUB
+        SocketType::XSUB,
+        SocketType::STREAM,
     ];
     for sock_type in socket_types.drain(..) {
         let sock = ctx.socket(sock_type).unwrap();
         assert_eq!(sock.get_socket_type().unwrap(), sock_type);
     }
 }
 
+#[test]
+fn test_create_stream_socket() {
+    let ctx = Context::new();
+    let sock = ctx.socket(STREAM).unwrap();
+    assert!(sock.bind("tcp://127.0.0.1:*").is_ok());
+    let ep = sock.get_last_endpoint().unwrap().unwrap();
+    let tcp = "tcp://";
+    assert!(ep.starts_with(tcp));
+    assert!(TcpStream::connect(&ep[tcp.len()..]).is_ok());
+}
+
 #[test]
 fn test_getset_maxmsgsize() {
     let ctx = Context::new();