diff --git a/src/Makefile.am b/src/Makefile.am
index 41fa3fa2..b3e9d166 100644
--- a/src/Makefile.am
+++ b/src/Makefile.am
@@ -39,9 +39,11 @@ libzmq_la_SOURCES = \
lb.hpp \
likely.hpp \
mailbox.hpp \
+ mechanism.hpp \
msg.hpp \
mtrie.hpp \
mutex.hpp \
+ null_mechanism.hpp \
object.hpp \
options.hpp \
own.hpp \
@@ -49,6 +51,7 @@ libzmq_la_SOURCES = \
pgm_sender.hpp \
pgm_socket.hpp \
pipe.hpp \
+ plain_mechanism.hpp \
platform.hpp \
poll.hpp \
poller.hpp \
@@ -100,8 +103,10 @@ libzmq_la_SOURCES = \
kqueue.cpp \
lb.cpp \
mailbox.cpp \
+ mechanism.cpp \
msg.cpp \
mtrie.cpp \
+ null_mechanism.cpp \
object.cpp \
options.cpp \
own.cpp \
@@ -110,6 +115,7 @@ libzmq_la_SOURCES = \
pgm_sender.cpp \
pgm_socket.cpp \
pipe.cpp \
+ plain_mechanism.cpp \
poll.cpp \
poller_base.cpp \
pull.cpp \
diff --git a/src/ipc_connecter.cpp b/src/ipc_connecter.cpp
index 716fef0f..30841cd1 100644
--- a/src/ipc_connecter.cpp
+++ b/src/ipc_connecter.cpp
@@ -112,7 +112,8 @@ void zmq::ipc_connecter_t::out_event ()
return;
}
// Create the engine object for this connection.
- stream_engine_t *engine = new (std::nothrow) stream_engine_t (fd, options, endpoint);
+ stream_engine_t *engine = new (std::nothrow)
+ stream_engine_t (fd, options, false, endpoint);
alloc_assert (engine);
// Attach the engine to the corresponding session object.
diff --git a/src/ipc_listener.cpp b/src/ipc_listener.cpp
index b3ad88bc..d503d73e 100644
--- a/src/ipc_listener.cpp
+++ b/src/ipc_listener.cpp
@@ -80,7 +80,8 @@ void zmq::ipc_listener_t::in_event ()
}
// Create the engine object for this connection.
- stream_engine_t *engine = new (std::nothrow) stream_engine_t (fd, options, endpoint);
+ stream_engine_t *engine = new (std::nothrow)
+ stream_engine_t (fd, options, true, endpoint);
alloc_assert (engine);
// Choose I/O thread to run connecter in. Given that we are already
diff --git a/src/mechanism.cpp b/src/mechanism.cpp
new file mode 100644
index 00000000..c89c7817
--- /dev/null
+++ b/src/mechanism.cpp
@@ -0,0 +1,73 @@
+/*
+ Copyright (c) 2007-2013 Contributors as noted in the AUTHORS file
+
+ This file is part of 0MQ.
+
+ 0MQ is free software; you can redistribute it and/or modify it under
+ the terms of the GNU Lesser General Public License as published by
+ the Free Software Foundation; either version 3 of the License, or
+ (at your option) any later version.
+
+ 0MQ is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ GNU Lesser General Public License for more details.
+
+ You should have received a copy of the GNU Lesser General Public License
+ along with this program. If not, see .
+*/
+
+#include
+
+#include "mechanism.hpp"
+#include "options.hpp"
+#include "msg.hpp"
+#include "err.hpp"
+#include "wire.hpp"
+
+zmq::mechanism_t::mechanism_t (const options_t &options_) :
+ options (options_)
+{
+}
+
+zmq::mechanism_t::~mechanism_t ()
+{
+}
+
+void zmq::mechanism_t::set_peer_identity (const void *id_ptr, size_t id_size)
+{
+ identity = blob_t (static_cast (id_ptr), id_size);
+}
+
+void zmq::mechanism_t::peer_identity (msg_t *msg_)
+{
+ const int rc = msg_->init_size (identity.size ());
+ errno_assert (rc == 0);
+ memcpy (msg_->data (), identity.data (), identity.size ());
+ msg_->set_flags (msg_t::identity);
+}
+
+const char *zmq::mechanism_t::socket_type_string (int socket_type) const
+{
+ static const char *names [] = {"PAIR", "PUB", "SUB", "REQ", "REP",
+ "DEALER", "ROUTER", "PULL", "PUSH",
+ "XPUB", "XSUB"};
+ zmq_assert (socket_type >= 0 && socket_type <= 10);
+ return names [socket_type];
+}
+
+size_t zmq::mechanism_t::add_property (unsigned char *ptr, const char *name,
+ const void *value, size_t value_len) const
+{
+ const size_t name_len = strlen (name);
+ zmq_assert (name_len <= 255);
+ *ptr++ = static_cast (name_len);
+ memcpy (ptr, name, name_len);
+ ptr += name_len;
+ zmq_assert (value_len <= (2^31) - 1);
+ put_uint32 (ptr, static_cast (value_len));
+ ptr += 4;
+ memcpy (ptr, value, value_len);
+
+ return 1 + name_len + 4 + value_len;
+}
diff --git a/src/mechanism.hpp b/src/mechanism.hpp
new file mode 100644
index 00000000..28ed0e14
--- /dev/null
+++ b/src/mechanism.hpp
@@ -0,0 +1,73 @@
+/*
+ Copyright (c) 2007-2013 Contributors as noted in the AUTHORS file
+
+ This file is part of 0MQ.
+
+ 0MQ is free software; you can redistribute it and/or modify it under
+ the terms of the GNU Lesser General Public License as published by
+ the Free Software Foundation; either version 3 of the License, or
+ (at your option) any later version.
+
+ 0MQ is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ GNU Lesser General Public License for more details.
+
+ You should have received a copy of the GNU Lesser General Public License
+ along with this program. If not, see .
+*/
+
+#ifndef __ZMQ_MECHANISM_HPP_INCLUDED__
+#define __ZMQ_MECHANISM_HPP_INCLUDED__
+
+#include
+
+#include "options.hpp"
+#include "blob.hpp"
+
+namespace zmq
+{
+
+ // Abstract class representing security mechanism.
+ // Different mechanism extedns this class.
+
+ class msg_t;
+
+ class mechanism_t
+ {
+ public:
+
+ mechanism_t (const options_t &options_);
+
+ virtual ~mechanism_t ();
+
+ // Prepare next handshake message that is to be sent to the peer.
+ virtual int next_handshake_message (msg_t *msg_) = 0;
+
+ // Process the handshake message received from the peer.
+ virtual int process_handshake_message (msg_t *msg_) = 0;
+
+ // True iff the handshake stage is complete?
+ virtual bool is_handshake_complete () const = 0;
+
+ void set_peer_identity (const void *id_ptr, size_t id_size);
+
+ void peer_identity (msg_t *msg_);
+
+ protected:
+
+ const char *socket_type_string (int socket_type) const;
+
+ size_t add_property (unsigned char *ptr, const char *name,
+ const void *value, size_t value_len) const;
+
+ options_t options;
+
+ private:
+
+ blob_t identity;
+ };
+
+}
+
+#endif
diff --git a/src/null_mechanism.cpp b/src/null_mechanism.cpp
new file mode 100644
index 00000000..e0f7ef52
--- /dev/null
+++ b/src/null_mechanism.cpp
@@ -0,0 +1,152 @@
+/*
+ Copyright (c) 2007-2013 Contributors as noted in the AUTHORS file
+
+ This file is part of 0MQ.
+
+ 0MQ is free software; you can redistribute it and/or modify it under
+ the terms of the GNU Lesser General Public License as published by
+ the Free Software Foundation; either version 3 of the License, or
+ (at your option) any later version.
+
+ 0MQ is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ GNU Lesser General Public License for more details.
+
+ You should have received a copy of the GNU Lesser General Public License
+ along with this program. If not, see .
+*/
+
+#include "platform.hpp"
+#ifdef ZMQ_HAVE_WINDOWS
+#include "windows.hpp"
+#endif
+
+#include
+#include
+#include
+
+#include "err.hpp"
+#include "msg.hpp"
+#include "wire.hpp"
+#include "null_mechanism.hpp"
+
+zmq::null_mechanism_t::null_mechanism_t (const options_t &options_) : mechanism_t (options_),
+ ready_command_sent (false),
+ ready_command_received (false)
+{
+}
+
+zmq::null_mechanism_t::~null_mechanism_t ()
+{
+}
+
+int zmq::null_mechanism_t::next_handshake_message (msg_t *msg_)
+{
+ if (ready_command_sent) {
+ errno = EAGAIN;
+ return -1;
+ }
+
+ unsigned char * const command_buffer = (unsigned char *) malloc (512);
+ alloc_assert (command_buffer);
+
+ unsigned char *ptr = command_buffer;
+
+ // Add mechanism string
+ memcpy (ptr, "READY ", 8);
+ ptr += 8;
+
+ // Add socket type property
+ const char *socket_type = socket_type_string (options.type);
+ ptr += add_property (ptr, "Socket-Type", socket_type, strlen (socket_type));
+
+ // Add identity property
+ if (options.type == ZMQ_REQ
+ || options.type == ZMQ_DEALER
+ || options.type == ZMQ_ROUTER) {
+ ptr += add_property (ptr, "Identity",
+ options.identity, options.identity_size);
+ }
+
+ const size_t command_size = ptr - command_buffer;
+ const int rc = msg_->init_size (command_size);
+ errno_assert (rc == 0);
+ memcpy (msg_->data (), command_buffer, command_size);
+ free (command_buffer);
+
+ ready_command_sent = true;
+
+ return 0;
+}
+
+int zmq::null_mechanism_t::process_handshake_message (msg_t *msg_)
+{
+ if (ready_command_received) {
+ errno = EPROTO;
+ return -1;
+ }
+
+ const unsigned char *ptr =
+ static_cast (msg_->data ());
+ size_t bytes_left = msg_->size ();
+
+ if (bytes_left < 8 || memcmp (ptr, "READY ", 8)) {
+ errno = EPROTO;
+ return -1;
+ }
+
+ ptr += 8;
+ bytes_left -= 8;
+
+ // Parse the property list
+ while (bytes_left > 1) {
+ const size_t name_length = static_cast (*ptr);
+ ptr += 1;
+ bytes_left -= 1;
+
+ if (bytes_left < name_length)
+ break;
+ const std::string name = std::string((const char *) ptr, name_length);
+ ptr += name_length;
+ bytes_left -= name_length;
+
+ if (bytes_left < 4)
+ break;
+ const size_t value_length = static_cast (get_uint32 (ptr));
+ ptr += 4;
+ bytes_left -= 4;
+
+ if (bytes_left < value_length)
+ break;
+ const unsigned char * const value = ptr;
+ ptr += value_length;
+ bytes_left -= value_length;
+
+ if (name == "Socket-Type") {
+ // TODO: Implement socket type checking
+ }
+ else
+ if (name == "Identity" && options.recv_identity)
+ set_peer_identity (value, value_length);
+ }
+
+ if (bytes_left > 0) {
+ errno = EPROTO;
+ return -1;
+ }
+
+ int rc = msg_->close ();
+ errno_assert (rc == 0);
+ rc = msg_->init ();
+ errno_assert (rc == 0);
+
+ ready_command_received = true;
+
+ return 0;
+}
+
+bool zmq::null_mechanism_t::is_handshake_complete () const
+{
+ return ready_command_received && ready_command_sent;
+}
diff --git a/src/null_mechanism.hpp b/src/null_mechanism.hpp
new file mode 100644
index 00000000..2c5d9c27
--- /dev/null
+++ b/src/null_mechanism.hpp
@@ -0,0 +1,51 @@
+/*
+ Copyright (c) 2007-2013 Contributors as noted in the AUTHORS file
+
+ This file is part of 0MQ.
+
+ 0MQ is free software; you can redistribute it and/or modify it under
+ the terms of the GNU Lesser General Public License as published by
+ the Free Software Foundation; either version 3 of the License, or
+ (at your option) any later version.
+
+ 0MQ is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ GNU Lesser General Public License for more details.
+
+ You should have received a copy of the GNU Lesser General Public License
+ along with this program. If not, see .
+*/
+
+#ifndef __ZMQ_NULL_MECHANISM_HPP_INCLUDED__
+#define __ZMQ_NULL_MECHANISM_HPP_INCLUDED__
+
+#include "mechanism.hpp"
+#include "options.hpp"
+
+namespace zmq
+{
+
+ class msg_t;
+
+ class null_mechanism_t : public mechanism_t
+ {
+ public:
+
+ null_mechanism_t (const options_t &options_);
+ virtual ~null_mechanism_t ();
+
+ // mechanism implementation
+ virtual int next_handshake_message (msg_t *msg_);
+ virtual int process_handshake_message (msg_t *msg_);
+ virtual bool is_handshake_complete () const;
+
+ private:
+
+ bool ready_command_sent;
+ bool ready_command_received;
+ };
+
+}
+
+#endif
diff --git a/src/plain_mechanism.cpp b/src/plain_mechanism.cpp
new file mode 100644
index 00000000..d235dfa9
--- /dev/null
+++ b/src/plain_mechanism.cpp
@@ -0,0 +1,359 @@
+/*
+ Copyright (c) 2007-2013 Contributors as noted in the AUTHORS file
+
+ This file is part of 0MQ.
+
+ 0MQ is free software; you can redistribute it and/or modify it under
+ the terms of the GNU Lesser General Public License as published by
+ the Free Software Foundation; either version 3 of the License, or
+ (at your option) any later version.
+
+ 0MQ is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ GNU Lesser General Public License for more details.
+
+ You should have received a copy of the GNU Lesser General Public License
+ along with this program. If not, see .
+*/
+
+#include "platform.hpp"
+#ifdef ZMQ_HAVE_WINDOWS
+#include "windows.hpp"
+#endif
+
+#include
+#include
+
+#include "msg.hpp"
+#include "err.hpp"
+#include "plain_mechanism.hpp"
+#include "wire.hpp"
+
+zmq::plain_mechanism_t::plain_mechanism_t (const options_t &options_,
+ bool as_server_) :
+ mechanism_t (options_),
+ state (as_server_? waiting_for_hello: sending_hello)
+{
+}
+
+zmq::plain_mechanism_t::~plain_mechanism_t ()
+{
+}
+
+int zmq::plain_mechanism_t::next_handshake_message (msg_t *msg_)
+{
+ int rc = 0;
+
+ switch (state) {
+ case sending_hello:
+ rc = hello_command (msg_);
+ if (rc == 0)
+ state = waiting_for_welcome;
+ break;
+ case sending_welcome:
+ rc = welcome_command (msg_);
+ if (rc == 0)
+ state = waiting_for_initiate;
+ break;
+ case sending_initiate:
+ rc = initiate_command (msg_);
+ if (rc == 0)
+ state = waiting_for_ready;
+ break;
+ case sending_ready:
+ rc = ready_command (msg_);
+ if (rc == 0)
+ state = ready;
+ break;
+ default:
+ errno = EAGAIN;
+ rc = -1;
+ }
+
+ return rc;
+}
+
+int zmq::plain_mechanism_t::process_handshake_message (msg_t *msg_)
+{
+ int rc = 0;
+
+ switch (state) {
+ case waiting_for_hello:
+ rc = process_hello_command (msg_);
+ if (rc == 0)
+ state = sending_welcome;
+ break;
+ case waiting_for_welcome:
+ rc = process_welcome_command (msg_);
+ if (rc == 0)
+ state = sending_initiate;
+ break;
+ case waiting_for_initiate:
+ rc = process_initiate_command (msg_);
+ if (rc == 0)
+ state = sending_ready;
+ break;
+ case waiting_for_ready:
+ rc = process_ready_command (msg_);
+ if (rc == 0)
+ state = ready;
+ break;
+ default:
+ errno = EAGAIN;
+ rc = -1;
+ }
+
+ if (rc == 0) {
+ rc = msg_->close ();
+ errno_assert (rc == 0);
+ rc = msg_->init ();
+ errno_assert (rc == 0);
+ }
+
+ return 0;
+}
+
+bool zmq::plain_mechanism_t::is_handshake_complete () const
+{
+ return state == ready;
+}
+
+int zmq::plain_mechanism_t::hello_command (msg_t *msg_) const
+{
+ // TODO: fetch these from options
+ const std::string username = "username";
+ const std::string password = "password";
+
+ zmq_assert (username.length () < 256);
+ zmq_assert (password.length () < 256);
+
+ const size_t command_size = 8 + 1 + username.length () +
+ 1 + password.length ();
+
+ const int rc = msg_->init_size (command_size);
+ errno_assert (rc == 0);
+
+ unsigned char *ptr = static_cast (msg_->data ());
+
+ memcpy (ptr, "HELLO ", 8);
+ ptr += 8;
+ *ptr++ = static_cast (username.length ());
+ memcpy (ptr, username.c_str (), username.length ());
+ ptr += username.length ();
+ *ptr++ = static_cast (password.length ());
+ memcpy (ptr, password.c_str (), password.length ());
+ ptr += password.length ();
+
+ // TODO: check username and password
+
+ return 0;
+}
+
+int zmq::plain_mechanism_t::process_hello_command (msg_t *msg_)
+{
+ const unsigned char *ptr = static_cast (msg_->data ());
+ size_t bytes_left = msg_->size ();
+
+ if (bytes_left < 8 || memcmp (ptr, "HELLO ", 8)) {
+ errno = EPROTO;
+ return -1;
+ }
+
+ ptr += 8;
+ bytes_left -= 8;
+
+ if (bytes_left < 1) {
+ errno = EPROTO;
+ return -1;
+ }
+
+ size_t username_length = static_cast (*ptr++);
+ bytes_left -= 1;
+
+ if (bytes_left < username_length) {
+ errno = EPROTO;
+ return -1;
+ }
+
+ const std::string username = std::string ((char *) ptr, username_length);
+ ptr += username_length;
+ bytes_left -= username_length;
+
+ if (bytes_left < 1) {
+ errno = EPROTO;
+ return -1;
+ }
+
+ size_t password_length = static_cast (*ptr++);
+ bytes_left -= 1;
+
+ if (bytes_left < password_length) {
+ errno = EPROTO;
+ return -1;
+ }
+
+ const std::string password = std::string ((char *) ptr, password_length);
+ ptr += password_length;
+ bytes_left -= password_length;
+
+ if (bytes_left > 0) {
+ errno = EPROTO;
+ return -1;
+ }
+
+ return 0;
+}
+
+int zmq::plain_mechanism_t::welcome_command (msg_t *msg_) const
+{
+ const int rc = msg_->init_size (8);
+ errno_assert (rc == 0);
+ memcpy (msg_->data (), "WELCOME ", 8);
+ return 0;
+}
+
+int zmq::plain_mechanism_t::process_welcome_command (msg_t *msg_)
+{
+ const unsigned char *ptr = static_cast (msg_->data ());
+ size_t bytes_left = msg_->size ();
+
+ if (bytes_left != 8 || memcmp (ptr, "WELCOME ", 8)) {
+ errno = EPROTO;
+ return -1;
+ }
+
+ return 0;
+}
+
+int zmq::plain_mechanism_t::initiate_command (msg_t *msg_) const
+{
+ unsigned char * const command_buffer = (unsigned char *) malloc (512);
+ alloc_assert (command_buffer);
+
+ unsigned char *ptr = command_buffer;
+
+ // Add mechanism string
+ memcpy (ptr, "INITIATE", 8);
+ ptr += 8;
+
+ // Add socket type property
+ const char *socket_type = socket_type_string (options.type);
+ ptr += add_property (ptr, "Socket-Type", socket_type, strlen (socket_type));
+
+ // Add identity property
+ if (options.type == ZMQ_REQ
+ || options.type == ZMQ_DEALER
+ || options.type == ZMQ_ROUTER) {
+ ptr += add_property (ptr, "Identity",
+ options.identity, options.identity_size);
+ }
+
+ const size_t command_size = ptr - command_buffer;
+ const int rc = msg_->init_size (command_size);
+ errno_assert (rc == 0);
+ memcpy (msg_->data (), command_buffer, command_size);
+ free (command_buffer);
+
+ return 0;
+}
+
+int zmq::plain_mechanism_t::process_initiate_command (msg_t *msg_)
+{
+ const unsigned char *ptr = static_cast (msg_->data ());
+ size_t bytes_left = msg_->size ();
+
+ if (bytes_left < 8 || memcmp (ptr, "INITIATE", 8)) {
+ errno = EPROTO;
+ return -1;
+ }
+
+ return parse_property_list (ptr + 8, bytes_left - 8);
+}
+
+int zmq::plain_mechanism_t::ready_command (msg_t *msg_) const
+{
+ unsigned char * const command_buffer = (unsigned char *) malloc (512);
+ alloc_assert (command_buffer);
+
+ unsigned char *ptr = command_buffer;
+
+ // Add mechanism string
+ memcpy (ptr, "READY ", 8);
+ ptr += 8;
+
+ // Add socket type property
+ const char *socket_type = socket_type_string (options.type);
+ ptr += add_property (ptr, "Socket-Type", socket_type, strlen (socket_type));
+
+ // Add identity property
+ if (options.type == ZMQ_REQ
+ || options.type == ZMQ_DEALER
+ || options.type == ZMQ_ROUTER) {
+ ptr += add_property (ptr, "Identity",
+ options.identity, options.identity_size);
+ }
+
+ const size_t command_size = ptr - command_buffer;
+ const int rc = msg_->init_size (command_size);
+ errno_assert (rc == 0);
+ memcpy (msg_->data (), command_buffer, command_size);
+ free (command_buffer);
+
+ return 0;
+}
+
+int zmq::plain_mechanism_t::process_ready_command (msg_t *msg_)
+{
+ const unsigned char *ptr = static_cast (msg_->data ());
+ size_t bytes_left = msg_->size ();
+
+ if (bytes_left < 8 || memcmp (ptr, "READY ", 8)) {
+ errno = EPROTO;
+ return -1;
+ }
+
+ return parse_property_list (ptr + 8, bytes_left - 8);
+}
+
+int zmq::plain_mechanism_t::parse_property_list (const unsigned char *ptr,
+ size_t bytes_left)
+{
+ while (bytes_left > 1) {
+ const size_t name_length = static_cast (*ptr);
+ ptr += 1;
+ bytes_left -= 1;
+
+ if (bytes_left < name_length)
+ break;
+ const std::string name = std::string((const char *) ptr, name_length);
+ ptr += name_length;
+ bytes_left -= name_length;
+
+ if (bytes_left < 4)
+ break;
+ const size_t value_length = static_cast (get_uint32 (ptr));
+ ptr += 4;
+ bytes_left -= 4;
+
+ if (bytes_left < value_length)
+ break;
+ const unsigned char * const value = ptr;
+ ptr += value_length;
+ bytes_left -= value_length;
+
+ if (name == "Socket-Type") {
+ // TODO: Implement socket type checking
+ }
+ else
+ if (name == "Identity" && options.recv_identity)
+ set_peer_identity (value, value_length);
+ }
+
+ if (bytes_left > 0) {
+ errno = EPROTO;
+ return -1;
+ }
+
+ return 0;
+}
diff --git a/src/plain_mechanism.hpp b/src/plain_mechanism.hpp
new file mode 100644
index 00000000..a89d0a41
--- /dev/null
+++ b/src/plain_mechanism.hpp
@@ -0,0 +1,74 @@
+/*
+ Copyright (c) 2007-2013 Contributors as noted in the AUTHORS file
+
+ This file is part of 0MQ.
+
+ 0MQ is free software; you can redistribute it and/or modify it under
+ the terms of the GNU Lesser General Public License as published by
+ the Free Software Foundation; either version 3 of the License, or
+ (at your option) any later version.
+
+ 0MQ is distributed in the hope that it will be useful,
+ but WITHOUT ANY WARRANTY; without even the implied warranty of
+ MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+ GNU Lesser General Public License for more details.
+
+ You should have received a copy of the GNU Lesser General Public License
+ along with this program. If not, see .
+*/
+
+#ifndef __ZMQ_PLAIN_MECHANISM_HPP_INCLUDED__
+#define __ZMQ_PLAIN_MECHANISM_HPP_INCLUDED__
+
+#include "mechanism.hpp"
+#include "options.hpp"
+
+namespace zmq
+{
+
+ class msg_t;
+
+ class plain_mechanism_t : public mechanism_t
+ {
+ public:
+
+ plain_mechanism_t (const options_t &options_, bool as_server_);
+ virtual ~plain_mechanism_t ();
+
+ // mechanism implementation
+ virtual int next_handshake_message (msg_t *msg_);
+ virtual int process_handshake_message (msg_t *msg_);
+ virtual bool is_handshake_complete () const;
+
+ private:
+
+ enum state_t {
+ sending_hello,
+ waiting_for_hello,
+ sending_welcome,
+ waiting_for_welcome,
+ sending_initiate,
+ waiting_for_initiate,
+ sending_ready,
+ waiting_for_ready,
+ ready
+ };
+
+ state_t state;
+
+ int hello_command (msg_t *msg_) const;
+ int welcome_command (msg_t *msg_) const;
+ int initiate_command (msg_t *msg_) const;
+ int ready_command (msg_t *msg_) const;
+
+ int process_hello_command (msg_t *msg_);
+ int process_welcome_command (msg_t *msg);
+ int process_ready_command (msg_t *msg_);
+ int process_initiate_command (msg_t *msg_);
+
+ int parse_property_list (const unsigned char *ptr, size_t length);
+ };
+
+}
+
+#endif
diff --git a/src/stream_engine.cpp b/src/stream_engine.cpp
index b894694a..4c604db8 100644
--- a/src/stream_engine.cpp
+++ b/src/stream_engine.cpp
@@ -40,6 +40,8 @@
#include "v1_decoder.hpp"
#include "v2_encoder.hpp"
#include "v2_decoder.hpp"
+#include "null_mechanism.hpp"
+#include "plain_mechanism.hpp"
#include "raw_decoder.hpp"
#include "raw_encoder.hpp"
#include "config.hpp"
@@ -48,8 +50,10 @@
#include "likely.hpp"
#include "wire.hpp"
-zmq::stream_engine_t::stream_engine_t (fd_t fd_, const options_t &options_, const std::string &endpoint_) :
+zmq::stream_engine_t::stream_engine_t (fd_t fd_, const options_t &options_,
+ bool as_server_, const std::string &endpoint_) :
s (fd_),
+ as_server (as_server_),
inpos (NULL),
insize (0),
decoder (NULL),
@@ -69,8 +73,9 @@ zmq::stream_engine_t::stream_engine_t (fd_t fd_, const options_t &options_, cons
io_error (false),
congested (false),
subscription_required (false),
+ mechanism (NULL),
+ input_paused (false),
output_paused (false),
- ready_command_received (false),
socket (NULL)
{
int rc = tx_msg.init ();
@@ -519,13 +524,21 @@ bool zmq::stream_engine_t::handshake ()
alloc_assert (decoder);
if (memcmp (greeting_recv + 12, "NULL\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0", 20) == 0) {
- read_msg = &stream_engine_t::send_ready_command;
- write_msg = &stream_engine_t::receive_ready_command;
+ mechanism = new (std::nothrow) null_mechanism_t (options);
+ alloc_assert (mechanism);
+ }
+ else
+ if (memcmp (greeting_recv + 12, "PLAIN\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0", 20) == 0) {
+ mechanism = new (std::nothrow) plain_mechanism_t (options, as_server);
+ alloc_assert (mechanism);
}
else {
error ();
return false;
}
+
+ read_msg = &stream_engine_t::next_handshake_message;
+ write_msg = &stream_engine_t::process_handshake_message;
}
// Start polling for output if necessary.
@@ -571,133 +584,59 @@ int zmq::stream_engine_t::write_identity (msg_t *msg_)
return 0;
}
-int zmq::stream_engine_t::send_ready_command (msg_t *msg_)
+int zmq::stream_engine_t::next_handshake_message (msg_t *msg_)
{
- unsigned char * const command_buffer = (unsigned char *) malloc (512);
- alloc_assert (command_buffer);
-
- unsigned char *ptr = command_buffer;
-
- // Add mechanism string
- memcpy (ptr, "READY ", 8);
- ptr += 8;
-
- // Add socket type property
- const char *socket_type = socket_type_string (options.type);
- ptr += add_property (ptr, "Socket-Type", socket_type, strlen (socket_type));
-
- // Add identity property
- if (options.type == ZMQ_REQ
- || options.type == ZMQ_DEALER
- || options.type == ZMQ_ROUTER) {
- ptr += add_property (ptr, "Identity",
- options.identity, options.identity_size);
+ zmq_assert (mechanism != NULL);
+
+ const int rc = mechanism->next_handshake_message (msg_);
+ if (rc == 0) {
+ if (mechanism->is_handshake_complete ())
+ mechanism_ready ();
+ if (input_paused) {
+ activate_in ();
+ input_paused = false;
+ }
}
-
- const size_t command_size = ptr - command_buffer;
- const int rc = msg_->init_size (command_size);
- errno_assert (rc == 0);
- memcpy (msg_->data (), command_buffer, command_size);
- free (command_buffer);
-
- if (ready_command_received)
- read_msg = &stream_engine_t::pull_msg_from_session;
else
- read_msg = &stream_engine_t::wait;
+ if (rc == -1) {
+ zmq_assert (errno == EAGAIN);
+ output_paused = true;
+ }
- return 0;
+ return rc;
}
-int zmq::stream_engine_t::receive_ready_command (msg_t *msg_)
+int zmq::stream_engine_t::process_handshake_message (msg_t *msg_)
{
- const unsigned char * const command_buffer =
- static_cast (msg_->data ());
- const size_t command_size = msg_->size ();
-
- const unsigned char *ptr = command_buffer;
- size_t bytes_left = command_size;
-
- if (bytes_left < 8 || memcmp(ptr, "READY ", 8)) {
- errno = EPROTO;
- return -1;
- }
-
- ptr += 8;
- bytes_left -= 8;
-
- // Parse the property list
- while (bytes_left > 1) {
- const size_t name_length = static_cast (*ptr);
- ptr += 1;
- bytes_left -= 1;
-
- if (bytes_left < name_length)
- break;
- const std::string name = std::string((const char *) ptr, name_length);
- ptr += name_length;
- bytes_left -= name_length;
-
- if (bytes_left < 4)
- break;
- const size_t value_length = static_cast (get_uint32 (ptr));
- ptr += 4;
- bytes_left -= 4;
-
- if (bytes_left < value_length)
- break;
- const unsigned char * const value = ptr;
- ptr += value_length;
- bytes_left -= value_length;
-
- if (name == "Socket-Type") {
- // Implement socket type checking
- }
- else
- if (name == "Identity") {
- if (options.recv_identity) {
- msg_t identity;
- int rc = identity.init_size (value_length);
- errno_assert (rc == 0);
- memcpy (identity.data (), value, value_length);
- identity.set_flags (msg_t::identity);
- rc = session->push_msg (&identity);
- errno_assert (rc == 0);
- }
+ zmq_assert (mechanism != NULL);
+
+ const int rc = mechanism->process_handshake_message (msg_);
+ if (rc == 0) {
+ if (mechanism->is_handshake_complete ())
+ mechanism_ready ();
+ if (output_paused) {
+ activate_out ();
+ output_paused = false;
}
}
+ else
+ if (rc == -1 && errno == EAGAIN)
+ input_paused = true;
- if (bytes_left > 0) {
- errno = EPROTO;
- return -1;
- }
-
- int rc = msg_->close ();
- errno_assert (rc == 0);
- rc = msg_->init ();
- errno_assert (rc == 0);
-
- write_msg = &stream_engine_t::push_msg_to_session;
-
- ready_command_received = true;
- if (output_paused) {
- activate_out ();
- output_paused = false;
- }
-
- return 0;
+ return rc;
}
-int zmq::stream_engine_t::wait (msg_t *msg_)
+void zmq::stream_engine_t::mechanism_ready ()
{
- if (ready_command_received) {
- read_msg = &stream_engine_t::pull_msg_from_session;
- return pull_msg_from_session (msg_);
- }
- else {
- output_paused = true;
- errno = EAGAIN;
- return -1;
+ if (options.recv_identity) {
+ msg_t identity;
+ mechanism->peer_identity (&identity);
+ const int rc = session->push_msg (&identity);
+ errno_assert (rc == 0);
}
+
+ read_msg = &stream_engine_t::pull_msg_from_session;
+ write_msg = &stream_engine_t::push_msg_to_session;
}
int zmq::stream_engine_t::pull_msg_from_session (msg_t *msg_)
@@ -727,29 +666,6 @@ int zmq::stream_engine_t::write_subscription_msg (msg_t *msg_)
return push_msg_to_session (msg_);
}
-size_t zmq::stream_engine_t::add_property (unsigned char *ptr,
- const char *name, const void *value, size_t value_len)
-{
- const size_t name_len = strlen (name);
- zmq_assert (name_len <= 255);
- *ptr++ = static_cast (name_len);
- memcpy (ptr, name, name_len);
- ptr += name_len;
- zmq_assert (value_len <= (2^31) - 1);
- put_uint32 (ptr, static_cast (value_len));
- ptr += 4;
- memcpy (ptr, value, value_len);
-
- return 1 + name_len + 4 + value_len;
-}
-
-const char *zmq::stream_engine_t::socket_type_string (int socket_type) {
- const char *names [] = {"PAIR", "PUB", "SUB", "REQ", "REP", "DEALER",
- "ROUTER", "PULL", "PUSH", "XPUB", "XSUB"};
- zmq_assert (socket_type >= 0 && socket_type <= 10);
- return names [socket_type];
-}
-
void zmq::stream_engine_t::error ()
{
zmq_assert (session);
diff --git a/src/stream_engine.hpp b/src/stream_engine.hpp
index cb8c75e1..31acc83c 100644
--- a/src/stream_engine.hpp
+++ b/src/stream_engine.hpp
@@ -43,6 +43,7 @@ namespace zmq
class io_thread_t;
class msg_t;
class session_base_t;
+ class mechanism_t;
// This engine handles any socket with SOCK_STREAM semantics,
// e.g. TCP socket or an UNIX domain socket.
@@ -51,7 +52,8 @@ namespace zmq
{
public:
- stream_engine_t (fd_t fd_, const options_t &options_, const std::string &endpoint);
+ stream_engine_t (fd_t fd_, const options_t &options_,
+ bool as_server_, const std::string &endpoint);
~stream_engine_t ();
// i_engine interface implementation.
@@ -93,13 +95,13 @@ namespace zmq
int read_identity (msg_t *msg_);
int write_identity (msg_t *msg_);
- int send_ready_command (msg_t *msg);
- int receive_ready_command (msg_t *msg);
+ int next_handshake_message (msg_t *msg);
+ int process_handshake_message (msg_t *msg);
int pull_msg_from_session (msg_t *msg_);
int push_msg_to_session (msg_t *msg);
- int wait (msg_t *msg_);
+ void mechanism_ready ();
int write_subscription_msg (msg_t *msg_);
@@ -111,6 +113,9 @@ namespace zmq
// Underlying socket.
fd_t s;
+ // True iff this is server's engine.
+ bool as_server;
+
msg_t tx_msg;
handle_t handle;
@@ -172,8 +177,10 @@ namespace zmq
// Needed to support old peers.
bool subscription_required;
+ mechanism_t *mechanism;
+
+ bool input_paused;
bool output_paused;
- bool ready_command_received;
// Socket
zmq::socket_base_t *socket;
diff --git a/src/tcp_connecter.cpp b/src/tcp_connecter.cpp
index e1ddf617..b83854ce 100644
--- a/src/tcp_connecter.cpp
+++ b/src/tcp_connecter.cpp
@@ -125,7 +125,8 @@ void zmq::tcp_connecter_t::out_event ()
tune_tcp_keepalives (fd, options.tcp_keepalive, options.tcp_keepalive_cnt, options.tcp_keepalive_idle, options.tcp_keepalive_intvl);
// Create the engine object for this connection.
- stream_engine_t *engine = new (std::nothrow) stream_engine_t (fd, options, endpoint);
+ stream_engine_t *engine = new (std::nothrow)
+ stream_engine_t (fd, options, false, endpoint);
alloc_assert (engine);
// Attach the engine to the corresponding session object.
diff --git a/src/tcp_listener.cpp b/src/tcp_listener.cpp
index 7de38cb1..a291f53c 100644
--- a/src/tcp_listener.cpp
+++ b/src/tcp_listener.cpp
@@ -91,7 +91,8 @@ void zmq::tcp_listener_t::in_event ()
tune_tcp_keepalives (fd, options.tcp_keepalive, options.tcp_keepalive_cnt, options.tcp_keepalive_idle, options.tcp_keepalive_intvl);
// Create the engine object for this connection.
- stream_engine_t *engine = new (std::nothrow) stream_engine_t (fd, options, endpoint);
+ stream_engine_t *engine = new (std::nothrow)
+ stream_engine_t (fd, options, true, endpoint);
alloc_assert (engine);
// Choose I/O thread to run connecter in. Given that we are already