Merge pull request #561 from shripchenko/master

Add ROUTER socket option to introduce self(send an empty message) to new peers, to allow ROUTER<->ROUTER auto-discovery problem.

include/zmq.h

@@ -274,6 +274,7 @@ ZMQ_EXPORT int zmq_msg_set (zmq_msg_t *msg, int option, int optval);
 #define ZMQ_CURVE_SERVER 47
 #define ZMQ_CURVE_PUBLICKEY 48
 #define ZMQ_CURVE_SERVERKEY 49
+#define ZMQ_ROUTER_ANNOUNCE_SELF 50
 
 /*  Message options                                                           */
 #define ZMQ_MORE 1

src/router.cpp

@@ -33,7 +33,8 @@ zmq::router_t::router_t (class ctx_t *parent_, uint32_t tid_, int sid_) :
     more_out (false),
     next_peer_id (generate_random ()),
     mandatory(false),
-    raw_sock(false)
+    raw_sock(false),
+    announce_self(false)
 {
     options.type = ZMQ_ROUTER;
     options.recv_identity = true;
@@ -94,6 +95,13 @@ int zmq::router_t::xsetsockopt (int option_, const void *optval_,
             printf ("E: invalid option value (int=%d value=%d)\n", is_int, value);
             break;
             
+        case ZMQ_ROUTER_ANNOUNCE_SELF:
+            if (is_int && value >= 0) {
+                announce_self = value;
+                return 0;
+            }
+            break;
+            
         default:
             break;
     }
@@ -389,6 +397,15 @@ bool zmq::router_t::identify_peer (pipe_t *pipe_)
     ok = outpipes.insert (outpipes_t::value_type (identity, outpipe)).second;
     zmq_assert (ok);
 
+    if (announce_self) {
+        msg_t tmp_;
+        tmp_.init ();
+        ok = pipe_->write (&tmp_);
+        zmq_assert (ok);
+        pipe_->flush ();
+        tmp_.close ();
+    };
+
     return true;
 }
 

src/router.hpp

@@ -112,6 +112,9 @@ namespace zmq
         bool mandatory;
         bool raw_sock;
 
+        // if true, send an empty message to every connected peer to solve 'who will write first' race condition
+        bool announce_self;
+
         router_t (const router_t&);
         const router_t &operator = (const router_t&);
     };