mirror of
https://github.com/zeromq/libzmq.git
synced 2024-12-12 10:33:52 +01:00
Add metadata to received messages
This commit is contained in:
parent
62bb403eea
commit
b9c2748146
@ -46,6 +46,7 @@ libzmq_la_SOURCES = \
|
||||
likely.hpp \
|
||||
mailbox.hpp \
|
||||
mechanism.hpp \
|
||||
metadata.hpp \
|
||||
msg.hpp \
|
||||
mtrie.hpp \
|
||||
mutex.hpp \
|
||||
@ -118,6 +119,7 @@ libzmq_la_SOURCES = \
|
||||
lb.cpp \
|
||||
mailbox.cpp \
|
||||
mechanism.cpp \
|
||||
metadata.cpp \
|
||||
msg.cpp \
|
||||
mtrie.cpp \
|
||||
norm_engine.cpp \
|
||||
|
@ -663,7 +663,7 @@ int zmq::curve_server_t::receive_and_process_zap_reply ()
|
||||
|
||||
// Process metadata frame
|
||||
rc = parse_metadata (static_cast <const unsigned char*> (msg [6].data ()),
|
||||
msg [6].size ());
|
||||
msg [6].size (), true);
|
||||
|
||||
error:
|
||||
for (int i = 0; i < 7; i++) {
|
||||
|
@ -268,7 +268,7 @@ int zmq::gssapi_server_t::receive_and_process_zap_reply ()
|
||||
|
||||
// Process metadata frame
|
||||
rc = parse_metadata (static_cast <const unsigned char*> (msg [6].data ()),
|
||||
msg [6].size ());
|
||||
msg [6].size (), true);
|
||||
|
||||
error:
|
||||
for (int i = 0; i < 7; i++) {
|
||||
|
@ -20,6 +20,8 @@
|
||||
#ifndef __ZMQ_I_PROPERTIES_HPP_INCLUDED__
|
||||
#define __ZMQ_I_PROPERTIES_HPP_INCLUDED__
|
||||
|
||||
#include <string>
|
||||
|
||||
namespace zmq
|
||||
{
|
||||
// Interface for accessing message properties.
|
||||
@ -32,7 +34,7 @@ namespace zmq
|
||||
|
||||
// Returns pointer to property value or NULL if
|
||||
// property not found.
|
||||
virtual const char *get (const char *property) const = 0;
|
||||
virtual const char *get (const std::string &property) const = 0;
|
||||
|
||||
virtual void add_ref () = 0;
|
||||
|
||||
|
@ -18,6 +18,7 @@
|
||||
*/
|
||||
|
||||
#include <string.h>
|
||||
#include <map>
|
||||
|
||||
#include "mechanism.hpp"
|
||||
#include "options.hpp"
|
||||
@ -26,12 +27,16 @@
|
||||
#include "wire.hpp"
|
||||
|
||||
zmq::mechanism_t::mechanism_t (const options_t &options_) :
|
||||
metadata (NULL),
|
||||
options (options_)
|
||||
{
|
||||
}
|
||||
|
||||
zmq::mechanism_t::~mechanism_t ()
|
||||
{
|
||||
if (metadata != NULL)
|
||||
if (metadata->drop_ref ())
|
||||
delete metadata;
|
||||
}
|
||||
|
||||
void zmq::mechanism_t::set_peer_identity (const void *id_ptr, size_t id_size)
|
||||
@ -83,8 +88,9 @@ size_t zmq::mechanism_t::add_property (unsigned char *ptr, const char *name,
|
||||
}
|
||||
|
||||
int zmq::mechanism_t::parse_metadata (const unsigned char *ptr_,
|
||||
size_t length_)
|
||||
size_t length_, bool zap_flag)
|
||||
{
|
||||
std::map <const std::string, const std::string> dict;
|
||||
size_t bytes_left = length_;
|
||||
|
||||
while (bytes_left > 1) {
|
||||
@ -125,11 +131,19 @@ int zmq::mechanism_t::parse_metadata (const unsigned char *ptr_,
|
||||
if (rc == -1)
|
||||
return -1;
|
||||
}
|
||||
|
||||
dict.insert (
|
||||
std::map <const std::string, const std::string>::value_type (
|
||||
name, std::string ((char *) value, value_length)));
|
||||
}
|
||||
if (bytes_left > 0) {
|
||||
errno = EPROTO;
|
||||
return -1;
|
||||
}
|
||||
if (zap_flag) {
|
||||
assert (metadata == NULL);
|
||||
metadata = new (std::nothrow) metadata_t (dict);
|
||||
}
|
||||
return 0;
|
||||
}
|
||||
|
||||
|
@ -23,6 +23,7 @@
|
||||
#include "stdint.hpp"
|
||||
#include "options.hpp"
|
||||
#include "blob.hpp"
|
||||
#include "metadata.hpp"
|
||||
|
||||
namespace zmq
|
||||
{
|
||||
@ -64,6 +65,8 @@ namespace zmq
|
||||
|
||||
blob_t get_user_id () const;
|
||||
|
||||
metadata_t *get_metadata () { return metadata; }
|
||||
|
||||
protected:
|
||||
|
||||
// Only used to identify the socket for the Socket-Type
|
||||
@ -77,7 +80,8 @@ namespace zmq
|
||||
// Metadata consists of a list of properties consisting of
|
||||
// name and value as size-specified strings.
|
||||
// Returns 0 on success and -1 on error, in which case errno is set.
|
||||
int parse_metadata (const unsigned char *ptr_, size_t length);
|
||||
int parse_metadata (
|
||||
const unsigned char *ptr_, size_t length, bool zap_flag = false);
|
||||
|
||||
// This is called by parse_property method whenever it
|
||||
// parses a new property. The function should return 0
|
||||
@ -89,6 +93,10 @@ namespace zmq
|
||||
virtual int property (const std::string& name_,
|
||||
const void *value_, size_t length_);
|
||||
|
||||
// Metadada as returned by ZAP protocol.
|
||||
// NULL if no metadata received.
|
||||
metadata_t *metadata;
|
||||
|
||||
options_t options;
|
||||
|
||||
private:
|
||||
|
49
src/metadata.cpp
Normal file
49
src/metadata.cpp
Normal file
@ -0,0 +1,49 @@
|
||||
/*
|
||||
Copyright (c) 2007-2014 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 <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
#include "metadata.hpp"
|
||||
|
||||
zmq::metadata_t::metadata_t (dict_t &dict) :
|
||||
ref_cnt (1),
|
||||
dict (dict)
|
||||
{
|
||||
}
|
||||
|
||||
zmq::metadata_t::~metadata_t ()
|
||||
{
|
||||
}
|
||||
|
||||
const char *zmq::metadata_t::get (const std::string &property) const
|
||||
{
|
||||
dict_t::const_iterator it = dict.find (property);
|
||||
if (it == dict.end ())
|
||||
return NULL;
|
||||
else
|
||||
return it->second.c_str ();
|
||||
}
|
||||
|
||||
void zmq::metadata_t::add_ref ()
|
||||
{
|
||||
ref_cnt.add (1);
|
||||
}
|
||||
|
||||
bool zmq::metadata_t::drop_ref ()
|
||||
{
|
||||
return !ref_cnt.sub (1);
|
||||
}
|
59
src/metadata.hpp
Normal file
59
src/metadata.hpp
Normal file
@ -0,0 +1,59 @@
|
||||
/*
|
||||
Copyright (c) 2007-2014 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 <http://www.gnu.org/licenses/>.
|
||||
*/
|
||||
|
||||
#ifndef __ZMQ_METADATA_HPP_INCLUDED__
|
||||
#define __ZMQ_METADATA_HPP_INCLUDED__
|
||||
|
||||
#include <map>
|
||||
|
||||
#include "atomic_counter.hpp"
|
||||
#include "i_properties.hpp"
|
||||
|
||||
namespace zmq
|
||||
{
|
||||
class metadata_t : public i_properties
|
||||
{
|
||||
public:
|
||||
|
||||
metadata_t (std::map <const std::string, const std::string> &dict);
|
||||
virtual ~metadata_t ();
|
||||
|
||||
// Returns pointer to property value or NULL if
|
||||
// property is not found.
|
||||
virtual const char *get (const std::string &property) const;
|
||||
|
||||
virtual void add_ref ();
|
||||
|
||||
// Drop reference. Returns true iff the reference
|
||||
// counter drops to zero.
|
||||
virtual bool drop_ref ();
|
||||
|
||||
private:
|
||||
|
||||
// Reference counter.
|
||||
atomic_counter_t ref_cnt;
|
||||
|
||||
// Dictionary holding metadata.
|
||||
typedef std::map <const std::string, const std::string> dict_t;
|
||||
const dict_t dict;
|
||||
};
|
||||
|
||||
}
|
||||
|
||||
#endif
|
@ -285,7 +285,7 @@ int zmq::null_mechanism_t::receive_and_process_zap_reply ()
|
||||
|
||||
// Process metadata frame
|
||||
rc = parse_metadata (static_cast <const unsigned char*> (msg [6].data ()),
|
||||
msg [6].size ());
|
||||
msg [6].size (), true);
|
||||
|
||||
error:
|
||||
for (int i = 0; i < 7; i++) {
|
||||
|
@ -502,7 +502,7 @@ int zmq::plain_mechanism_t::receive_and_process_zap_reply ()
|
||||
|
||||
// Process metadata frame
|
||||
rc = parse_metadata (static_cast <const unsigned char*> (msg [6].data ()),
|
||||
msg [6].size ());
|
||||
msg [6].size (), true);
|
||||
|
||||
error:
|
||||
for (int i = 0; i < 7; i++) {
|
||||
|
@ -780,6 +780,9 @@ int zmq::stream_engine_t::decode_and_push (msg_t *msg_)
|
||||
|
||||
if (mechanism->decode (msg_) == -1)
|
||||
return -1;
|
||||
metadata_t *metadata = mechanism->get_metadata ();
|
||||
if (metadata)
|
||||
msg_->set_properties (metadata);
|
||||
if (session->push_msg (msg_) == -1) {
|
||||
if (errno == EAGAIN)
|
||||
write_msg = &stream_engine_t::push_one_then_decode_and_push;
|
||||
|
@ -649,7 +649,7 @@ const char *zmq_msg_gets (zmq_msg_t *msg_, const char *property_)
|
||||
{
|
||||
zmq::i_properties *properties = ((zmq::msg_t*) msg_)->properties ();
|
||||
if (properties)
|
||||
return properties->get (property_);
|
||||
return properties->get (std::string (property_));
|
||||
else
|
||||
return NULL;
|
||||
}
|
||||
|
Loading…
Reference in New Issue
Block a user