summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
-rw-r--r--src/command.cpp4
-rw-r--r--src/command.hpp2
-rw-r--r--src/object.cpp28
-rw-r--r--src/object.hpp4
-rw-r--r--src/options.cpp2
-rw-r--r--src/options.hpp5
-rw-r--r--src/pgm_socket.cpp7
-rw-r--r--src/session.cpp2
-rw-r--r--src/socket_base.cpp5
-rw-r--r--src/socket_base.hpp3
10 files changed, 46 insertions, 16 deletions
diff --git a/src/command.cpp b/src/command.cpp
index 7564fe2..8bf7ea2 100644
--- a/src/command.cpp
+++ b/src/command.cpp
@@ -28,6 +28,10 @@ void zmq::deallocate_command (command_t *cmd_)
if (cmd_->args.attach.peer_identity)
free (cmd_->args.attach.peer_identity);
break;
+ case command_t::bind:
+ if (cmd_->args.bind.peer_identity)
+ free (cmd_->args.bind.peer_identity);
+ break;
default:
/* noop */;
}
diff --git a/src/command.hpp b/src/command.hpp
index 6187b72..150cad1 100644
--- a/src/command.hpp
+++ b/src/command.hpp
@@ -75,6 +75,8 @@ namespace zmq
struct {
class reader_t *in_pipe;
class writer_t *out_pipe;
+ unsigned char peer_identity_size;
+ unsigned char *peer_identity;
} bind;
// Sent by pipe writer to inform dormant pipe reader that there
diff --git a/src/object.cpp b/src/object.cpp
index eaa67c9..356fcd1 100644
--- a/src/object.cpp
+++ b/src/object.cpp
@@ -89,7 +89,9 @@ void zmq::object_t::process_command (command_t &cmd_)
break;
case command_t::bind:
- process_bind (cmd_.args.bind.in_pipe, cmd_.args.bind.out_pipe);
+ process_bind (cmd_.args.bind.in_pipe, cmd_.args.bind.out_pipe,
+ blob_t (cmd_.args.bind.peer_identity,
+ cmd_.args.bind.peer_identity_size));
process_seqnum ();
break;
@@ -198,7 +200,9 @@ void zmq::object_t::send_attach (session_t *destination_, i_engine *engine_,
cmd.args.attach.peer_identity = NULL;
}
else {
- cmd.args.attach.peer_identity_size = peer_identity_.size ();
+ zmq_assert (peer_identity_.size () <= 0xff);
+ cmd.args.attach.peer_identity_size =
+ (unsigned char) peer_identity_.size ();
cmd.args.attach.peer_identity =
(unsigned char*) malloc (peer_identity_.size ());
zmq_assert (cmd.args.attach.peer_identity_size);
@@ -209,7 +213,8 @@ void zmq::object_t::send_attach (session_t *destination_, i_engine *engine_,
}
void zmq::object_t::send_bind (socket_base_t *destination_,
- reader_t *in_pipe_, writer_t *out_pipe_, bool inc_seqnum_)
+ reader_t *in_pipe_, writer_t *out_pipe_, const blob_t &peer_identity_,
+ bool inc_seqnum_)
{
if (inc_seqnum_)
destination_->inc_seqnum ();
@@ -219,6 +224,20 @@ void zmq::object_t::send_bind (socket_base_t *destination_,
cmd.type = command_t::bind;
cmd.args.bind.in_pipe = in_pipe_;
cmd.args.bind.out_pipe = out_pipe_;
+ if (peer_identity_.empty ()) {
+ cmd.args.bind.peer_identity_size = 0;
+ cmd.args.bind.peer_identity = NULL;
+ }
+ else {
+ zmq_assert (peer_identity_.size () <= 0xff);
+ cmd.args.bind.peer_identity_size =
+ (unsigned char) peer_identity_.size ();
+ cmd.args.bind.peer_identity =
+ (unsigned char*) malloc (peer_identity_.size ());
+ zmq_assert (cmd.args.bind.peer_identity_size);
+ memcpy (cmd.args.bind.peer_identity, peer_identity_.data (),
+ peer_identity_.size ());
+ }
send_command (cmd);
}
@@ -293,7 +312,8 @@ void zmq::object_t::process_attach (i_engine *engine_,
zmq_assert (false);
}
-void zmq::object_t::process_bind (reader_t *in_pipe_, writer_t *out_pipe_)
+void zmq::object_t::process_bind (reader_t *in_pipe_, writer_t *out_pipe_,
+ const blob_t &peer_identity_)
{
zmq_assert (false);
}
diff --git a/src/object.hpp b/src/object.hpp
index d492695..1544109 100644
--- a/src/object.hpp
+++ b/src/object.hpp
@@ -69,7 +69,7 @@ namespace zmq
bool inc_seqnum_ = true);
void send_bind (class socket_base_t *destination_,
class reader_t *in_pipe_, class writer_t *out_pipe_,
- bool inc_seqnum_ = true);
+ const blob_t &peer_identity_, bool inc_seqnum_ = true);
void send_revive (class object_t *destination_);
void send_pipe_term (class writer_t *destination_);
void send_pipe_term_ack (class reader_t *destination_);
@@ -86,7 +86,7 @@ namespace zmq
virtual void process_attach (struct i_engine *engine_,
const blob_t &peer_identity_);
virtual void process_bind (class reader_t *in_pipe_,
- class writer_t *out_pipe_);
+ class writer_t *out_pipe_, const blob_t &peer_identity_);
virtual void process_revive ();
virtual void process_pipe_term ();
virtual void process_pipe_term_ack ();
diff --git a/src/options.cpp b/src/options.cpp
index cdfccc6..a70b9a3 100644
--- a/src/options.cpp
+++ b/src/options.cpp
@@ -76,7 +76,7 @@ int zmq::options_t::setsockopt (int option_, const void *optval_,
return 0;
case ZMQ_IDENTITY:
- identity.assign ((const char*) optval_, optvallen_);
+ identity.assign ((const unsigned char*) optval_, optvallen_);
return 0;
case ZMQ_RATE:
diff --git a/src/options.hpp b/src/options.hpp
index f9ff6e4..541e6e8 100644
--- a/src/options.hpp
+++ b/src/options.hpp
@@ -20,10 +20,9 @@
#ifndef __ZMQ_OPTIONS_HPP_INCLUDED__
#define __ZMQ_OPTIONS_HPP_INCLUDED__
-#include <string>
-
#include "stddef.h"
#include "stdint.hpp"
+#include "blob.hpp"
namespace zmq
{
@@ -38,7 +37,7 @@ namespace zmq
int64_t lwm;
int64_t swap;
uint64_t affinity;
- std::string identity;
+ blob_t identity;
// Maximum tranfer rate [kb/s]. Default 100kb/s.
uint32_t rate;
diff --git a/src/pgm_socket.cpp b/src/pgm_socket.cpp
index 1eeb34f..462a3a9 100644
--- a/src/pgm_socket.cpp
+++ b/src/pgm_socket.cpp
@@ -89,8 +89,11 @@ int zmq::pgm_socket_t::init (bool udp_encapsulation_, const char *network_)
if (options.identity.size () > 0) {
- // Create gsi from identity string.
- gsi_base = options.identity;
+ // Create gsi from identity.
+ // TODO: We assume that identity is standard C string here.
+ // What if it contains binary zeroes?
+ gsi_base.assign ((const char*) options.identity.data (),
+ options.identity.size ());
} else {
// Generate random gsi.
diff --git a/src/session.cpp b/src/session.cpp
index b2393d8..909501a 100644
--- a/src/session.cpp
+++ b/src/session.cpp
@@ -245,7 +245,7 @@ void zmq::session_t::process_attach (i_engine *engine_,
}
send_bind (owner, outbound ? &outbound->reader : NULL,
- inbound ? &inbound->writer : NULL);
+ inbound ? &inbound->writer : NULL, peer_identity);
}
// Plug in the engine.
diff --git a/src/socket_base.cpp b/src/socket_base.cpp
index 7d90236..1d4eae6 100644
--- a/src/socket_base.cpp
+++ b/src/socket_base.cpp
@@ -171,7 +171,7 @@ int zmq::socket_base_t::connect (const char *addr_)
// was incremented in find_endpoint function. The callee is notified
// about the fact via the last parameter.
send_bind (peer, out_pipe ? &out_pipe->reader : NULL,
- in_pipe ? &in_pipe->writer : NULL, false);
+ in_pipe ? &in_pipe->writer : NULL, options.identity, false);
return 0;
}
@@ -564,7 +564,8 @@ void zmq::socket_base_t::process_own (owned_t *object_)
io_objects.insert (object_);
}
-void zmq::socket_base_t::process_bind (reader_t *in_pipe_, writer_t *out_pipe_)
+void zmq::socket_base_t::process_bind (reader_t *in_pipe_, writer_t *out_pipe_,
+ const blob_t &peer_identity_)
{
attach_pipes (in_pipe_, out_pipe_);
}
diff --git a/src/socket_base.hpp b/src/socket_base.hpp
index 39f09de..a1702a7 100644
--- a/src/socket_base.hpp
+++ b/src/socket_base.hpp
@@ -122,7 +122,8 @@ namespace zmq
// Handlers for incoming commands.
void process_own (class owned_t *object_);
- void process_bind (class reader_t *in_pipe_, class writer_t *out_pipe_);
+ void process_bind (class reader_t *in_pipe_, class writer_t *out_pipe_,
+ const blob_t &peer_identity_);
void process_term_req (class owned_t *object_);
void process_term_ack ();
void process_seqnum ();