mirror of
https://github.com/zeromq/libzmq.git
synced 2025-01-03 03:45:43 +08:00
Merge pull request #409 from hurtonm/master
Exchange greeting messages for all socket types
This commit is contained in:
commit
12833804c9
@ -35,7 +35,6 @@ zmq::dealer_t::dealer_t (class ctx_t *parent_, uint32_t tid_, int sid_) :
|
||||
// be noone to receive the replies anyway.
|
||||
// options.delay_on_close = false;
|
||||
|
||||
options.send_identity = true;
|
||||
options.recv_identity = true;
|
||||
|
||||
prefetched_msg.init ();
|
||||
|
@ -48,7 +48,6 @@ zmq::options_t::options_t () :
|
||||
delay_on_close (true),
|
||||
delay_on_disconnect (true),
|
||||
filter (false),
|
||||
send_identity (false),
|
||||
recv_identity (false),
|
||||
tcp_keepalive (-1),
|
||||
tcp_keepalive_cnt (-1),
|
||||
|
@ -112,10 +112,7 @@ namespace zmq
|
||||
// If 1, (X)SUB socket should filter the messages. If 0, it should not.
|
||||
bool filter;
|
||||
|
||||
// Sends identity to all new connections.
|
||||
bool send_identity;
|
||||
|
||||
// Receivers identity from all new connections.
|
||||
// If true, the identity message is forwarded to the socket.
|
||||
bool recv_identity;
|
||||
|
||||
// TCP keep-alive settings.
|
||||
|
@ -45,7 +45,6 @@ zmq::router_t::router_t (class ctx_t *parent_, uint32_t tid_, int sid_) :
|
||||
// all the outstanding requests from that peer.
|
||||
// options.delay_on_disconnect = false;
|
||||
|
||||
options.send_identity = true;
|
||||
options.recv_identity = true;
|
||||
|
||||
prefetched_id.init ();
|
||||
|
@ -117,8 +117,8 @@ zmq::session_base_t::session_base_t (class io_thread_t *io_thread_,
|
||||
socket (socket_),
|
||||
io_thread (io_thread_),
|
||||
has_linger_timer (false),
|
||||
send_identity (options_.send_identity),
|
||||
recv_identity (options_.recv_identity),
|
||||
identity_sent (false),
|
||||
identity_received (false),
|
||||
addr (addr_)
|
||||
{
|
||||
}
|
||||
@ -152,13 +152,13 @@ void zmq::session_base_t::attach_pipe (pipe_t *pipe_)
|
||||
|
||||
int zmq::session_base_t::read (msg_t *msg_)
|
||||
{
|
||||
// First message to send is identity (if required).
|
||||
if (send_identity) {
|
||||
// First message to send is identity
|
||||
if (!identity_sent) {
|
||||
zmq_assert (!(msg_->flags () & msg_t::more));
|
||||
int rc = msg_->init_size (options.identity_size);
|
||||
errno_assert (rc == 0);
|
||||
memcpy (msg_->data (), options.identity, options.identity_size);
|
||||
send_identity = false;
|
||||
identity_sent = true;
|
||||
incomplete_in = false;
|
||||
return 0;
|
||||
}
|
||||
@ -174,10 +174,17 @@ int zmq::session_base_t::read (msg_t *msg_)
|
||||
|
||||
int zmq::session_base_t::write (msg_t *msg_)
|
||||
{
|
||||
// First message to receive is identity (if required).
|
||||
if (recv_identity) {
|
||||
// First message to receive is identity
|
||||
if (!identity_received) {
|
||||
msg_->set_flags (msg_t::identity);
|
||||
recv_identity = false;
|
||||
identity_received = true;
|
||||
if (!options.recv_identity) {
|
||||
int rc = msg_->close ();
|
||||
errno_assert (rc == 0);
|
||||
rc = msg_->init ();
|
||||
errno_assert (rc == 0);
|
||||
return 0;
|
||||
}
|
||||
}
|
||||
|
||||
if (pipe && pipe->write (msg_)) {
|
||||
@ -193,8 +200,8 @@ int zmq::session_base_t::write (msg_t *msg_)
|
||||
void zmq::session_base_t::reset ()
|
||||
{
|
||||
// Restore identity flags.
|
||||
send_identity = options.send_identity;
|
||||
recv_identity = options.recv_identity;
|
||||
identity_sent = false;
|
||||
identity_received = false;
|
||||
}
|
||||
|
||||
void zmq::session_base_t::flush ()
|
||||
|
@ -130,9 +130,9 @@ namespace zmq
|
||||
// True is linger timer is running.
|
||||
bool has_linger_timer;
|
||||
|
||||
// If true, identity is to be sent/recvd from the network.
|
||||
bool send_identity;
|
||||
bool recv_identity;
|
||||
// If true, identity has been sent/received from the network.
|
||||
bool identity_sent;
|
||||
bool identity_received;
|
||||
|
||||
// Protocol and address to use when connecting.
|
||||
const address_t *addr;
|
||||
|
@ -448,7 +448,7 @@ int zmq::socket_base_t::connect (const char *addr_)
|
||||
attach_pipe (pipes [0]);
|
||||
|
||||
// If required, send the identity of the local socket to the peer.
|
||||
if (options.send_identity) {
|
||||
if (peer.options.recv_identity) {
|
||||
msg_t id;
|
||||
rc = id.init_size (options.identity_size);
|
||||
errno_assert (rc == 0);
|
||||
@ -460,7 +460,7 @@ int zmq::socket_base_t::connect (const char *addr_)
|
||||
}
|
||||
|
||||
// If required, send the identity of the peer to the local socket.
|
||||
if (peer.options.send_identity) {
|
||||
if (options.recv_identity) {
|
||||
msg_t id;
|
||||
rc = id.init_size (peer.options.identity_size);
|
||||
errno_assert (rc == 0);
|
||||
|
Loading…
x
Reference in New Issue
Block a user