Both STREAM and ROUTER sockets suffer from a naming problem on outbound connections. While these connections can be created, they can't be immediately used. Traffic must be received before it can be sent. This prevents practical, minimal usage of STREAM or ROUTER as a true N fan in/out socket.

This change simply provides the user with a socket option that sets a user defined name of the next outbound connection:

zmq_setsockopt(routerSock,ZMQ_NEXT_IDENTITY,"myname",6);
if(0 > zmq_connect(routerSock,"tcp://127.0.0.1:1234")) return 1;
ret = zmq_send(routerSock,"myname",6,ZMQ_SNDMORE);
zmq_send(routerSock,b.mem,b.used,0);

In this example, the socket is immediately given the name "myname", and is capable of immediately sending traffic.

This approach is more effective in three ways:
1) It prevents all sorts of malicious peer naming attacks that can cause undefined behavior in existing ROUTER connections. (Two connections are made that both transmit the same name to the ROUTER, the ROUTER behavior is undefined)
2) It allows immediate control of connections made to external parties for STREAM sockets. Something that is not possible right now. Before an outbound connection had no name for STREAM or ROUTER sockets because outbound connections cannot be sent to without first receiving traffic.
3) It is simpler and more general than expecting two ROUTER sockets to handshake on assigned connection names. Plus it allows inline sending to new connections on ROUTER.

include/zmq.h

@@ -293,7 +293,7 @@ ZMQ_EXPORT int zmq_msg_set (zmq_msg_t *msg, int option, int optval);
 #define ZMQ_IPC_FILTER_PID 58
 #define ZMQ_IPC_FILTER_UID 59
 #define ZMQ_IPC_FILTER_GID 60
-
+#define ZMQ_NEXT_IDENTITY 56
 /*  Message options                                                           */
 #define ZMQ_MORE 1
 #define ZMQ_SRCFD 2

src/router.cpp

@@ -88,6 +88,12 @@ int zmq::router_t::xsetsockopt (int option_, const void *optval_,
     int value = is_int? *((int *) optval_): 0;
 
     switch (option_) {
+        case ZMQ_NEXT_IDENTITY:
+            if(optval_ && optvallen_) {
+                next_identity.assign((char*)optval_,optvallen_);
+                return 0;
+            }
+        break;
         case ZMQ_ROUTER_RAW:
             if (is_int && value >= 0) {
                 raw_sock = (value != 0);
@@ -382,8 +388,13 @@ bool zmq::router_t::identify_peer (pipe_t *pipe_)
     blob_t identity;
     bool ok;
 
-    if (options.raw_sock) { //  Always assign identity for raw-socket
-        unsigned char buf [5];
+    if (next_identity.length()) {
+        identity = blob_t((unsigned char*) next_identity.c_str(),
+            next_identity.length());
+        next_identity.clear();
+    }
+    else if (options.raw_sock) { //  Always assign identity for raw-socket
+       unsigned char buf [5];
         buf [0] = 0;
         put_uint32 (buf + 1, next_peer_id++);
         identity = blob_t (buf, sizeof buf);

src/stream.cpp

@@ -163,7 +163,25 @@ int zmq::stream_t::xsend (msg_t *msg_)
 
     return 0;
 }
+int zmq::stream_t::xsetsockopt (int option_, const void *optval_,
+    size_t optvallen_)
+{
+    bool is_int = (optvallen_ == sizeof (int));
+    int value = is_int? *((int *) optval_): 0;
 
+    switch (option_) {
+        case ZMQ_NEXT_IDENTITY:
+            if(optval_ && optvallen_) {
+                next_identity.assign((char*)optval_,optvallen_);
+                return 0;
+            }
+            break;
+        default:
+            break;
+    }
+    errno = EINVAL;
+    return -1;
+}
 int zmq::stream_t::xrecv (msg_t *msg_)
 {
     if (prefetched) {
@@ -244,12 +262,18 @@ void zmq::stream_t::identify_peer (pipe_t *pipe_)
     //  Always assign identity for raw-socket
     unsigned char buffer [5];
     buffer [0] = 0;
-    put_uint32 (buffer + 1, next_peer_id++);
-    blob_t identity = blob_t (buffer, sizeof buffer);
-
-    memcpy (options.identity, identity.data (), identity.size ());
-    options.identity_size = identity.size ();
-
+    blob_t identity;
+    if (next_identity.length()) {
+        identity = blob_t((unsigned char*) next_identity.c_str(),
+            next_identity.length());
+        next_identity.clear();
+    }
+    else { 
+        put_uint32 (buffer + 1, next_peer_id++);
+        blob_t identity = blob_t (buffer, sizeof buffer);
+        memcpy (options.identity, identity.data (), identity.size ());
+        options.identity_size = identity.size ();
+    }
     pipe_->set_identity (identity);
     //  Add the record into output pipes lookup table
     outpipe_t outpipe = {pipe_, true};

src/stream.hpp

@@ -47,7 +47,7 @@ namespace zmq
         void xread_activated (zmq::pipe_t *pipe_);
         void xwrite_activated (zmq::pipe_t *pipe_);
         void xpipe_terminated (zmq::pipe_t *pipe_);
-
+        int xsetsockopt (int option_, const void *optval_, size_t optvallen_); 
     private:
         //  Generate peer's id and update lookup map
         void identify_peer (pipe_t *pipe_);