3party/libzmq
0
0
Fork 0
mirror of https://github.com/zeromq/libzmq.git synced 2025-01-15 10:18:01 +08:00
Code Issues Projects Releases Wiki Activity

Backport resolve #1357 Support limited metadata for STREAM sockets

Browse Source
This commit is contained in:
Thomas Rodgers 2015-02-18 13:09:11 -06:00
parent 9fe4fb3929
commit d172875f52
4 changed files with 41 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/stream.cpp
View File

@ -211,6 +211,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);

19
src/stream_engine.cpp
View File

@ -192,13 +192,22 @@ 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;
if (!peer_address.empty()) {
// Compile metadata.
typedef metadata_t::dict_t properties_t;
properties_t properties;
properties.insert(std::make_pair("Peer-Address", peer_address));
zmq_assert (metadata == NULL);
metadata = new (std::nothrow) metadata_t (properties);
}
// 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 ();
}
@ -822,6 +831,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);

2
src/stream_engine.hpp
View File

@ -96,6 +96,8 @@ namespace zmq
int next_handshake_command (msg_t *msg);
int process_handshake_command (msg_t *msg);
int push_raw_msg_to_session (msg_t *msg);
int pull_msg_from_session (msg_t *msg_);
int push_msg_to_session (msg_t *msg);

6
tests/test_stream.cpp
View File

@ -77,6 +77,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);
@ -88,6 +91,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);