Merge pull request #1357 from rodgert/master

Support limited metadata for STREAM sockets

.gitignore

@@ -101,6 +101,10 @@ test_xpub_nodrop
 test_xpub_manual
 test_xpub_welcome_msg
 test_atomics
+test_client_drop_more
+test_client_server
+test_server_drop_more
+test_thread_safe
 tests/test*.log
 tests/test*.trs
 src/platform.hpp*

src/stream.cpp

@@ -221,6 +221,12 @@ int zmq::stream_t::xrecv (msg_t *msg_)
     blob_t identity = pipe->get_identity ();
     rc = msg_->init_size (identity.size ());
     errno_assert (rc == 0);
+
+    // forward metadata (if any)
+    metadata_t *metadata = prefetched_msg.metadata();
+    if (metadata)
+        msg_->set_metadata(metadata);
+
     memcpy (msg_->data (), identity.data (), identity.size ());
     msg_->set_flags (msg_t::more);
 

src/stream_engine.cpp

@@ -36,7 +36,6 @@
 #include <string.h>
 #include <new>
 #include <sstream>
-#include <iostream>
 
 #include "stream_engine.hpp"
 #include "io_thread.hpp"
@@ -192,14 +191,21 @@ void zmq::stream_engine_t::plug (io_thread_t *io_thread_,
         handshaking = false;
 
         next_msg = &stream_engine_t::pull_msg_from_session;
-        process_msg = &stream_engine_t::push_msg_to_session;
+        process_msg = &stream_engine_t::push_raw_msg_to_session;
+
+        properties_t properties;
+        if (init_properties(properties)) {
+            //  Compile metadata.
+            zmq_assert (metadata == NULL);
+            metadata = new (std::nothrow) metadata_t (properties);
+        }
 
         if (options.raw_notify) {
             //  For raw sockets, send an initial 0-length message to the
             // application so that it knows a peer has connected.
             msg_t connector;
             connector.init();
-            push_msg_to_session (&connector);
+            push_raw_msg_to_session (&connector);
             connector.close();
             session->flush ();
         }
@@ -804,13 +810,8 @@ void zmq::stream_engine_t::mechanism_ready ()
     process_msg = &stream_engine_t::write_credential;
 
     //  Compile metadata.
-    typedef metadata_t::dict_t properties_t;
     properties_t properties;
-
-    //  If we have a peer_address, add it to metadata
-    if (!peer_address.empty()) {
-        properties.insert(std::make_pair("Peer-Address", peer_address));
-    }
+    init_properties(properties);
 
     //  Add ZAP properties.
     const properties_t& zap_properties = mechanism->get_zap_properties ();
@@ -835,6 +836,12 @@ int zmq::stream_engine_t::push_msg_to_session (msg_t *msg_)
     return session->push_msg (msg_);
 }
 
+int zmq::stream_engine_t::push_raw_msg_to_session (msg_t *msg_) {
+    if (metadata)
+        msg_->set_metadata(metadata);
+    return push_msg_to_session(msg_);
+}
+
 int zmq::stream_engine_t::write_credential (msg_t *msg_)
 {
     zmq_assert (mechanism != NULL);
@@ -938,6 +945,12 @@ void zmq::stream_engine_t::set_handshake_timer ()
     }
 }
 
+bool zmq::stream_engine_t::init_properties (properties_t & properties) {
+    if (peer_address.empty()) return false;
+    properties.insert (std::make_pair("Peer-Address", peer_address));
+    return true;
+}
+
 void zmq::stream_engine_t::timer_event (int id_)
 {
     zmq_assert (id_ == handshake_timer_id);

src/stream_engine.hpp

@@ -77,7 +77,6 @@ namespace zmq
         void timer_event (int id_);
 
     private:
-
         //  Unplug the engine from the session.
         void unplug ();
 
@@ -99,6 +98,8 @@ namespace zmq
         int pull_msg_from_session (msg_t *msg_);
         int push_msg_to_session (msg_t *msg);
 
+        int push_raw_msg_to_session (msg_t *msg);
+
         int write_credential (msg_t *msg_);
         int pull_and_encode (msg_t *msg_);
         int decode_and_push (msg_t *msg_);
@@ -113,6 +114,9 @@ namespace zmq
 
         void set_handshake_timer();
 
+        typedef metadata_t::dict_t properties_t;
+        bool init_properties (properties_t & properties);
+
         //  Underlying socket.
         fd_t s;
 

tests/test_stream.cpp

@@ -80,6 +80,9 @@ test_stream_to_dealer (void)
     assert (rc > 0);
     assert (zmq_msg_more (&identity));
 
+    // Verify the existence of Peer-Address metadata
+    assert (streq (zmq_msg_gets (&identity, "Peer-Address"), "127.0.0.1"));
+
     //  Second frame is zero
     byte buffer [255];
     rc = zmq_recv (stream, buffer, 255, 0);
@@ -91,6 +94,9 @@ test_stream_to_dealer (void)
     assert (rc > 0);
     assert (zmq_msg_more (&identity));
 
+    // Verify the existence of Peer-Address metadata
+    assert (streq (zmq_msg_gets (&identity, "Peer-Address"), "127.0.0.1"));
+
     //  Second frame is greeting signature
     rc = zmq_recv (stream, buffer, 255, 0);
     assert (rc == 10);