merge to upstream

src/stream_engine.hpp

@@ -30,6 +30,7 @@
 #include "options.hpp"
 #include "socket_base.hpp"
 #include "../include/zmq.h"
+#include "metadata.hpp"
 
 namespace zmq
 {
@@ -67,6 +68,7 @@ namespace zmq
         //  i_poll_events interface implementation.
         void in_event ();
         void out_event ();
+        void timer_event (int id_);
 
         // export s via i_engine so it is possible to link a pipe to fd
         fd_t get_assoc_fd (){ return s; };
@@ -94,8 +96,8 @@ namespace zmq
         //  Zero indicates the peer has closed the connection.
         int read (void *data_, size_t size_);
 
-        int read_identity (msg_t *msg_);
-        int write_identity (msg_t *msg_);
+        int identity_msg (msg_t *msg_);
+        int process_identity_msg (msg_t *msg_);
 
         int next_handshake_command (msg_t *msg);
         int process_handshake_command (msg_t *msg);
@@ -115,6 +117,8 @@ namespace zmq
         size_t add_property (unsigned char *ptr,
             const char *name, const void *value, size_t value_len);
 
+        void set_handshake_timer();
+
         //  Underlying socket.
         fd_t s;
 
@@ -133,6 +137,9 @@ namespace zmq
         size_t outsize;
         i_encoder *encoder;
 
+        //  Metadata to be attached to received messages. May be NULL.
+        metadata_t *metadata;
+
         //  When true, we are still trying to determine whether
         //  the peer is using versioned protocol, and if so, which
         //  version.  When false, normal message flow has started.
@@ -166,9 +173,9 @@ namespace zmq
 
         bool plugged;
 
-        int (stream_engine_t::*read_msg) (msg_t *msg_);
+        int (stream_engine_t::*next_msg) (msg_t *msg_);
 
-        int (stream_engine_t::*write_msg) (msg_t *msg_);
+        int (stream_engine_t::*process_msg) (msg_t *msg_);
 
         bool io_error;
 
@@ -185,6 +192,12 @@ namespace zmq
         //  True iff the engine doesn't have any message to encode.
         bool output_stopped;
 
+        //  ID of the handshake timer
+        enum {handshake_timer_id = 0x40};
+
+        //  True is linger timer is running.
+        bool has_handshake_timer;
+
         // Socket
         zmq::socket_base_t *socket;