Updated man entries, and added behavior in case the user duplicates peer ids. …

Updated man entries, and added behavior in case the user duplicates peer ids.  In this case the socket reverts to default behavior.

doc/zmq_setsockopt.txt

@@ -67,6 +67,32 @@ Default value:: 100
 Applicable socket types:: all, only for connection-oriented transports.
 
 
+ZMQ_CONNECT_RID: Assign the next outbound connection id 
+~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+The 'ZMQ_CONNECT_RID' option sets the peer id of the next host connected 
+via the zmq_connect() call, and immediately readies that connection for 
+data transfer with the named id.  This option applies only to the first
+subsequent call to zmq_connect(), calls thereafter use default connection 
+behavior. 
+
+Typical use is to set this socket option on each zmq_connect() attempt 
+to a new host. Each connection should be assigned a unique name. Duplicated 
+names will trigger default connection behavior.
+
+Useful when connecting ROUTER to ROUTER, or STREAM to STREAM, as it 
+allows for immediate sending to peers. Outbound id framing requirements 
+for ROUTER and STREAM sockets apply.
+
+The peer id should be from 1 to 255 bytes long and MAY NOT start with 
+binary zero.
+
+[horizontal]
+Option value type:: binary data
+Option value unit:: N/A
+Default value:: NULL
+Applicable socket types:: ZMQ_ROUTER, ZMQ_STREAM
+
+
 ZMQ_CONFLATE: Keep only last message
 ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
 If set, a socket shall keep only one message in its inbound/outbound

src/router.cpp

@@ -387,13 +387,15 @@ bool zmq::router_t::identify_peer (pipe_t *pipe_)
     msg_t msg;
     blob_t identity;
     bool ok;
-    bool connect_rid_used = false;
 
     if (connect_rid.length()) {
         identity = blob_t ((unsigned char*) connect_rid.c_str (),
             connect_rid.length());
         connect_rid.clear ();
-        connect_rid_used = true;
+        outpipes_t::iterator it = outpipes.find (identity);
+        if (it != outpipes.end ()) {
+            return false; // duplicate connection
+        }
     }
     else 
     if (options.raw_sock) { //  Always assign identity for raw-socket

src/stream.cpp

@@ -266,9 +266,12 @@ void zmq::stream_t::identify_peer (pipe_t *pipe_)
         identity = blob_t ((unsigned char*) connect_rid.c_str(),
             connect_rid.length ());
         connect_rid.clear ();
+        outpipes_t::iterator it = outpipes.find (identity);
+        if (it != outpipes.end ()) 
+            goto d;
     }
     else {
-        put_uint32 (buffer + 1, next_rid++);
+d:      put_uint32 (buffer + 1, next_rid++);
         identity = blob_t (buffer, sizeof buffer);
         memcpy (options.identity, identity.data (), identity.size ());
         options.identity_size = identity.size ();