Core: Implemented IPC (Inter-process communication) between worldserver and bnetserver using ZeroMQ library.
* Implemented ToonReady and ToonLoggedOut battle.net packets
This commit is contained in:
1 parent
40cff79b25
commit
f773a9e053
72 files changed
+4922
-68
No files matched your search
@@ -36,6 +36,7 @@ endif()
|
||||
|
||||
if(SERVERS)
|
||||
add_subdirectory(gsoap)
|
||||
add_subdirectory(zmqpp)
|
||||
endif()
|
||||
|
||||
if(TOOLS)
|
||||
|
||||
@@ -39,3 +39,7 @@ recastnavigation (Recast is state of the art navigation mesh construction toolse
|
||||
StormLib (a pack of modules, written in C++, which are able to read and also to write files from/to the MPQ archives)
|
||||
http://www.zezula.net/en/mpq/stormlib.html
|
||||
Version: 8.04
|
||||
|
||||
zmqpp (C++ binding for 0mq/zmq is a 'high-level' library that hides most of the c-style interface core 0mq provides.)
|
||||
https://github.com/zeromq/zmqpp
|
||||
Version: 3.2.0 17e9f6afa98f56ecac1e3f3eecbfc112357a6732
|
||||
@@ -0,0 +1,31 @@
|
||||
# Copyright (C) 2008-2014 TrinityCore <http://www.trinitycore.org/>
|
||||
#
|
||||
# This file is free software; as a special exception the author gives
|
||||
# unlimited permission to copy and/or distribute it, with or without
|
||||
# modifications, as long as this notice is preserved.
|
||||
#
|
||||
# This program is distributed in the hope that it will be useful, but
|
||||
# WITHOUT ANY WARRANTY, to the extent permitted by law; without even the
|
||||
# implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
|
||||
|
||||
file(GLOB_RECURSE sources_zmqpp zmqpp/*.cpp zmqpp/*.hpp zmqpp/*.h)
|
||||
|
||||
set(zmqpp_STAT_SRCS
|
||||
${sources_zmqpp}
|
||||
)
|
||||
|
||||
include_directories(${ZMQ_INCLUDE_DIR})
|
||||
|
||||
add_library(zmqpp STATIC
|
||||
${zmqpp_STAT_SRCS}
|
||||
)
|
||||
|
||||
if (WIN32)
|
||||
add_definitions(-DBUILD_VERSION=\\"3.2.0\\")
|
||||
else()
|
||||
add_definitions(-DBUILD_VERSION='"3.2.0"')
|
||||
endif()
|
||||
|
||||
add_definitions(-DBUILD_VERSION_MAJOR=3)
|
||||
add_definitions(-DBUILD_VERSION_MINOR=2)
|
||||
add_definitions(-DBUILD_VERSION_REVISION=0)
|
||||
@@ -0,0 +1,97 @@
|
||||
/**
|
||||
* \file
|
||||
*
|
||||
* \date 10 Sep 2011
|
||||
* \author ron
|
||||
* \author Ben Gray (\@benjamg)
|
||||
*
|
||||
* A fair number of C++0x (or more accurately C++11) features are used in this
|
||||
* library and as this project is used where I work on older compilers this
|
||||
* file was created to help.
|
||||
*
|
||||
* C++ features and their workaround where not supported:
|
||||
* \li lambda functions - disabled, these are only used in the test anyway.
|
||||
* \li typesafe enums - replaced with enum where comparisons needed.
|
||||
* \li nullptr - defined to null.
|
||||
*
|
||||
* As of the port to version 3.1 (libzmqpp version 1.1.0) this file will also
|
||||
* be used to maintain compatablity with multiple versions of 0mq
|
||||
*/
|
||||
|
||||
#ifndef ZMQPP_COMPATIBILITY_HPP_
|
||||
#define ZMQPP_COMPATIBILITY_HPP_
|
||||
|
||||
#include <zmq.h>
|
||||
#include <cstdint>
|
||||
|
||||
// Currently we require at least 0mq version 2.2.x
|
||||
#define ZMQPP_REQUIRED_ZMQ_MAJOR 2
|
||||
#define ZMQPP_REQUIRED_ZMQ_MINOR 2
|
||||
|
||||
#if (ZMQ_VERSION_MAJOR < ZMQPP_REQUIRED_ZMQ_MAJOR) || ((ZMQ_VERSION_MAJOR == ZMQPP_REQUIRED_ZMQ_MAJOR) && (ZMQ_VERSION_MINOR < ZMQPP_REQUIRED_ZMQ_MINOR))
|
||||
#error zmqpp requires a later version of 0mq
|
||||
#endif
|
||||
|
||||
// Experimental feature support
|
||||
#if (ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR == 0)
|
||||
#define ZMQ_EXPERIMENTAL_LABELS
|
||||
#endif
|
||||
|
||||
// Deal with older versions of gcc
|
||||
#if defined(__GNUC__) && !defined(__clang__)
|
||||
#if __GNUC__ == 4
|
||||
|
||||
// Deal with older gcc not supporting C++0x typesafe enum class name {} comparison
|
||||
#if __GNUC_MINOR__ < 4
|
||||
#define ZMQPP_COMPARABLE_ENUM enum
|
||||
#endif
|
||||
|
||||
#if __GNUC_MINOR__ == 4
|
||||
#if __GNUC_PATCHLEVEL__ < 1
|
||||
#undef ZMQPP_COMPARABLE_ENUM
|
||||
#define ZMQPP_COMPARABLE_ENUM enum
|
||||
#endif // if __GNUC_PATCHLEVEL__ < 1
|
||||
#endif // if __GNUC_MINOR__ == 4
|
||||
|
||||
// Deal with older gcc not supporting C++0x lambda function
|
||||
#if __GNUC_MINOR__ < 5
|
||||
#define ZMQPP_IGNORE_LAMBDA_FUNCTION_TESTS
|
||||
#define ZMQPP_EXPLICITLY_DELETED
|
||||
#endif // if __GNUC_MINOR__ < 5
|
||||
|
||||
// Deal with older gcc not supporting C++0x nullptr
|
||||
#if __GNUC_MINOR__ < 6
|
||||
#define nullptr NULL
|
||||
#define NOEXCEPT
|
||||
#endif // if __GNUC_MINOR__ < 6
|
||||
|
||||
#endif // if __GNUC_ == 4
|
||||
#endif // if defined(__GNUC__) && !defined(__clang__)
|
||||
|
||||
#if defined(_MSC_VER)
|
||||
#define NOEXCEPT throw()
|
||||
#if _MSC_VER < 1800
|
||||
#define ZMQPP_EXPLICITLY_DELETED
|
||||
#endif // if _MSC_VER < 1800
|
||||
#if _MSC_VER < 1600
|
||||
#define nullptr NULL
|
||||
#define ZMQPP_IGNORE_LAMBDA_FUNCTION_TESTS
|
||||
#define ZMQPP_COMPARABLE_ENUM enum
|
||||
#endif // if _MSC_VER < 1600
|
||||
#endif // if defined(_MSC_VER)
|
||||
|
||||
// Generic state, assume a modern compiler
|
||||
#ifndef ZMQPP_COMPARABLE_ENUM
|
||||
#define ZMQPP_COMPARABLE_ENUM enum class
|
||||
#endif
|
||||
|
||||
#ifndef ZMQPP_EXPLICITLY_DELETED
|
||||
#define ZMQPP_EXPLICITLY_DELETED = delete
|
||||
#endif
|
||||
|
||||
#ifndef NOEXCEPT
|
||||
#define NOEXCEPT noexcept
|
||||
#endif
|
||||
|
||||
#endif /* ZMQPP_COMPATIBILITY_HPP_ */
|
||||
|
||||
@@ -0,0 +1,54 @@
|
||||
/**
|
||||
* \file
|
||||
*
|
||||
* \date 9 Aug 2011
|
||||
* \author Ben Gray (\@benjamg)
|
||||
*/
|
||||
|
||||
#include "context.hpp"
|
||||
|
||||
namespace zmqpp
|
||||
{
|
||||
|
||||
void context::terminate()
|
||||
{
|
||||
int result;
|
||||
do
|
||||
{
|
||||
#if (ZMQ_VERSION_MAJOR < 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR < 2))
|
||||
result = zmq_term(_context);
|
||||
#else
|
||||
result = zmq_ctx_destroy(_context);
|
||||
#endif
|
||||
} while (result != 0 && zmq_errno() == EINTR);
|
||||
if (result != 0) { throw zmq_internal_exception(); }
|
||||
_context = nullptr;
|
||||
}
|
||||
|
||||
#if (ZMQ_VERSION_MAJOR > 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR >= 2))
|
||||
void context::set(context_option const option, int const value)
|
||||
{
|
||||
if (nullptr == _context) { throw invalid_instance("context is invalid"); }
|
||||
|
||||
if (0 != zmq_ctx_set(_context, static_cast<int>(option), value))
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
}
|
||||
|
||||
int context::get(context_option const option)
|
||||
{
|
||||
if (nullptr == _context) { throw invalid_instance("context is invalid"); }
|
||||
|
||||
int result = zmq_ctx_get(_context, static_cast<int>(option));
|
||||
|
||||
if (result < 0)
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
|
||||
return result;
|
||||
}
|
||||
#endif
|
||||
|
||||
}
|
||||
@@ -0,0 +1,184 @@
|
||||
/**
|
||||
* \file
|
||||
*
|
||||
* \date 9 Aug 2011
|
||||
* \author Ben Gray (\@benjamg)
|
||||
*/
|
||||
|
||||
#ifndef ZMQPP_CONTEXT_HPP_
|
||||
#define ZMQPP_CONTEXT_HPP_
|
||||
|
||||
#include <cassert>
|
||||
|
||||
#include <zmq.h>
|
||||
|
||||
#include "compatibility.hpp"
|
||||
#include "exception.hpp"
|
||||
#if (ZMQ_VERSION_MAJOR > 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR >= 2))
|
||||
#include "context_options.hpp"
|
||||
#endif
|
||||
|
||||
namespace zmqpp
|
||||
{
|
||||
|
||||
/*!
|
||||
* The context class represents internal zmq context and io threads.
|
||||
*
|
||||
* By default the context class will create one thread, however this can be
|
||||
* overridden in the constructor.
|
||||
*
|
||||
* The context class is the only object that can be considered thread safe.
|
||||
*
|
||||
* All sockets using endpoints other than inproc require the context to have
|
||||
* at least one thread.
|
||||
*
|
||||
* This class is c++0x move supporting and cannot be copied.
|
||||
*/
|
||||
class context
|
||||
{
|
||||
public:
|
||||
|
||||
#if (ZMQ_VERSION_MAJOR < 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR < 2))
|
||||
/*!
|
||||
* Initialise the 0mq context.
|
||||
*
|
||||
* If only inproc is used then the context may be created with zero threads.
|
||||
* Any inproc endpoint using sockets must be created using the same context.
|
||||
*
|
||||
* The context is thread safe an may be used anywhere in your application,
|
||||
* however there is no requirement (other than inproc restrictions) for you
|
||||
* to do this.
|
||||
*
|
||||
* \param threads an integer argument for the number of required threads. Defaults to 1.
|
||||
*/
|
||||
context(int const& threads = 1)
|
||||
#else
|
||||
/*!
|
||||
* Initialise the 0mq context.
|
||||
*
|
||||
* The context is thread safe an may be used anywhere in your application,
|
||||
* however there is no requirement (other than inproc restrictions) for you
|
||||
* to do this.
|
||||
*/
|
||||
context()
|
||||
#endif
|
||||
: _context(nullptr)
|
||||
{
|
||||
#if (ZMQ_VERSION_MAJOR < 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR < 2))
|
||||
_context = zmq_init(threads);
|
||||
#else
|
||||
_context = zmq_ctx_new();
|
||||
#endif
|
||||
|
||||
if (nullptr == _context)
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
}
|
||||
|
||||
/*!
|
||||
* Closes the 0mq context.
|
||||
*
|
||||
* Any blocking calls other than a socket close will return with an error.
|
||||
*
|
||||
* If there are open sockets will block while zmq internal buffers are
|
||||
* processed up to a limit specified by that sockets linger option.
|
||||
*/
|
||||
~context() NOEXCEPT
|
||||
{
|
||||
if (nullptr != _context)
|
||||
{
|
||||
terminate();
|
||||
}
|
||||
}
|
||||
|
||||
/*!
|
||||
* Move supporting constructor.
|
||||
*
|
||||
* Allows zero-copy move semantics to be used with this class.
|
||||
*
|
||||
* \param source a rvalue instance of the object who's internals we wish to steal.
|
||||
*/
|
||||
context(context&& source) NOEXCEPT
|
||||
: _context(source._context)
|
||||
{
|
||||
source._context = nullptr;
|
||||
}
|
||||
|
||||
/*!
|
||||
* Move supporting operator.
|
||||
*
|
||||
* Allows zero-copy move semantics to be used with this class.
|
||||
*
|
||||
* \param source an rvalue instance of the context who's internals we wish to steal.
|
||||
*/
|
||||
context& operator=(context&& source) NOEXCEPT
|
||||
{
|
||||
std::swap( _context, source._context );
|
||||
return *this;
|
||||
}
|
||||
|
||||
/*!
|
||||
* Terminate the current context.
|
||||
*
|
||||
* Any blocking calls other than a socket close will return with an error.
|
||||
*
|
||||
* If there are open sockets will block while zmq internal buffers are
|
||||
* processed up to a limit specified by that sockets linger option.
|
||||
*/
|
||||
void terminate();
|
||||
|
||||
#if (ZMQ_VERSION_MAJOR > 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR >= 2))
|
||||
/*!
|
||||
* Set the value of an option in the underlaying zmq context.
|
||||
*
|
||||
* \param option a valid ::context_option
|
||||
* \param value to set the option to
|
||||
*/
|
||||
void set(context_option const option, int const value);
|
||||
|
||||
/*!
|
||||
* Get a context option from the underlaying zmq context.
|
||||
*
|
||||
* \param option a valid ::context_option
|
||||
* \return context option value
|
||||
*/
|
||||
int get(context_option const option);
|
||||
#endif
|
||||
|
||||
/*!
|
||||
* Validity checking of the context
|
||||
*
|
||||
* Checks if the underlying 0mq context for this instance is valid.
|
||||
*
|
||||
* Contexts should always be valid unless people are doing 'fun' things with
|
||||
* std::move.
|
||||
*
|
||||
* \return boolean true if the object is valid.
|
||||
*/
|
||||
operator bool() const NOEXCEPT
|
||||
{
|
||||
return nullptr != _context;
|
||||
}
|
||||
|
||||
/*!
|
||||
* Access to the raw 0mq context
|
||||
*
|
||||
* \return void pointer to the underlying 0mq context.
|
||||
*/
|
||||
operator void*() const NOEXCEPT
|
||||
{
|
||||
return _context;
|
||||
}
|
||||
|
||||
private:
|
||||
void* _context;
|
||||
|
||||
// No copy - private and not implemented
|
||||
context(context const&) ZMQPP_EXPLICITLY_DELETED;
|
||||
context& operator=(context const&) NOEXCEPT ZMQPP_EXPLICITLY_DELETED;
|
||||
};
|
||||
|
||||
}
|
||||
|
||||
#endif /* ZMQPP_CONTEXT_HPP_ */
|
||||
@@ -0,0 +1,26 @@
|
||||
/**
|
||||
* \file
|
||||
*
|
||||
* \date 3 Jul 2013
|
||||
* \author Ben Gray (\@benjamg)
|
||||
*/
|
||||
|
||||
#ifndef ZMQPP_CONTEXT_OPTIONS_HPP_
|
||||
#define ZMQPP_CONTEXT_OPTIONS_HPP_
|
||||
|
||||
namespace zmqpp
|
||||
{
|
||||
|
||||
/** \todo Expand the information on the options to make it actually useful. */
|
||||
/*!
|
||||
* \brief possible Context options in zmq
|
||||
*/
|
||||
|
||||
enum class context_option {
|
||||
io_threads = ZMQ_IO_THREADS, /*!< I/O thread count */
|
||||
max_sockets = ZMQ_MAX_SOCKETS, /*!< Maximum supported sockets */
|
||||
};
|
||||
|
||||
}
|
||||
|
||||
#endif /* ZMQPP_CONTEXT_OPTIONS_HPP_ */
|
||||
@@ -0,0 +1,87 @@
|
||||
/**
|
||||
* \file
|
||||
*
|
||||
* \date 9 Aug 2011
|
||||
* \author Ben Gray (\@benjamg)
|
||||
*/
|
||||
|
||||
#ifndef ZMQPP_EXCEPTION_HPP_
|
||||
#define ZMQPP_EXCEPTION_HPP_
|
||||
|
||||
#include <stdexcept>
|
||||
#include <string>
|
||||
|
||||
#include <zmq.h>
|
||||
|
||||
namespace zmqpp
|
||||
{
|
||||
|
||||
/** \todo Have a larger variety of exceptions with better state debug information */
|
||||
|
||||
/*!
|
||||
* Represents the base zmqpp exception.
|
||||
*
|
||||
* All zmqpp runtime exceptions are children of this class.
|
||||
* The class itself does not provide any special access fields but it only
|
||||
* for convince when catching exceptions.
|
||||
*
|
||||
* The class extends std::runtime_error.
|
||||
*
|
||||
*/
|
||||
class exception : public std::runtime_error
|
||||
{
|
||||
public:
|
||||
/*!
|
||||
* Standard exception constructor.
|
||||
*
|
||||
* \param message a string representing the error message.
|
||||
*/
|
||||
exception(std::string const& message)
|
||||
: std::runtime_error(message)
|
||||
{ }
|
||||
};
|
||||
|
||||
/*!
|
||||
* Represents an attempt to use an invalid object.
|
||||
*
|
||||
* Objects may be invalid initially or after a shutdown or close.
|
||||
*/
|
||||
class invalid_instance : public exception
|
||||
{
|
||||
public:
|
||||
invalid_instance(std::string const& message)
|
||||
: exception(message)
|
||||
{ }
|
||||
};
|
||||
|
||||
/*!
|
||||
* Represents internal zmq errors.
|
||||
*
|
||||
* Any error response from the zmq bindings will be wrapped in this error.
|
||||
*
|
||||
* The class provides access to the zmq error number via zmq_error().
|
||||
*/
|
||||
class zmq_internal_exception : public exception
|
||||
{
|
||||
public:
|
||||
/*!
|
||||
* Uses the zmq functions to pull out error messages and numbers.
|
||||
*/
|
||||
zmq_internal_exception()
|
||||
: exception(zmq_strerror(zmq_errno()))
|
||||
, _error(zmq_errno())
|
||||
{ }
|
||||
|
||||
/*!
|
||||
* Retrieve the zmq error number associated with this exception.
|
||||
* \return zmq error number
|
||||
*/
|
||||
int zmq_error() const { return _error; }
|
||||
|
||||
private:
|
||||
int _error;
|
||||
};
|
||||
|
||||
}
|
||||
|
||||
#endif /* ZMQPP_EXCEPTION_HPP_ */
|
||||
@@ -0,0 +1,95 @@
|
||||
/**
|
||||
* \file
|
||||
*
|
||||
* \date 8 Jan 2014
|
||||
* \author Ben Gray (\@benjamg)
|
||||
*/
|
||||
|
||||
#include <cassert>
|
||||
#include <cstring>
|
||||
|
||||
#include "exception.hpp"
|
||||
#include "frame.hpp"
|
||||
|
||||
namespace zmqpp {
|
||||
|
||||
frame::frame()
|
||||
: _sent( false )
|
||||
{
|
||||
if( 0 != zmq_msg_init( &_msg ) )
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
}
|
||||
|
||||
frame::frame(size_t const size)
|
||||
: _sent( false )
|
||||
{
|
||||
if( 0 != zmq_msg_init_size( &_msg, size ) )
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
}
|
||||
|
||||
frame::frame(void const* part, size_t const size)
|
||||
: _sent( false )
|
||||
{
|
||||
if( 0 != zmq_msg_init_size( &_msg, size ) )
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
|
||||
void* msg_data = zmq_msg_data( &_msg );
|
||||
memcpy( msg_data, part, size );
|
||||
}
|
||||
|
||||
frame::frame(void* part, size_t const size, zmq_free_fn *ffn, void *hint)
|
||||
: _sent( false )
|
||||
{
|
||||
if( 0 != zmq_msg_init_data( &_msg, part, size, ffn, hint ) )
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
}
|
||||
|
||||
frame::~frame()
|
||||
{
|
||||
#ifndef NDEBUG // unused assert variable in release
|
||||
int result = zmq_msg_close( &_msg );
|
||||
assert(0 == result);
|
||||
#else
|
||||
zmq_msg_close( &_msg );
|
||||
#endif // NDEBUG
|
||||
}
|
||||
|
||||
frame::frame(frame&& other)
|
||||
: _sent( other._sent )
|
||||
{
|
||||
zmq_msg_init( &_msg );
|
||||
zmq_msg_move( &_msg, &other._msg );
|
||||
other._sent = false;
|
||||
}
|
||||
|
||||
frame& frame::operator=(frame&& other)
|
||||
{
|
||||
zmq_msg_init( &_msg );
|
||||
zmq_msg_move( &_msg, &other._msg );
|
||||
std::swap( _sent, other._sent );
|
||||
|
||||
return *this;
|
||||
}
|
||||
|
||||
frame frame::copy() const
|
||||
{
|
||||
frame other( size() );
|
||||
other._sent = _sent;
|
||||
|
||||
if( 0 != zmq_msg_copy( &other._msg, const_cast<zmq_msg_t*>(&_msg) ) )
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
|
||||
return other;
|
||||
}
|
||||
|
||||
} // namespace zmqpp
|
||||
@@ -0,0 +1,58 @@
|
||||
/**
|
||||
* \file
|
||||
*
|
||||
* \date 8 Jan 2014
|
||||
* \author Ben Gray (\@benjamg)
|
||||
*/
|
||||
|
||||
#ifndef ZMQPP_MESSAGE_FRAME_HPP_
|
||||
#define ZMQPP_MESSAGE_FRAME_HPP_
|
||||
|
||||
#include <zmq.h>
|
||||
|
||||
#include "compatibility.hpp"
|
||||
|
||||
namespace zmqpp {
|
||||
|
||||
/*!
|
||||
* \brief an internal frame wrapper for a single zmq message
|
||||
*
|
||||
* This frame wrapper consists of a zmq message and meta data it is used
|
||||
* by the zmqpp message class to keep track of parts in the internal
|
||||
* queue. It is unlikely you need to use this class.
|
||||
*/
|
||||
class frame
|
||||
{
|
||||
public:
|
||||
frame();
|
||||
frame(size_t const size);
|
||||
frame(void const* part, size_t const size);
|
||||
frame(void* part, size_t const size, zmq_free_fn *ffn, void *hint);
|
||||
|
||||
~frame();
|
||||
|
||||
bool is_sent() const { return _sent; }
|
||||
void const* data() const { return zmq_msg_data( const_cast<zmq_msg_t*>(&_msg) ); }
|
||||
size_t size() const { return zmq_msg_size( const_cast<zmq_msg_t*>(&_msg) ); }
|
||||
|
||||
void mark_sent() { _sent = true; }
|
||||
zmq_msg_t& msg() { return _msg; }
|
||||
|
||||
// Move operators
|
||||
frame(frame&& other);
|
||||
frame& operator=(frame&& other);
|
||||
|
||||
frame copy() const;
|
||||
|
||||
private:
|
||||
bool _sent;
|
||||
zmq_msg_t _msg;
|
||||
|
||||
// Disable implicit copy support, code must request a copy to clone
|
||||
frame(frame const&) NOEXCEPT ZMQPP_EXPLICITLY_DELETED;
|
||||
frame& operator=(frame const&) NOEXCEPT ZMQPP_EXPLICITLY_DELETED;
|
||||
};
|
||||
|
||||
} // namespace zmqpp
|
||||
|
||||
#endif /* ZMQPP_MESSAGE_FRAME_HPP_ */
|
||||
@@ -0,0 +1,171 @@
|
||||
/**
|
||||
* \file
|
||||
*
|
||||
* \date 10 Aug 2011
|
||||
* \author Ben Gray (\@benjamg)
|
||||
*/
|
||||
|
||||
#ifndef ZMQPP_INET_HPP_
|
||||
#define ZMQPP_INET_HPP_
|
||||
|
||||
/** \todo cross-platform version of including headers. */
|
||||
// We get htons and htonl from here
|
||||
#ifdef _WIN32
|
||||
#include <WinSock2.h>
|
||||
#else
|
||||
#include <netinet/in.h>
|
||||
#endif
|
||||
|
||||
#include "compatibility.hpp"
|
||||
|
||||
namespace zmqpp
|
||||
{
|
||||
|
||||
/*!
|
||||
* \brief Possible byte order types.
|
||||
*
|
||||
* An enumeration of all the known order types, all two of them.
|
||||
* There is also an entry for unknown which is just used as a default.
|
||||
*/
|
||||
ZMQPP_COMPARABLE_ENUM order {
|
||||
big_endian, /*!< byte order is big endian */
|
||||
little_endian /*!< byte order is little endian */
|
||||
};
|
||||
|
||||
/*!
|
||||
* Common code for the 64bit versions of htons/htons and ntohs/ntohl
|
||||
*
|
||||
* As htons and ntohs (or htonl and ntohs) always just do the same thing, ie
|
||||
* swap bytes if the host order differs from network order or otherwise don't
|
||||
* do anything, it seemed silly to type the code twice.
|
||||
*
|
||||
* \note This code assumes network order is always big endian. Which it is.
|
||||
* \note The host endian is only checked once and afterwards assumed to remain
|
||||
* the same.
|
||||
*
|
||||
* \param value_to_check unsigned 64 bit integer to swap
|
||||
* \return swapped (or not) unsigned 64 bit integer
|
||||
*/
|
||||
inline uint64_t swap_if_needed(uint64_t const value_to_check)
|
||||
{
|
||||
static order host_order = (htonl(42) == 42) ? order::big_endian : order::little_endian;
|
||||
|
||||
if (order::big_endian == host_order)
|
||||
{
|
||||
return value_to_check;
|
||||
}
|
||||
|
||||
union {
|
||||
uint64_t integer;
|
||||
uint8_t bytes[8];
|
||||
} value { value_to_check };
|
||||
|
||||
std::swap(value.bytes[0], value.bytes[7]);
|
||||
std::swap(value.bytes[1], value.bytes[6]);
|
||||
std::swap(value.bytes[2], value.bytes[5]);
|
||||
std::swap(value.bytes[3], value.bytes[4]);
|
||||
|
||||
return value.integer;
|
||||
}
|
||||
|
||||
/*!
|
||||
* 64 bit version of the htons/htonl
|
||||
*
|
||||
* I've used the name htonll to try and keep with the htonl naming scheme.
|
||||
*
|
||||
* \param hostlonglong unsigned 64 bit host order integer
|
||||
* \return unsigned 64 bit network order integer
|
||||
*/
|
||||
inline uint64_t htonll(uint64_t const hostlonglong)
|
||||
{
|
||||
return zmqpp::swap_if_needed(hostlonglong);
|
||||
}
|
||||
|
||||
/*!
|
||||
* 64 bit version of the ntohs/ntohl
|
||||
*
|
||||
* I've used the name htonll to try and keep with the htonl naming scheme.
|
||||
*
|
||||
* \param networklonglong unsigned 64 bit network order integer
|
||||
* \return unsigned 64 bit host order integer
|
||||
*/
|
||||
inline uint64_t ntohll(uint64_t const networklonglong)
|
||||
{
|
||||
return zmqpp::swap_if_needed(networklonglong);
|
||||
}
|
||||
|
||||
/*!
|
||||
* floating point version of the htons/htonl
|
||||
*
|
||||
* \param value host order floating point
|
||||
* \returns network order floating point
|
||||
*/
|
||||
inline float htonf(float value)
|
||||
{
|
||||
assert(sizeof(float) == sizeof(uint32_t));
|
||||
|
||||
uint32_t temp;
|
||||
memcpy(&temp, &value, sizeof(uint32_t));
|
||||
temp = htonl( temp );
|
||||
memcpy(&value, &temp, sizeof(uint32_t));
|
||||
|
||||
return value;
|
||||
}
|
||||
|
||||
/*!
|
||||
* floating point version of the ntohs/ntohl
|
||||
*
|
||||
* \param value network order float
|
||||
* \returns host order float
|
||||
*/
|
||||
inline float ntohf(float value)
|
||||
{
|
||||
assert(sizeof(float) == sizeof(uint32_t));
|
||||
|
||||
uint32_t temp;
|
||||
memcpy(&temp, &value, sizeof(uint32_t));
|
||||
temp = ntohl( temp );
|
||||
memcpy(&value, &temp, sizeof(uint32_t));
|
||||
|
||||
return value;
|
||||
}
|
||||
|
||||
/*!
|
||||
* double precision floating point version of the htons/htonl
|
||||
*
|
||||
* \param value host order double precision floating point
|
||||
* \returns network order double precision floating point
|
||||
*/
|
||||
inline double htond(double value)
|
||||
{
|
||||
assert(sizeof(double) == sizeof(uint64_t));
|
||||
|
||||
uint64_t temp;
|
||||
memcpy(&temp, &value, sizeof(uint64_t));
|
||||
temp = zmqpp::htonll(temp);
|
||||
memcpy(&value, &temp, sizeof(uint64_t));
|
||||
|
||||
return value;
|
||||
}
|
||||
|
||||
/*!
|
||||
* double precision floating point version of the ntohs/ntohl
|
||||
*
|
||||
* \param value network order double precision floating point
|
||||
* \returns host order double precision floating point
|
||||
*/
|
||||
inline double ntohd(double value)
|
||||
{
|
||||
assert(sizeof(double) == sizeof(uint64_t));
|
||||
|
||||
uint64_t temp;
|
||||
memcpy(&temp, &value, sizeof(uint64_t));
|
||||
temp = zmqpp::ntohll(temp);
|
||||
memcpy(&value, &temp, sizeof(uint64_t));
|
||||
|
||||
return value;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
#endif /* INET_HPP_ */
|
||||
@@ -0,0 +1,454 @@
|
||||
/*
|
||||
* Created on: 9 Aug 2011
|
||||
* Author: Ben Gray (@benjamg)
|
||||
*/
|
||||
|
||||
#include <cassert>
|
||||
#include <cstring>
|
||||
|
||||
#include "exception.hpp"
|
||||
#include "inet.hpp"
|
||||
#include "message.hpp"
|
||||
|
||||
namespace zmqpp
|
||||
{
|
||||
|
||||
/*!
|
||||
* \brief internal construct
|
||||
* \internal handles bubbling callback from zmq c style to the c++ functor provided
|
||||
*/
|
||||
struct callback_releaser
|
||||
{
|
||||
message::release_function func;
|
||||
};
|
||||
|
||||
message::message()
|
||||
: _parts()
|
||||
, _read_cursor(0)
|
||||
{
|
||||
}
|
||||
|
||||
message::~message()
|
||||
{
|
||||
_parts.clear();
|
||||
}
|
||||
|
||||
size_t message::parts() const
|
||||
{
|
||||
return _parts.size();
|
||||
}
|
||||
|
||||
/*
|
||||
* The two const_casts in size and raw_data are a little bit hacky
|
||||
* but neither of these methods called this way actually modify data
|
||||
* so accurately represent the intent of these calls.
|
||||
*/
|
||||
|
||||
size_t message::size(size_t const part /* = 0 */) const
|
||||
{
|
||||
if(part >= _parts.size())
|
||||
{
|
||||
throw exception("attempting to request a message part outside the valid range");
|
||||
}
|
||||
|
||||
return _parts[part].size();
|
||||
}
|
||||
|
||||
void const* message::raw_data(size_t const part /* = 0 */) const
|
||||
{
|
||||
if(part >= _parts.size())
|
||||
{
|
||||
throw exception("attempting to request a message part outside the valid range");
|
||||
}
|
||||
|
||||
return _parts[part].data();
|
||||
}
|
||||
|
||||
zmq_msg_t& message::raw_msg(size_t const part /* = 0 */)
|
||||
{
|
||||
if(part >= _parts.size())
|
||||
{
|
||||
throw exception("attempting to request a message part outside the valid range");
|
||||
}
|
||||
|
||||
return _parts[part].msg();
|
||||
}
|
||||
|
||||
zmq_msg_t& message::raw_new_msg()
|
||||
{
|
||||
_parts.push_back( frame() );
|
||||
|
||||
return _parts.back().msg();
|
||||
}
|
||||
|
||||
zmq_msg_t& message::raw_new_msg(size_t const reserve_data_size)
|
||||
{
|
||||
_parts.push_back( frame(reserve_data_size) );
|
||||
|
||||
return _parts.back().msg();
|
||||
}
|
||||
|
||||
std::string message::get(size_t const part /* = 0 */) const
|
||||
{
|
||||
return std::string(static_cast<char const*>(raw_data(part)), size(part));
|
||||
}
|
||||
|
||||
|
||||
// Move operators will take ownership of message parts without copying
|
||||
void message::move(void* part, size_t const size, release_function const& release)
|
||||
{
|
||||
callback_releaser* hint = new callback_releaser();
|
||||
hint->func = release;
|
||||
|
||||
_parts.push_back( frame( part, size, &message::release_callback, hint ) );
|
||||
}
|
||||
|
||||
// Stream reader style
|
||||
void message::reset_read_cursor()
|
||||
{
|
||||
_read_cursor = 0;
|
||||
}
|
||||
|
||||
void message::get(int8_t& integer, size_t const part) const
|
||||
{
|
||||
assert(sizeof(int8_t) == size(part));
|
||||
|
||||
int8_t const* byte = static_cast<int8_t const*>(raw_data(part));
|
||||
integer = *byte;
|
||||
}
|
||||
|
||||
void message::get(int16_t& integer, size_t const part) const
|
||||
{
|
||||
assert(sizeof(int16_t) == size(part));
|
||||
|
||||
uint16_t const* network_order = static_cast<uint16_t const*>(raw_data(part));
|
||||
integer = static_cast<int16_t>(ntohs(*network_order));
|
||||
}
|
||||
|
||||
void message::get(int32_t& integer, size_t const part) const
|
||||
{
|
||||
assert(sizeof(int32_t) == size(part));
|
||||
|
||||
uint32_t const* network_order = static_cast<uint32_t const*>(raw_data(part));
|
||||
integer = static_cast<int32_t>(htonl(*network_order));
|
||||
}
|
||||
|
||||
void message::get(int64_t& integer, size_t const part) const
|
||||
{
|
||||
assert(sizeof(int64_t) == size(part));
|
||||
|
||||
uint64_t const* network_order = static_cast<uint64_t const*>(raw_data(part));
|
||||
integer = static_cast<int64_t>(zmqpp::htonll(*network_order));
|
||||
}
|
||||
|
||||
void message::get(uint8_t& unsigned_integer, size_t const part) const
|
||||
{
|
||||
assert(sizeof(uint8_t) == size(part));
|
||||
|
||||
uint8_t const* byte = static_cast<uint8_t const*>(raw_data(part));
|
||||
unsigned_integer = *byte;
|
||||
}
|
||||
|
||||
void message::get(uint16_t& unsigned_integer, size_t const part) const
|
||||
{
|
||||
assert(sizeof(uint16_t) == size(part));
|
||||
|
||||
uint16_t const* network_order = static_cast<uint16_t const*>(raw_data(part));
|
||||
unsigned_integer = ntohs(*network_order);
|
||||
}
|
||||
|
||||
void message::get(uint32_t& unsigned_integer, size_t const part) const
|
||||
{
|
||||
assert(sizeof(uint32_t) == size(part));
|
||||
|
||||
uint32_t const* network_order = static_cast<uint32_t const*>(raw_data(part));
|
||||
unsigned_integer = ntohl(*network_order);
|
||||
}
|
||||
|
||||
void message::get(uint64_t& unsigned_integer, size_t const part) const
|
||||
{
|
||||
assert(sizeof(uint64_t) == size(part));
|
||||
|
||||
uint64_t const* network_order = static_cast<uint64_t const*>(raw_data(part));
|
||||
unsigned_integer = zmqpp::ntohll(*network_order);
|
||||
}
|
||||
|
||||
void message::get(float& floating_point, size_t const part) const
|
||||
{
|
||||
assert(sizeof(float) == size(part));
|
||||
|
||||
float const* network_order = static_cast<float const*>(raw_data(part));
|
||||
floating_point = zmqpp::ntohf(*network_order);
|
||||
}
|
||||
|
||||
void message::get(double& double_precision, size_t const part) const
|
||||
{
|
||||
assert(sizeof(double) == size(part));
|
||||
|
||||
double const* network_order = static_cast<double const*>(raw_data(part));
|
||||
double_precision = zmqpp::ntohd(*network_order);
|
||||
}
|
||||
|
||||
void message::get(bool& boolean, size_t const part) const
|
||||
{
|
||||
assert(sizeof(uint8_t) == size(part));
|
||||
|
||||
uint8_t const* byte = static_cast<uint8_t const*>(raw_data(part));
|
||||
boolean = (*byte != 0);
|
||||
}
|
||||
|
||||
void message::get(std::string& string, size_t const part) const
|
||||
{
|
||||
string.assign( get(part) );
|
||||
}
|
||||
|
||||
|
||||
// Stream writer style - these all use copy styles
|
||||
message& message::operator<<(int8_t const integer)
|
||||
{
|
||||
add(reinterpret_cast<void const*>(&integer), sizeof(int8_t));
|
||||
return *this;
|
||||
}
|
||||
|
||||
message& message::operator<<(int16_t const integer)
|
||||
{
|
||||
uint16_t network_order = htons(static_cast<uint16_t>(integer));
|
||||
add(reinterpret_cast<void const*>(&network_order), sizeof(uint16_t));
|
||||
|
||||
return *this;
|
||||
}
|
||||
|
||||
message& message::operator<<(int32_t const integer)
|
||||
{
|
||||
uint32_t network_order = htonl(static_cast<uint32_t>(integer));
|
||||
add(reinterpret_cast<void const*>(&network_order), sizeof(uint32_t));
|
||||
|
||||
return *this;
|
||||
}
|
||||
|
||||
message& message::operator<<(int64_t const integer)
|
||||
{
|
||||
uint64_t network_order = zmqpp::htonll(static_cast<uint64_t>(integer));
|
||||
add(reinterpret_cast<void const*>(&network_order), sizeof(uint64_t));
|
||||
|
||||
return *this;
|
||||
}
|
||||
|
||||
|
||||
message& message::operator<<(uint8_t const unsigned_integer)
|
||||
{
|
||||
add(reinterpret_cast<void const*>(&unsigned_integer), sizeof(uint8_t));
|
||||
return *this;
|
||||
}
|
||||
|
||||
message& message::operator<<(uint16_t const unsigned_integer)
|
||||
{
|
||||
uint16_t network_order = htons(unsigned_integer);
|
||||
add(reinterpret_cast<void const*>(&network_order), sizeof(uint16_t));
|
||||
|
||||
return *this;
|
||||
}
|
||||
|
||||
message& message::operator<<(uint32_t const unsigned_integer)
|
||||
{
|
||||
uint32_t network_order = htonl(unsigned_integer);
|
||||
add(reinterpret_cast<void const*>(&network_order), sizeof(uint32_t));
|
||||
|
||||
return *this;
|
||||
}
|
||||
|
||||
message& message::operator<<(uint64_t const unsigned_integer)
|
||||
{
|
||||
uint64_t network_order = zmqpp::htonll(unsigned_integer);
|
||||
add(reinterpret_cast<void const*>(&network_order), sizeof(uint64_t));
|
||||
|
||||
return *this;
|
||||
}
|
||||
|
||||
message& message::operator<<(float const floating_point)
|
||||
{
|
||||
assert(sizeof(float) == 4);
|
||||
|
||||
float network_order = zmqpp::htonf(floating_point);
|
||||
add(&network_order, sizeof(float));
|
||||
|
||||
return *this;
|
||||
}
|
||||
|
||||
message& message::operator<<(double const double_precision)
|
||||
{
|
||||
assert(sizeof(double) == 8);
|
||||
|
||||
double network_order = zmqpp::htond(double_precision);
|
||||
add(&network_order, sizeof(double));
|
||||
|
||||
return *this;
|
||||
}
|
||||
|
||||
message& message::operator<<(bool const boolean)
|
||||
{
|
||||
uint8_t byte = (boolean) ? 1 : 0;
|
||||
add(reinterpret_cast<void const*>(&byte), sizeof(uint8_t));
|
||||
|
||||
return *this;
|
||||
}
|
||||
|
||||
message& message::operator<<(char const* c_string)
|
||||
{
|
||||
add(reinterpret_cast<void const*>(c_string), strlen(c_string));
|
||||
return *this;
|
||||
}
|
||||
|
||||
message& message::operator<<(std::string const& string)
|
||||
{
|
||||
add(reinterpret_cast<void const*>(string.data()), string.size());
|
||||
return *this;
|
||||
}
|
||||
|
||||
void message::push_front(void const* part, size_t const size)
|
||||
{
|
||||
_parts.emplace( _parts.begin(), part, size );
|
||||
}
|
||||
|
||||
void message::push_front(int8_t const integer)
|
||||
{
|
||||
push_front(&integer, sizeof(int8_t));
|
||||
}
|
||||
|
||||
void message::push_front(int16_t const integer)
|
||||
{
|
||||
uint16_t network_order = htons(static_cast<uint16_t>(integer));
|
||||
push_front(&network_order, sizeof(uint16_t));
|
||||
}
|
||||
|
||||
void message::push_front(int32_t const integer)
|
||||
{
|
||||
uint32_t network_order = htonl(static_cast<uint32_t>(integer));
|
||||
push_front(&network_order, sizeof(uint32_t));
|
||||
}
|
||||
|
||||
void message::push_front(int64_t const integer)
|
||||
{
|
||||
uint64_t network_order = zmqpp::htonll(static_cast<uint64_t>(integer));
|
||||
push_front(&network_order, sizeof(uint64_t));
|
||||
}
|
||||
|
||||
|
||||
void message::push_front(uint8_t const unsigned_integer)
|
||||
{
|
||||
push_front(&unsigned_integer, sizeof(uint8_t));
|
||||
}
|
||||
|
||||
void message::push_front(uint16_t const unsigned_integer)
|
||||
{
|
||||
uint16_t network_order = htons(unsigned_integer);
|
||||
push_front(&network_order, sizeof(uint16_t));
|
||||
}
|
||||
|
||||
void message::push_front(uint32_t const unsigned_integer)
|
||||
{
|
||||
uint32_t network_order = htonl(unsigned_integer);
|
||||
push_front(&network_order, sizeof(uint32_t));
|
||||
}
|
||||
|
||||
void message::push_front(uint64_t const unsigned_integer)
|
||||
{
|
||||
uint64_t network_order = zmqpp::htonll(unsigned_integer);
|
||||
push_front(&network_order, sizeof(uint64_t));
|
||||
}
|
||||
|
||||
void message::push_front(float const floating_point)
|
||||
{
|
||||
assert(sizeof(float) == 4);
|
||||
|
||||
float network_order = zmqpp::htonf(floating_point);
|
||||
push_front(&network_order, sizeof(float));
|
||||
}
|
||||
|
||||
void message::push_front(double const double_precision)
|
||||
{
|
||||
assert(sizeof(double) == 8);
|
||||
|
||||
double network_order = zmqpp::htond(double_precision);
|
||||
push_front(&network_order, sizeof(double));
|
||||
}
|
||||
|
||||
void message::push_front(bool const boolean)
|
||||
{
|
||||
uint8_t byte = (boolean) ? 1 : 0;
|
||||
push_front(&byte, sizeof(uint8_t));
|
||||
}
|
||||
|
||||
void message::push_front(char const* c_string)
|
||||
{
|
||||
push_front(c_string, strlen(c_string));
|
||||
}
|
||||
|
||||
void message::push_front(std::string const& string)
|
||||
{
|
||||
push_front(string.data(), string.size());
|
||||
}
|
||||
|
||||
void message::pop_front()
|
||||
{
|
||||
_parts.erase( _parts.begin() );
|
||||
}
|
||||
|
||||
void message::pop_back()
|
||||
{
|
||||
_parts.pop_back();
|
||||
}
|
||||
|
||||
message::message(message&& source) NOEXCEPT
|
||||
: _parts()
|
||||
, _read_cursor(0)
|
||||
{
|
||||
std::swap(_parts, source._parts);
|
||||
}
|
||||
|
||||
message& message::operator=(message&& source) NOEXCEPT
|
||||
{
|
||||
std::swap(_parts, source._parts);
|
||||
return *this;
|
||||
}
|
||||
|
||||
message message::copy() const
|
||||
{
|
||||
message msg;
|
||||
msg.copy(*this);
|
||||
return msg;
|
||||
}
|
||||
|
||||
void message::copy(message const& source)
|
||||
{
|
||||
_parts.resize( source._parts.size() );
|
||||
for(size_t i = 0; i < source._parts.size(); ++i)
|
||||
{
|
||||
_parts[i] = source._parts[i].copy();
|
||||
}
|
||||
|
||||
// we don't need a copy of the releasers as we did data copies of the internal data,
|
||||
//_releasers = source._releasers;
|
||||
//_strings = source._strings
|
||||
}
|
||||
|
||||
// Used for internal tracking
|
||||
void message::sent(size_t const part)
|
||||
{
|
||||
// sanity check
|
||||
assert(!_parts[part].is_sent());
|
||||
_parts[part].mark_sent();
|
||||
}
|
||||
|
||||
// Note that these releasers are not thread safe, the only safety is provided by
|
||||
// the socket class taking ownership so no updates can happen while zmq does it's thing
|
||||
// If used in a custom class this has to be dealt with.
|
||||
void message::release_callback(void* data, void* hint)
|
||||
{
|
||||
callback_releaser* releaser = static_cast<callback_releaser*>(hint);
|
||||
releaser->func(data);
|
||||
|
||||
delete releaser;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,253 @@
|
||||
/**
|
||||
* \file
|
||||
*
|
||||
* \date 9 Aug 2011
|
||||
* \author Ben Gray (\@benjamg)
|
||||
*/
|
||||
|
||||
#ifndef ZMQPP_MESSAGE_HPP_
|
||||
#define ZMQPP_MESSAGE_HPP_
|
||||
|
||||
#include <functional>
|
||||
#include <string>
|
||||
#include <unordered_map>
|
||||
#include <vector>
|
||||
#include <utility>
|
||||
|
||||
#include <zmq.h>
|
||||
|
||||
#include "compatibility.hpp"
|
||||
#include "frame.hpp"
|
||||
|
||||
namespace zmqpp
|
||||
{
|
||||
|
||||
/*!
|
||||
* \brief a zmq message with optional multipart support
|
||||
*
|
||||
* A zmq message is made up of one or more parts which are sent together to
|
||||
* the target endpoints. zmq guarantees either the whole message or none
|
||||
* of the message will be delivered.
|
||||
*/
|
||||
class message
|
||||
{
|
||||
public:
|
||||
/*!
|
||||
* \brief callback to release user allocated data.
|
||||
*
|
||||
* The release function will be called on any void* moved part.
|
||||
* It must be thread safe to the extent that the callback may occur on
|
||||
* one of the context threads.
|
||||
*
|
||||
* The function called will be passed a single variable which is the
|
||||
* pointer to the memory allocated.
|
||||
*/
|
||||
typedef std::function<void (void*)> release_function;
|
||||
|
||||
message();
|
||||
~message();
|
||||
|
||||
template <typename T, typename ...Args>
|
||||
message(T const &part, Args &&...args)
|
||||
: message()
|
||||
{
|
||||
add(part, std::forward<Args>(args)...);
|
||||
}
|
||||
|
||||
size_t parts() const;
|
||||
size_t size(size_t const part) const;
|
||||
std::string get(size_t const part) const;
|
||||
|
||||
void get(int8_t& integer, size_t const part) const;
|
||||
void get(int16_t& integer, size_t const part) const;
|
||||
void get(int32_t& integer, size_t const part) const;
|
||||
void get(int64_t& integer, size_t const part) const;
|
||||
|
||||
void get(uint8_t& unsigned_integer, size_t const part) const;
|
||||
void get(uint16_t& unsigned_integer, size_t const part) const;
|
||||
void get(uint32_t& unsigned_integer, size_t const part) const;
|
||||
void get(uint64_t& unsigned_integer, size_t const part) const;
|
||||
|
||||
void get(float& floating_point, size_t const part) const;
|
||||
void get(double& double_precision, size_t const part) const;
|
||||
void get(bool& boolean, size_t const part) const;
|
||||
|
||||
void get(std::string& string, size_t const part) const;
|
||||
|
||||
// Warn: If a pointer type is requested the message (well zmq) still 'owns'
|
||||
// the data and will release it when the message object is freed.
|
||||
template<typename Type>
|
||||
Type get(size_t const part)
|
||||
{
|
||||
Type value;
|
||||
get(value, part);
|
||||
return value;
|
||||
}
|
||||
|
||||
template<int part=0, typename T, typename ...Args>
|
||||
void extract(T &nextpart, Args &...args)
|
||||
{
|
||||
assert(part < parts());
|
||||
get(nextpart,part);
|
||||
extract<part+1>(args...);
|
||||
}
|
||||
|
||||
template<int part=0, typename T>
|
||||
void extract(T &nextpart)
|
||||
{
|
||||
assert(part < parts());
|
||||
get(nextpart,part);
|
||||
}
|
||||
|
||||
// Raw get data operations, useful with data structures more than anything else
|
||||
// Warn: The message (well zmq) still 'owns' the data and will release it
|
||||
// when the message object is freed.
|
||||
template<typename Type>
|
||||
void get(Type*& value, size_t const part) const
|
||||
{
|
||||
value = static_cast<Type*>(raw_data(part));
|
||||
}
|
||||
|
||||
// Warn: The message (well zmq) still 'owns' the data and will release it
|
||||
// when the message object is freed.
|
||||
template<typename Type>
|
||||
void get(Type** value, size_t const part) const
|
||||
{
|
||||
*value = static_cast<Type*>(raw_data(part));
|
||||
}
|
||||
|
||||
// Move operators will take ownership of message parts without copying
|
||||
void move(void* part, size_t const size, release_function const& release);
|
||||
|
||||
// Raw move data operation, useful with data structures more than anything else
|
||||
template<typename Object>
|
||||
void move(Object *part)
|
||||
{
|
||||
move(part, sizeof(Object), &deleter_callback<Object>);
|
||||
}
|
||||
|
||||
// Copy operators will take copies of any data
|
||||
template<typename Type>
|
||||
void add(Type *part, size_t const size)
|
||||
{
|
||||
_parts.push_back( frame( part, size ) );
|
||||
}
|
||||
|
||||
|
||||
template<typename Type, typename ...Args>
|
||||
void add(Type const& part, Args &&...args)
|
||||
{
|
||||
*this << part;
|
||||
add(std::forward<Args>(args)...);
|
||||
}
|
||||
|
||||
template<typename Type>
|
||||
void add(Type const part)
|
||||
{
|
||||
*this << part;
|
||||
}
|
||||
|
||||
// Stream reader style
|
||||
void reset_read_cursor();
|
||||
|
||||
template<typename Type>
|
||||
message& operator>>(Type& value)
|
||||
{
|
||||
get(value, _read_cursor++);
|
||||
return *this;
|
||||
}
|
||||
|
||||
// Stream writer style - these all use copy styles
|
||||
message& operator<<(int8_t const integer);
|
||||
message& operator<<(int16_t const integer);
|
||||
message& operator<<(int32_t const integer);
|
||||
message& operator<<(int64_t const integer);
|
||||
|
||||
message& operator<<(uint8_t const unsigned_integer);
|
||||
message& operator<<(uint16_t const unsigned_integer);
|
||||
message& operator<<(uint32_t const unsigned_integer);
|
||||
message& operator<<(uint64_t const unsigned_integer);
|
||||
|
||||
message& operator<<(float const floating_point);
|
||||
message& operator<<(double const double_precision);
|
||||
message& operator<<(bool const boolean);
|
||||
|
||||
message& operator<<(char const* c_string);
|
||||
message& operator<<(std::string const& string);
|
||||
|
||||
// Queue manipulation
|
||||
void push_front(void const* part, size_t const size);
|
||||
|
||||
// TODO: unify conversion of types with the stream operators
|
||||
void push_front(int8_t const integer);
|
||||
void push_front(int16_t const integer);
|
||||
void push_front(int32_t const integer);
|
||||
void push_front(int64_t const integer);
|
||||
|
||||
void push_front(uint8_t const unsigned_integer);
|
||||
void push_front(uint16_t const unsigned_integer);
|
||||
void push_front(uint32_t const unsigned_integer);
|
||||
void push_front(uint64_t const unsigned_integer);
|
||||
|
||||
void push_front(float const floating_point);
|
||||
void push_front(double const double_precision);
|
||||
void push_front(bool const boolean);
|
||||
|
||||
void push_front(char const* c_string);
|
||||
void push_front(std::string const& string);
|
||||
|
||||
void pop_front();
|
||||
|
||||
void push_back(void const* part, size_t const size)
|
||||
{
|
||||
add( part, size );
|
||||
}
|
||||
|
||||
template<typename Type>
|
||||
void push_back(Type const part)
|
||||
{
|
||||
*this << part;
|
||||
}
|
||||
|
||||
void pop_back();
|
||||
|
||||
void remove(size_t const part);
|
||||
|
||||
// Move supporting
|
||||
message(message&& source) NOEXCEPT;
|
||||
message& operator=(message&& source) NOEXCEPT;
|
||||
|
||||
// Copy support
|
||||
message copy() const;
|
||||
void copy(message const& source);
|
||||
|
||||
// Used for internal tracking
|
||||
void sent(size_t const part);
|
||||
|
||||
// Access to raw zmq details
|
||||
void const* raw_data(size_t const part = 0) const;
|
||||
zmq_msg_t& raw_msg(size_t const part = 0);
|
||||
zmq_msg_t& raw_new_msg();
|
||||
zmq_msg_t& raw_new_msg(size_t const reserve_data_size);
|
||||
|
||||
private:
|
||||
typedef std::vector<frame> parts_type;
|
||||
parts_type _parts;
|
||||
size_t _read_cursor;
|
||||
|
||||
// Disable implicit copy support, code must request a copy to clone
|
||||
message(message const&) NOEXCEPT ZMQPP_EXPLICITLY_DELETED;
|
||||
message& operator=(message const&) NOEXCEPT ZMQPP_EXPLICITLY_DELETED;
|
||||
|
||||
static void release_callback(void* data, void* hint);
|
||||
|
||||
template<typename Object>
|
||||
static void deleter_callback(void* data)
|
||||
{
|
||||
delete static_cast<Object*>(data);
|
||||
}
|
||||
};
|
||||
|
||||
}
|
||||
|
||||
#endif /* ZMQPP_MESSAGE_HPP_ */
|
||||
@@ -0,0 +1,182 @@
|
||||
/*
|
||||
* Created on: 16 Aug 2011
|
||||
* Author: Ben Gray (@benjamg)
|
||||
*/
|
||||
|
||||
#include "exception.hpp"
|
||||
#include "socket.hpp"
|
||||
#include "poller.hpp"
|
||||
|
||||
#include <zmq.h>
|
||||
|
||||
namespace zmqpp
|
||||
{
|
||||
|
||||
const long poller::wait_forever = -1;
|
||||
const short poller::poll_none = 0;
|
||||
const short poller::poll_in = ZMQ_POLLIN;
|
||||
const short poller::poll_out = ZMQ_POLLOUT;
|
||||
const short poller::poll_error = ZMQ_POLLERR;
|
||||
|
||||
poller::poller()
|
||||
: _items()
|
||||
, _index()
|
||||
, _fdindex()
|
||||
{
|
||||
|
||||
}
|
||||
|
||||
poller::~poller()
|
||||
{
|
||||
_items.clear();
|
||||
_index.clear();
|
||||
_fdindex.clear();
|
||||
}
|
||||
|
||||
void poller::add(socket& socket, short const event /* = POLL_IN */)
|
||||
{
|
||||
zmq_pollitem_t item { socket, 0, event, 0 };
|
||||
|
||||
size_t index = _items.size();
|
||||
_items.push_back(item);
|
||||
_index[socket] = index;
|
||||
}
|
||||
|
||||
void poller::add(int const descriptor, short const event /* = POLL_IN */)
|
||||
{
|
||||
zmq_pollitem_t item { nullptr, descriptor, event, 0 };
|
||||
|
||||
size_t index = _items.size();
|
||||
_items.push_back(item);
|
||||
_fdindex[descriptor] = index;
|
||||
}
|
||||
|
||||
bool poller::has(socket_t const& socket)
|
||||
{
|
||||
return _index.find(socket) != _index.end();
|
||||
}
|
||||
|
||||
bool poller::has(int const descriptor)
|
||||
{
|
||||
return _fdindex.find(descriptor) != _fdindex.end();
|
||||
}
|
||||
|
||||
void poller::reindex(size_t const index)
|
||||
{
|
||||
if ( nullptr != _items[index].socket )
|
||||
{
|
||||
auto found = _index.find( _items[index].socket );
|
||||
if (_index.end() == found) { throw exception("unable to reindex socket in poller"); }
|
||||
found->second = index;
|
||||
}
|
||||
else
|
||||
{
|
||||
auto found = _fdindex.find( _items[index].fd );
|
||||
if (_fdindex.end() == found) { throw exception("unable to reindex file descriptor in poller"); }
|
||||
found->second = index;
|
||||
}
|
||||
}
|
||||
|
||||
void poller::remove(socket_t const& socket)
|
||||
{
|
||||
auto found = _index.find(socket);
|
||||
if (_index.end() == found) { return; }
|
||||
|
||||
if ( _items.size() - 1 == found->second )
|
||||
{
|
||||
_items.pop_back();
|
||||
_index.erase(found);
|
||||
return;
|
||||
}
|
||||
|
||||
std::swap(_items[found->second], _items.back());
|
||||
_items.pop_back();
|
||||
|
||||
auto index = found->second;
|
||||
_index.erase(found);
|
||||
|
||||
reindex( index );
|
||||
}
|
||||
|
||||
void poller::remove(int const descriptor)
|
||||
{
|
||||
auto found = _fdindex.find(descriptor);
|
||||
if (_fdindex.end() == found) { return; }
|
||||
|
||||
if ( _items.size() - 1 == found->second )
|
||||
{
|
||||
_items.pop_back();
|
||||
_fdindex.erase(found);
|
||||
return;
|
||||
}
|
||||
|
||||
std::swap(_items[found->second], _items.back());
|
||||
_items.pop_back();
|
||||
|
||||
auto index = found->second;
|
||||
_fdindex.erase(found);
|
||||
|
||||
reindex( index );
|
||||
}
|
||||
|
||||
void poller::check_for(socket const& socket, short const event)
|
||||
{
|
||||
auto found = _index.find(socket);
|
||||
if (_index.end() == found)
|
||||
{
|
||||
throw exception("this socket is not represented within this poller");
|
||||
}
|
||||
|
||||
_items[found->second].events = event;
|
||||
}
|
||||
|
||||
void poller::check_for(int const descriptor, short const event)
|
||||
{
|
||||
auto found = _fdindex.find(descriptor);
|
||||
if (_fdindex.end() == found)
|
||||
{
|
||||
throw exception("this socket is not represented within this poller");
|
||||
}
|
||||
|
||||
_items[found->second].events = event;
|
||||
}
|
||||
|
||||
bool poller::poll(long timeout /* = WAIT_FOREVER */)
|
||||
{
|
||||
int result = zmq_poll(_items.data(), _items.size(), timeout);
|
||||
if (result < 0)
|
||||
{
|
||||
if(EINTR == zmq_errno())
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
|
||||
return (result > 0);
|
||||
}
|
||||
|
||||
short poller::events(socket const& socket) const
|
||||
{
|
||||
auto found = _index.find(socket);
|
||||
if (_index.end() == found)
|
||||
{
|
||||
throw exception("this socket is not represented within this poller");
|
||||
}
|
||||
|
||||
return _items[found->second].revents;
|
||||
}
|
||||
|
||||
short poller::events(int const descriptor) const
|
||||
{
|
||||
auto found = _fdindex.find(descriptor);
|
||||
if (_fdindex.end() == found)
|
||||
{
|
||||
throw exception("this file descriptor is not represented within this poller");
|
||||
}
|
||||
|
||||
return _items[found->second].revents;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,186 @@
|
||||
/**
|
||||
* \file
|
||||
*
|
||||
* \date 9 Aug 2011
|
||||
* \author Ben Gray (\@benjamg)
|
||||
*/
|
||||
|
||||
#ifndef ZMQPP_POLLER_HPP_
|
||||
#define ZMQPP_POLLER_HPP_
|
||||
|
||||
#include <unordered_map>
|
||||
#include <vector>
|
||||
|
||||
#include "compatibility.hpp"
|
||||
|
||||
namespace zmqpp
|
||||
{
|
||||
|
||||
class socket;
|
||||
typedef socket socket_t;
|
||||
|
||||
/*!
|
||||
* Polling wrapper.
|
||||
*
|
||||
* Allows access to polling for any number of sockets or file descriptors.
|
||||
*/
|
||||
class poller
|
||||
{
|
||||
public:
|
||||
static const long wait_forever; /*!< Block forever flag, default setting. */
|
||||
|
||||
static const short poll_none; /*!< No polling flags set. */
|
||||
static const short poll_in; /*!< Monitor inbound flag. */
|
||||
static const short poll_out; /*!< Monitor output flag. */
|
||||
static const short poll_error; /*!< Monitor error flag.\n Only for file descriptors. */
|
||||
|
||||
/*!
|
||||
* Construct an empty polling model.
|
||||
*/
|
||||
poller();
|
||||
|
||||
/*!
|
||||
* Cleanup poller.
|
||||
*
|
||||
* Any sockets will need to be closed separately.
|
||||
*/
|
||||
~poller();
|
||||
|
||||
/*!
|
||||
* Add a socket to the polling model and set which events to monitor.
|
||||
*
|
||||
* \param socket the socket to monitor.
|
||||
* \param event the event flags to monitor on the socket.
|
||||
*/
|
||||
void add(socket_t& socket, short const event = poll_in);
|
||||
|
||||
/*!
|
||||
* Add a file descriptor to the polling model and set which events to monitor.
|
||||
*
|
||||
* \param descriptor the file descriptor to monitor.
|
||||
* \param event the event flags to monitor.
|
||||
*/
|
||||
void add(int const descriptor, short const event = poll_in | poll_error);
|
||||
|
||||
/*!
|
||||
* Check if we are monitoring a given socket with this poller.
|
||||
*
|
||||
* \param socket the socket to check.
|
||||
* \return true if it is there.
|
||||
*/
|
||||
bool has(socket_t const& socket);
|
||||
|
||||
/*!
|
||||
* Check if we are monitoring a given file descriptor with this poller.
|
||||
*
|
||||
* \param descriptor the file descriptor to check.
|
||||
* \return true if it is there.
|
||||
*/
|
||||
bool has(int const descriptor);
|
||||
|
||||
/*!
|
||||
* Stop monitoring a socket.
|
||||
*
|
||||
* \param socket the socket to stop monitoring.
|
||||
*/
|
||||
void remove(socket_t const& socket);
|
||||
|
||||
/*!
|
||||
* Stop monitoring a file descriptor.
|
||||
*
|
||||
* \param descriptor the file descriptor to stop monitoring.
|
||||
*/
|
||||
void remove(int const descriptor);
|
||||
|
||||
/*!
|
||||
* Update the monitored event flags for a given socket.
|
||||
*
|
||||
* \param socket the socket to update event flags.
|
||||
* \param event the event flags to monitor on the socket.
|
||||
*/
|
||||
void check_for(socket_t const& socket, short const event);
|
||||
|
||||
/*!
|
||||
* Update the monitored event flags for a given file descriptor.
|
||||
*
|
||||
* \param descriptor the file descriptor to update event flags.
|
||||
* \param event the event flags to monitor on the socket.
|
||||
*/
|
||||
void check_for(int const descriptor, short const event);
|
||||
|
||||
/*!
|
||||
* Poll for monitored events.
|
||||
*
|
||||
* By default this method will block forever or until at least one of the monitored
|
||||
* sockets or file descriptors has events.
|
||||
*
|
||||
* If a timeout is set and was reached then this function returns false.
|
||||
*
|
||||
* \param timeout milliseconds to timeout.
|
||||
* \return true if there is an event..
|
||||
*/
|
||||
bool poll(long timeout = wait_forever);
|
||||
|
||||
/*!
|
||||
* Get the event flags triggered for a socket.
|
||||
*
|
||||
* \param socket the socket to get triggered event flags for.
|
||||
* \return the event flags.
|
||||
*/
|
||||
short events(socket_t const& socket) const;
|
||||
|
||||
/*!
|
||||
* Get the event flags triggered for a file descriptor.
|
||||
*
|
||||
* \param descriptor the file descriptor to get triggered event flags for.
|
||||
* \return the event flags.
|
||||
*/
|
||||
short events(int const descriptor) const;
|
||||
|
||||
/*!
|
||||
* Check either a file descriptor or socket for input events.
|
||||
*
|
||||
* Templated helper method that calls through to event and checks for a given flag
|
||||
*
|
||||
* \param watchable either a file descriptor or socket known to the poller.
|
||||
* \return true if there is input.
|
||||
*/
|
||||
template<typename Watched>
|
||||
bool has_input(Watched const& watchable) const { return events(watchable) & poll_in; }
|
||||
|
||||
/*!
|
||||
* Check either a file descriptor or socket for output events.
|
||||
*
|
||||
* Templated helper method that calls through to event and checks for a given flag
|
||||
*
|
||||
* \param watchable either a file descriptor or socket known to the poller.
|
||||
* \return true if there is output.
|
||||
*/
|
||||
template<typename Watched>
|
||||
bool has_output(Watched const& watchable) const { return events(watchable) & poll_out; }
|
||||
|
||||
/*!
|
||||
* Check a file descriptor.
|
||||
*
|
||||
* Templated helper method that calls through to event and checks for a given flag
|
||||
*
|
||||
* Technically this template works for sockets as well but the error flag is never set for
|
||||
* sockets so I have no idea why someone would call it.
|
||||
*
|
||||
* \param watchable a file descriptor know to the poller.
|
||||
* \return true if there is an error.
|
||||
*/
|
||||
template<typename Watched>
|
||||
bool has_error(Watched const& watchable) const { return events(watchable) & poll_error; }
|
||||
|
||||
private:
|
||||
std::vector<zmq_pollitem_t> _items;
|
||||
std::unordered_map<void *, size_t> _index;
|
||||
std::unordered_map<int, size_t> _fdindex;
|
||||
|
||||
void reindex(size_t const index);
|
||||
};
|
||||
|
||||
}
|
||||
|
||||
#endif /* ZMQPP_POLLER_HPP_ */
|
||||
@@ -0,0 +1,758 @@
|
||||
/*
|
||||
* Created on: 9 Aug 2011
|
||||
* Author: Ben Gray (@benjamg)
|
||||
*/
|
||||
|
||||
#include <array>
|
||||
#include <cassert>
|
||||
#include <cstring>
|
||||
#include <functional>
|
||||
|
||||
#include "context.hpp"
|
||||
#include "exception.hpp"
|
||||
#include "message.hpp"
|
||||
#include "socket.hpp"
|
||||
|
||||
namespace zmqpp
|
||||
{
|
||||
|
||||
const int socket::normal = 0;
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
const int socket::dont_wait = ZMQ_NOBLOCK;
|
||||
#else
|
||||
const int socket::dont_wait = ZMQ_DONTWAIT;
|
||||
#endif
|
||||
const int socket::send_more = ZMQ_SNDMORE;
|
||||
#ifdef ZMQ_EXPERIMENTAL_LABELS
|
||||
const int socket::send_label = ZMQ_SNDLABEL;
|
||||
#endif
|
||||
|
||||
const int max_socket_option_buffer_size = 256;
|
||||
const int max_stream_buffer_size = 4096;
|
||||
|
||||
socket::socket(const context& context, socket_type const type)
|
||||
: _socket(nullptr)
|
||||
, _type(type)
|
||||
, _recv_buffer()
|
||||
{
|
||||
_socket = zmq_socket(context, static_cast<int>(type));
|
||||
if(nullptr == _socket)
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
|
||||
zmq_msg_init(&_recv_buffer);
|
||||
}
|
||||
|
||||
socket::~socket()
|
||||
{
|
||||
zmq_msg_close(&_recv_buffer);
|
||||
|
||||
if (nullptr != _socket)
|
||||
{
|
||||
|
||||
#ifndef NDEBUG // unused assert variable in release
|
||||
int result = zmq_close(_socket);
|
||||
assert(0 == result);
|
||||
#else
|
||||
zmq_close(_socket);
|
||||
#endif // NDEBUG
|
||||
|
||||
_socket = nullptr;
|
||||
}
|
||||
}
|
||||
|
||||
void socket::bind(endpoint_t const& endpoint)
|
||||
{
|
||||
int result = zmq_bind(_socket, endpoint.c_str());
|
||||
|
||||
if (0 != result)
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
}
|
||||
|
||||
void socket::unbind(endpoint_t const& endpoint)
|
||||
{
|
||||
int result = zmq_unbind(_socket, endpoint.c_str());
|
||||
|
||||
if (0 != result)
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
}
|
||||
|
||||
void socket::connect(endpoint_t const& endpoint)
|
||||
{
|
||||
int result = zmq_connect(_socket, endpoint.c_str());
|
||||
|
||||
if (0 != result)
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
}
|
||||
|
||||
void socket::disconnect(endpoint_t const& endpoint)
|
||||
{
|
||||
int result = zmq_disconnect(_socket, endpoint.c_str());
|
||||
|
||||
if (0 != result)
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
}
|
||||
|
||||
void socket::close()
|
||||
{
|
||||
int result = zmq_close(_socket);
|
||||
|
||||
if (0 != result)
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
|
||||
_socket = nullptr;
|
||||
}
|
||||
|
||||
bool socket::send(message& message, bool const dont_block /* = false */)
|
||||
{
|
||||
size_t parts = message.parts();
|
||||
if (parts == 0)
|
||||
{
|
||||
throw std::invalid_argument("sending requires messages have at least one part");
|
||||
}
|
||||
|
||||
bool dont_wait = dont_block;
|
||||
for(size_t i = 0; i < parts; ++i)
|
||||
{
|
||||
int flag = socket::normal;
|
||||
if(dont_wait) { flag |= socket::dont_wait; }
|
||||
if(i < (parts - 1)) { flag |= socket::send_more; }
|
||||
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
int result = zmq_send( _socket, &message.raw_msg(i), flag );
|
||||
#elif (ZMQ_VERSION_MAJOR < 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR < 2))
|
||||
int result = zmq_sendmsg( _socket, &message.raw_msg(i), flag );
|
||||
#else
|
||||
int result = zmq_msg_send( &message.raw_msg(i), _socket, flag );
|
||||
#endif
|
||||
|
||||
if (result < 0)
|
||||
{
|
||||
// the zmq framework should not block if the first part is accepted
|
||||
// so we should only ever get this error on the first part
|
||||
if((0 == i) && (EAGAIN == zmq_errno()))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
if(EINTR == zmq_errno())
|
||||
{
|
||||
if (0 == message.parts())
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
// If we have an interrupt but it's not on the first part then we
|
||||
// know we can safely send out the rest of the message as we can
|
||||
// enforce that it won't wait on a blocking action
|
||||
dont_wait = true;
|
||||
continue;
|
||||
}
|
||||
|
||||
// sanity checking
|
||||
assert(EAGAIN != zmq_errno());
|
||||
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
|
||||
message.sent(i);
|
||||
}
|
||||
|
||||
// Leave message reference in a stable state
|
||||
zmqpp::message local;
|
||||
std::swap(local, message);
|
||||
return true;
|
||||
}
|
||||
|
||||
bool socket::receive(message& message, bool const dont_block /* = false */)
|
||||
{
|
||||
if (message.parts() > 0)
|
||||
{
|
||||
// swap and discard old message
|
||||
zmqpp::message local;
|
||||
std::swap(local, message);
|
||||
}
|
||||
|
||||
int flags = (dont_block) ? socket::dont_wait : socket::normal;
|
||||
bool more = true;
|
||||
|
||||
while(more)
|
||||
{
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
int result = zmq_recv( _socket, &_recv_buffer, flags );
|
||||
#elif (ZMQ_VERSION_MAJOR < 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR < 2))
|
||||
int result = zmq_recvmsg( _socket, &_recv_buffer, flags );
|
||||
#else
|
||||
int result = zmq_msg_recv( &_recv_buffer, _socket, flags );
|
||||
#endif
|
||||
|
||||
if(result < 0)
|
||||
{
|
||||
if ((0 == message.parts()) && (EAGAIN == zmq_errno()))
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
if(EINTR == zmq_errno())
|
||||
{
|
||||
if (0 == message.parts())
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
// If we have an interrupt but it's not on the first part then we
|
||||
// know we can safely pull out the rest of the message as it will
|
||||
// not be blocking
|
||||
continue;
|
||||
}
|
||||
|
||||
assert(EAGAIN != zmq_errno());
|
||||
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
|
||||
zmq_msg_t& dest = message.raw_new_msg();
|
||||
zmq_msg_move(&dest, &_recv_buffer);
|
||||
|
||||
get(socket_option::receive_more, more);
|
||||
}
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
|
||||
bool socket::send(std::string const& string, int const flags /* = NORMAL */)
|
||||
{
|
||||
return send_raw(string.data(), string.size(), flags);
|
||||
}
|
||||
|
||||
bool socket::receive(std::string& string, int const flags /* = NORMAL */)
|
||||
{
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
int result = zmq_recv( _socket, &_recv_buffer, flags );
|
||||
#elif (ZMQ_VERSION_MAJOR < 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR < 2))
|
||||
int result = zmq_recvmsg( _socket, &_recv_buffer, flags );
|
||||
#else
|
||||
int result = zmq_msg_recv( &_recv_buffer, _socket, flags );
|
||||
#endif
|
||||
|
||||
if(result >= 0)
|
||||
{
|
||||
string.reserve(zmq_msg_size(&_recv_buffer));
|
||||
string.assign(static_cast<char*>(zmq_msg_data(&_recv_buffer)), zmq_msg_size(&_recv_buffer));
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
if (EAGAIN == zmq_errno() || EINTR == zmq_errno())
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
|
||||
|
||||
bool socket::send_raw(char const* buffer, int const length, int const flags /* = NORMAL */)
|
||||
{
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
zmq_msg_t msg;
|
||||
int result = zmq_msg_init_size(&msg, length);
|
||||
if (result != 0)
|
||||
{
|
||||
zmq_internal_exception();
|
||||
}
|
||||
|
||||
memcpy(zmq_msg_data(&msg), buffer, length);
|
||||
result = zmq_send(_socket, &msg, flags);
|
||||
#else
|
||||
int result = zmq_send(_socket, buffer, length, flags);
|
||||
#endif
|
||||
if(result >= 0)
|
||||
{
|
||||
return true;
|
||||
}
|
||||
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
// only actually need to close this on error
|
||||
zmq_msg_close(&msg);
|
||||
#endif
|
||||
|
||||
if (EAGAIN == zmq_errno() || EINTR == zmq_errno())
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
|
||||
bool socket::receive_raw(char* buffer, int& length, int const flags /* = NORMAL */)
|
||||
{
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
int result = zmq_recv( _socket, &_recv_buffer, flags );
|
||||
#elif (ZMQ_VERSION_MAJOR < 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR < 2))
|
||||
int result = zmq_recvmsg( _socket, &_recv_buffer, flags );
|
||||
#else
|
||||
int result = zmq_msg_recv( &_recv_buffer, _socket, flags );
|
||||
#endif
|
||||
|
||||
if(result >= 0)
|
||||
{
|
||||
length = zmq_msg_size(&_recv_buffer);
|
||||
memcpy(buffer, zmq_msg_data(&_recv_buffer), length);
|
||||
|
||||
return true;
|
||||
}
|
||||
|
||||
if (EAGAIN == zmq_errno() || EINTR == zmq_errno())
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
|
||||
|
||||
// Helper
|
||||
void socket::subscribe(std::string const& topic)
|
||||
{
|
||||
set(socket_option::subscribe, topic);
|
||||
}
|
||||
|
||||
void socket::unsubscribe(std::string const& topic)
|
||||
{
|
||||
set(socket_option::unsubscribe, topic);
|
||||
}
|
||||
|
||||
bool socket::has_more_parts() const
|
||||
{
|
||||
return get<bool>(socket_option::receive_more);
|
||||
}
|
||||
|
||||
|
||||
// Set socket options for different types of option
|
||||
void socket::set(socket_option const option, int const value)
|
||||
{
|
||||
switch(option)
|
||||
{
|
||||
// unsigned 64bit Integers
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
case socket_option::high_water_mark:
|
||||
case socket_option::send_buffer_size:
|
||||
case socket_option::receive_buffer_size:
|
||||
#endif
|
||||
case socket_option::affinity:
|
||||
if (value < 0) { throw exception("attempting to set an unsigned 64 bit integer option with a negative integer"); }
|
||||
set(option, static_cast<uint64_t>(value));
|
||||
break;
|
||||
|
||||
// 64bit Integers
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
case socket_option::rate:
|
||||
case socket_option::recovery_interval:
|
||||
case socket_option::recovery_interval_seconds:
|
||||
case socket_option::swap_size:
|
||||
#else
|
||||
case socket_option::max_messsage_size:
|
||||
#endif
|
||||
set(option, static_cast<int64_t>(value));
|
||||
break;
|
||||
|
||||
// Boolean
|
||||
#if (ZMQ_VERSION_MAJOR > 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR >= 1))
|
||||
case socket_option::ipv4_only:
|
||||
#endif
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
case socket_option::multicast_loopback:
|
||||
#endif
|
||||
#if (ZMQ_VERSION_MAJOR > 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR >= 2))
|
||||
#if (ZMQ_VERSION_MINOR == 2)
|
||||
case socket_option::delay_attach_on_connect:
|
||||
#else
|
||||
case socket_option::immediate:
|
||||
#endif
|
||||
case socket_option::router_mandatory:
|
||||
case socket_option::xpub_verbose:
|
||||
#endif
|
||||
if (value == 0) { set(option, false); }
|
||||
else if (value == 1) { set(option, true); }
|
||||
else { throw exception("attempting to set a boolean option with a non 0 or 1 integer"); }
|
||||
break;
|
||||
|
||||
// Default or Boolean
|
||||
#if (ZMQ_VERSION_MAJOR > 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR >= 2))
|
||||
case socket_option::tcp_keepalive:
|
||||
if (value < -1 || value > 1) { throw exception("attempting to set a default or boolean option with a non -1, 0 or 1 integer"); }
|
||||
if (0 != zmq_setsockopt(_socket, static_cast<int>(option), &value, sizeof(value)))
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
break;
|
||||
#endif
|
||||
|
||||
// Integers that require +ve numbers
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
case socket_option::reconnect_interval_max:
|
||||
#else
|
||||
case socket_option::reconnect_interval_max:
|
||||
case socket_option::send_buffer_size:
|
||||
case socket_option::recovery_interval:
|
||||
case socket_option::receive_buffer_size:
|
||||
case socket_option::send_high_water_mark:
|
||||
case socket_option::receive_high_water_mark:
|
||||
case socket_option::multicast_hops:
|
||||
case socket_option::rate:
|
||||
#endif
|
||||
case socket_option::backlog:
|
||||
if (value < 0) { throw exception("attempting to set a positive only integer option with a negative integer"); }
|
||||
// Integers
|
||||
case socket_option::reconnect_interval:
|
||||
case socket_option::linger:
|
||||
case socket_option::receive_timeout:
|
||||
case socket_option::send_timeout:
|
||||
#if (ZMQ_VERSION_MAJOR > 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR >= 2))
|
||||
case socket_option::tcp_keepalive_idle:
|
||||
case socket_option::tcp_keepalive_count:
|
||||
case socket_option::tcp_keepalive_interval:
|
||||
#endif
|
||||
if (0 != zmq_setsockopt(_socket, static_cast<int>(option), &value, sizeof(value)))
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
break;
|
||||
default:
|
||||
throw exception("attempting to set a non signed integer option with a signed integer value");
|
||||
}
|
||||
}
|
||||
|
||||
void socket::set(socket_option const option, bool const value)
|
||||
{
|
||||
switch(option)
|
||||
{
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
case socket_option::multicast_loopback:
|
||||
#endif
|
||||
#if (ZMQ_VERSION_MAJOR > 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR >= 1))
|
||||
case socket_option::ipv4_only:
|
||||
#endif
|
||||
#if (ZMQ_VERSION_MAJOR > 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR >= 2))
|
||||
#if (ZMQ_VERSION_MINOR == 2)
|
||||
case socket_option::delay_attach_on_connect:
|
||||
#else
|
||||
case socket_option::immediate:
|
||||
#endif
|
||||
case socket_option::router_mandatory:
|
||||
case socket_option::xpub_verbose:
|
||||
#endif
|
||||
{
|
||||
int ivalue = value ? 1 : 0;
|
||||
if (0 != zmq_setsockopt(_socket, static_cast<int>(option), &ivalue, sizeof(ivalue)))
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
break;
|
||||
}
|
||||
default:
|
||||
throw exception("attempting to set a non boolean option with a boolean value");
|
||||
}
|
||||
}
|
||||
|
||||
void socket::set(socket_option const option, uint64_t const value)
|
||||
{
|
||||
switch(option)
|
||||
{
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
// unsigned 64bit Integers
|
||||
case socket_option::high_water_mark:
|
||||
case socket_option::send_buffer_size:
|
||||
case socket_option::receive_buffer_size:
|
||||
#endif
|
||||
case socket_option::affinity:
|
||||
if (0 != zmq_setsockopt(_socket, static_cast<int>(option), &value, sizeof(value)))
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
break;
|
||||
default:
|
||||
throw exception("attempting to set a non unsigned 64 bit integer option with a unsigned 64 bit integer value");
|
||||
}
|
||||
}
|
||||
|
||||
void socket::set(socket_option const option, int64_t const value)
|
||||
{
|
||||
switch(option)
|
||||
{
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
case socket_option::rate:
|
||||
case socket_option::recovery_interval:
|
||||
case socket_option::recovery_interval_seconds:
|
||||
case socket_option::swap_size:
|
||||
#else
|
||||
case socket_option::max_messsage_size:
|
||||
#endif
|
||||
// zmq only allowed +ve int64_t options
|
||||
if (value < 0) { throw exception("attempting to set a positive only 64 bit integer option with a negative 64bit integer"); }
|
||||
if (0 != zmq_setsockopt(_socket, static_cast<int>(option), &value, sizeof(value)))
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
break;
|
||||
default:
|
||||
throw exception("attempting to set a non 64 bit integer option with a 64 bit integer value");
|
||||
}
|
||||
}
|
||||
|
||||
void socket::set(socket_option const option, char const* value, size_t const length)
|
||||
{
|
||||
switch(option)
|
||||
{
|
||||
case socket_option::identity:
|
||||
case socket_option::subscribe:
|
||||
case socket_option::unsubscribe:
|
||||
#if (ZMQ_VERSION_MAJOR > 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR >= 2))
|
||||
case socket_option::tcp_accept_filter:
|
||||
#endif
|
||||
if (0 != zmq_setsockopt(_socket, static_cast<int>(option), value, length))
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
break;
|
||||
default:
|
||||
throw exception("attempting to set a non string option with a string value");
|
||||
}
|
||||
}
|
||||
|
||||
// Get socket options, multiple versions for easy of use
|
||||
void socket::get(socket_option const option, int& value) const
|
||||
{
|
||||
size_t value_size = sizeof(int);
|
||||
|
||||
switch(option)
|
||||
{
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
case socket_option::receive_more:
|
||||
case socket_option::multicast_loopback:
|
||||
value = static_cast<int>(get<int64_t>(option));
|
||||
break;
|
||||
#endif
|
||||
case socket_option::type:
|
||||
case socket_option::linger:
|
||||
case socket_option::backlog:
|
||||
case socket_option::reconnect_interval:
|
||||
case socket_option::reconnect_interval_max:
|
||||
case socket_option::receive_timeout:
|
||||
case socket_option::send_timeout:
|
||||
case socket_option::file_descriptor:
|
||||
case socket_option::events:
|
||||
#if (ZMQ_VERSION_MAJOR > 2)
|
||||
case socket_option::receive_more:
|
||||
case socket_option::send_buffer_size:
|
||||
case socket_option::receive_buffer_size:
|
||||
case socket_option::rate:
|
||||
case socket_option::recovery_interval:
|
||||
case socket_option::send_high_water_mark:
|
||||
case socket_option::receive_high_water_mark:
|
||||
case socket_option::multicast_hops:
|
||||
#endif
|
||||
#if (ZMQ_VERSION_MAJOR > 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR >= 1))
|
||||
case socket_option::ipv4_only:
|
||||
#endif
|
||||
#if (ZMQ_VERSION_MAJOR > 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR >= 2))
|
||||
#if (ZMQ_VERSION_MINOR == 2)
|
||||
case socket_option::delay_attach_on_connect:
|
||||
#else
|
||||
case socket_option::immediate:
|
||||
#endif
|
||||
case socket_option::tcp_keepalive:
|
||||
case socket_option::tcp_keepalive_idle:
|
||||
case socket_option::tcp_keepalive_count:
|
||||
case socket_option::tcp_keepalive_interval:
|
||||
#endif
|
||||
#ifdef ZMQ_EXPERIMENTAL_LABELS
|
||||
case socket_option::receive_label:
|
||||
#endif
|
||||
if (0 != zmq_getsockopt(_socket, static_cast<int>(option), &value, &value_size))
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
|
||||
// sanity check
|
||||
assert(value_size <= sizeof(int));
|
||||
break;
|
||||
default:
|
||||
throw exception("attempting to get a non integer option with an integer value");
|
||||
}
|
||||
}
|
||||
|
||||
void socket::get(socket_option const option, bool& value) const
|
||||
{
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
int64_t int_value = 0;
|
||||
size_t value_size = sizeof(int64_t);
|
||||
#else
|
||||
int int_value = 0;
|
||||
size_t value_size = sizeof(int);
|
||||
#endif
|
||||
|
||||
switch(option)
|
||||
{
|
||||
case socket_option::receive_more:
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
case socket_option::multicast_loopback:
|
||||
#endif
|
||||
#if (ZMQ_VERSION_MAJOR > 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR >= 1))
|
||||
case socket_option::ipv4_only:
|
||||
#endif
|
||||
#if (ZMQ_VERSION_MAJOR > 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR >= 2))
|
||||
#if (ZMQ_VERSION_MINOR == 2)
|
||||
case socket_option::delay_attach_on_connect:
|
||||
#else
|
||||
case socket_option::immediate:
|
||||
#endif
|
||||
#endif
|
||||
#ifdef ZMQ_EXPERIMENTAL_LABELS
|
||||
case socket_option::receive_label:
|
||||
#endif
|
||||
if (0 != zmq_getsockopt(_socket, static_cast<int>(option), &int_value, &value_size))
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
|
||||
value = (int_value == 1) ? true : false;
|
||||
break;
|
||||
default:
|
||||
throw exception("attempting to get a non boolean option with a boolean value");
|
||||
}
|
||||
}
|
||||
|
||||
void socket::get(socket_option const option, uint64_t& value) const
|
||||
{
|
||||
size_t value_size = sizeof(uint64_t);
|
||||
|
||||
switch(option)
|
||||
{
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
case socket_option::high_water_mark:
|
||||
case socket_option::send_buffer_size:
|
||||
case socket_option::receive_buffer_size:
|
||||
#endif
|
||||
case socket_option::affinity:
|
||||
if(0 != zmq_getsockopt(_socket, static_cast<int>(option), &value, &value_size))
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
break;
|
||||
default:
|
||||
throw exception("attempting to get a non unsigned 64 bit integer option with an unsigned 64 bit integer value");
|
||||
}
|
||||
}
|
||||
|
||||
void socket::get(socket_option const option, int64_t& value) const
|
||||
{
|
||||
size_t value_size = sizeof(int64_t);
|
||||
|
||||
switch(option)
|
||||
{
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
case socket_option::rate:
|
||||
case socket_option::recovery_interval:
|
||||
case socket_option::recovery_interval_seconds:
|
||||
case socket_option::swap_size:
|
||||
case socket_option::receive_more:
|
||||
case socket_option::multicast_loopback:
|
||||
#else
|
||||
case socket_option::max_messsage_size:
|
||||
#endif
|
||||
if(0 != zmq_getsockopt(_socket, static_cast<int>(option), &value, &value_size))
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
break;
|
||||
default:
|
||||
throw exception("attempting to get a non 64 bit integer option with an 64 bit integer value");
|
||||
}
|
||||
}
|
||||
|
||||
void socket::get(socket_option const option, std::string& value) const
|
||||
{
|
||||
static std::array<char, max_socket_option_buffer_size> buffer;
|
||||
size_t size = max_socket_option_buffer_size;
|
||||
|
||||
switch(option)
|
||||
{
|
||||
case socket_option::identity:
|
||||
#if (ZMQ_VERSION_MAJOR > 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR >= 2))
|
||||
case socket_option::last_endpoint:
|
||||
#endif
|
||||
if(0 != zmq_getsockopt(_socket, static_cast<int>(option), buffer.data(), &size))
|
||||
{
|
||||
throw zmq_internal_exception();
|
||||
}
|
||||
|
||||
value.assign(buffer.data(), size);
|
||||
break;
|
||||
default:
|
||||
throw exception("attempting to get a non string option with a string value");
|
||||
}
|
||||
}
|
||||
|
||||
socket::socket(socket&& source) NOEXCEPT
|
||||
: _socket(source._socket)
|
||||
, _type(source._type)
|
||||
, _recv_buffer()
|
||||
{
|
||||
// we steal the zmq_msg_t from the valid socket, we only init our own because it's cheap
|
||||
// and zmq_msg_move does a valid check
|
||||
zmq_msg_init(&_recv_buffer);
|
||||
zmq_msg_move(&_recv_buffer, &source._recv_buffer);
|
||||
|
||||
// Clean up source a little, we will handle the deinit, it doesn't need to
|
||||
source._socket = nullptr;
|
||||
}
|
||||
|
||||
socket& socket::operator=(socket&& source) NOEXCEPT
|
||||
{
|
||||
std::swap(_socket, source._socket);
|
||||
|
||||
_type = source._type; // just clone?
|
||||
|
||||
// we steal the zmq_msg_t from the valid socket, we only init our own because it's cheap
|
||||
// and zmq_msg_move does a valid check
|
||||
zmq_msg_init(&_recv_buffer);
|
||||
zmq_msg_move(&_recv_buffer, &source._recv_buffer);
|
||||
|
||||
return *this;
|
||||
}
|
||||
|
||||
|
||||
socket::operator bool() const
|
||||
{
|
||||
return nullptr != _socket;
|
||||
}
|
||||
|
||||
|
||||
socket::operator void*() const
|
||||
{
|
||||
return _socket;
|
||||
}
|
||||
|
||||
void socket::track_message(message const& /* message */, uint32_t const parts, bool& should_delete)
|
||||
{
|
||||
if (parts == 0)
|
||||
{
|
||||
should_delete = true;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,500 @@
|
||||
/**
|
||||
* \file
|
||||
*
|
||||
* \date 9 Aug 2011
|
||||
* \author Ben Gray (\@benjamg)
|
||||
*/
|
||||
|
||||
#ifndef ZMQPP_SOCKET_HPP_
|
||||
#define ZMQPP_SOCKET_HPP_
|
||||
|
||||
#include <cstring>
|
||||
#include <string>
|
||||
#include <list>
|
||||
|
||||
#include <zmq.h>
|
||||
|
||||
#include "compatibility.hpp"
|
||||
|
||||
#include "socket_types.hpp"
|
||||
#include "socket_options.hpp"
|
||||
|
||||
namespace zmqpp
|
||||
{
|
||||
|
||||
class context;
|
||||
class message;
|
||||
|
||||
typedef std::string endpoint_t;
|
||||
typedef context context_t;
|
||||
typedef message message_t;
|
||||
|
||||
/*!
|
||||
* The socket class represents the zmq sockets.
|
||||
*
|
||||
* A socket can be bound and/or connected to as many endpoints as required
|
||||
* with the sole exception of a ::pair socket.
|
||||
*
|
||||
* The routing is handled by zmq based on the type set.
|
||||
*
|
||||
* The bound side of an inproc connection must occur first and inproc can only
|
||||
* connect to other inproc sockets of the same context.
|
||||
*
|
||||
* This class is c++0x move supporting and cannot be copied.
|
||||
*/
|
||||
class socket
|
||||
{
|
||||
public:
|
||||
static const int normal; /*!< /brief default send type, no flags set */
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
static const int dont_wait; /*!< /brief don't block if sending is not currently possible */
|
||||
#else
|
||||
static const int dont_wait; /*!< /brief don't block if sending is not currently possible */
|
||||
#endif
|
||||
static const int send_more; /*!< /brief more parts will follow this one */
|
||||
#ifdef ZMQ_EXPERIMENTAL_LABELS
|
||||
static const int send_label; /*!< /brief this message part is an internal zmq label */
|
||||
#endif
|
||||
|
||||
/*!
|
||||
* Create a socket for a given type.
|
||||
*
|
||||
* \param context the zmq context under which the socket will live
|
||||
* \param type a valid ::socket_type for the socket
|
||||
*/
|
||||
socket(context_t const& context, socket_type const type);
|
||||
|
||||
/*!
|
||||
* This will close any socket still open before returning
|
||||
*/
|
||||
~socket();
|
||||
|
||||
/*!
|
||||
* Get the type of the socket, this works on zmqpp types and not the zmq internal types.
|
||||
* Use the socket::get method if you wish to intergoate the zmq internal ones.
|
||||
*
|
||||
* \return the type of the socket
|
||||
*/
|
||||
socket_type type() const { return _type; }
|
||||
|
||||
/*!
|
||||
* Asynchronously binds to an endpoint.
|
||||
*
|
||||
* \param endpoint the zmq endpoint to bind to
|
||||
*/
|
||||
void bind(endpoint_t const& endpoint);
|
||||
|
||||
/*!
|
||||
* Unbinds from a previously bound endpoint.
|
||||
*
|
||||
* \param endpoint the zmq endpoint to bind to
|
||||
*/
|
||||
void unbind(endpoint_t const& endpoint);
|
||||
|
||||
/*!
|
||||
* Asynchronously connects to an endpoint.
|
||||
* If the endpoint is not inproc then zmq will happily keep trying
|
||||
* to connect until there is something there.
|
||||
*
|
||||
* Inproc sockets must have a valid target already bound before connection
|
||||
* will work.
|
||||
*
|
||||
* \param endpoint the zmq endpoint to connect to
|
||||
*/
|
||||
void connect(endpoint_t const& endpoint);
|
||||
|
||||
/*!
|
||||
* Asynchronously connects to multiple endpoints.
|
||||
* If the endpoint is not inproc then zmq will happily keep trying
|
||||
* to connect until there is something there.
|
||||
*
|
||||
* Inproc sockets must have a valid target already bound before connection
|
||||
* will work.
|
||||
*
|
||||
* This is a helper function that wraps the single item connect in a loop
|
||||
*
|
||||
* \param connections_begin the starting iterator for zmq endpoints.
|
||||
* \param connections_end the final iterator for zmq endpoints.
|
||||
*/
|
||||
template<typename InputIterator>
|
||||
void connect(InputIterator const& connections_begin, InputIterator const& connections_end)
|
||||
{
|
||||
for(InputIterator it = connections_begin; it != connections_end; ++it)
|
||||
{
|
||||
connect(*it);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
/*!
|
||||
* Disconnects a previously connected endpoint.
|
||||
*
|
||||
* \param endpoint the zmq endpoint to disconnect from
|
||||
*/
|
||||
void disconnect(endpoint_t const& endpoint);
|
||||
|
||||
/*!
|
||||
* Disconnects from multiple previously connected endpoints.
|
||||
*
|
||||
* This is a helper function that wraps the single item disconnect in a loop
|
||||
*
|
||||
* \param disconnections_begin the starting iterator for zmq endpoints.
|
||||
* \param disconnections_end the final iterator for zmq endpoints.
|
||||
*/
|
||||
template<typename InputIterator>
|
||||
void disconnect(InputIterator const& disconnections_begin, InputIterator const& disconnections_end)
|
||||
{
|
||||
for(InputIterator it = disconnections_begin; it != disconnections_end; ++it)
|
||||
{
|
||||
disconnect(*it);
|
||||
}
|
||||
}
|
||||
|
||||
/*!
|
||||
* Closes the internal zmq socket and marks this instance
|
||||
* as invalid.
|
||||
*/
|
||||
void close();
|
||||
|
||||
/*!
|
||||
* Sends the message over the connection, this may be a multipart message.
|
||||
*
|
||||
* If dont_block is true and we are unable to add a new message then this
|
||||
* function will return false.
|
||||
*
|
||||
* \param message message to send
|
||||
* \param dont_block boolean to dictate if we wait while sending.
|
||||
* \return true if message sent, false if it would have blocked
|
||||
*/
|
||||
bool send(message_t& message, bool const dont_block = false);
|
||||
|
||||
/*!
|
||||
* Gets a message from the connection, this may be a multipart message.
|
||||
*
|
||||
* If dont_block is true and we are unable to get a message then this
|
||||
* function will return false.
|
||||
*
|
||||
* \param message reference to fill with received data
|
||||
* \param dont_block boolean to dictate if we wait for data.
|
||||
* \return true if message sent, false if it would have blocked
|
||||
*/
|
||||
bool receive(message_t& message, bool const dont_block = false);
|
||||
|
||||
/*!
|
||||
* Sends the byte data held by the string as the next message part.
|
||||
*
|
||||
* If the socket::DONT_WAIT flag and we are unable to add a new message to
|
||||
* socket then this function will return false.
|
||||
*
|
||||
* \param string message part to send
|
||||
* \param flags message send flags
|
||||
* \return true if message part sent, false if it would have blocked
|
||||
*/
|
||||
bool send(std::string const& string, int const flags = normal);
|
||||
|
||||
/*!
|
||||
* If there is a message ready then get the next part as a string
|
||||
*
|
||||
* If the socket::DONT_WAIT flag and there is no message ready to receive
|
||||
* then this function will return false.
|
||||
*
|
||||
* \param string message part to receive into
|
||||
* \param flags message receive flags
|
||||
* \return true if message part received, false if it would have blocked
|
||||
*/
|
||||
bool receive(std::string& string, int const flags = normal);
|
||||
|
||||
/*!
|
||||
* Sends the byte data pointed to by buffer as the next part of the message.
|
||||
*
|
||||
* If the socket::DONT_WAIT flag and we are unable to add a new message to
|
||||
* socket then this function will return false.
|
||||
*
|
||||
* \param buffer byte buffer pointer to start writing from
|
||||
* \param length max length of the buffer
|
||||
* \param flags message send flags
|
||||
* \return true if message part sent, false if it would have blocked
|
||||
*/
|
||||
bool send_raw(char const* buffer, int const length, int const flags = normal);
|
||||
|
||||
/*!
|
||||
* \warning If the buffer is not large enough for the message part then the
|
||||
* data will be truncated. The rest of the part is lost forever.
|
||||
*
|
||||
* If there is a message ready then get the next part of it as a raw
|
||||
* byte buffer.
|
||||
*
|
||||
* If the socket::DONT_WAIT flag and there is no message ready to receive
|
||||
* then this function will return false.
|
||||
*
|
||||
* \param buffer byte buffer pointer to start writing to
|
||||
* \param length max length of the buffer
|
||||
* \param flags message receive flags
|
||||
* \return true if message part received, false if it would have blocked
|
||||
*/
|
||||
bool receive_raw(char* buffer, int& length, int const flags = normal);
|
||||
|
||||
/*!
|
||||
*
|
||||
* Subscribe to a topic
|
||||
*
|
||||
* Helper function that is equivalent of calling
|
||||
* \code
|
||||
* set(zmqpp::socket_option::subscribe, topic);
|
||||
* \endcode
|
||||
*
|
||||
* This method is only useful for subscribe type sockets.
|
||||
*
|
||||
* \param topic the topic to subscribe to.
|
||||
*/
|
||||
void subscribe(std::string const& topic);
|
||||
|
||||
/*!
|
||||
* Subscribe to a topic
|
||||
*
|
||||
* Helper function that is equivalent of a loop calling
|
||||
* \code
|
||||
* set(zmqpp::socket_option::subscribe, topic);
|
||||
* \endcode
|
||||
*
|
||||
* This method is only useful for subscribe type sockets.
|
||||
*
|
||||
* Generally this will be used with stl collections using begin() and
|
||||
* end() functions to get the iterators.
|
||||
* For this reason the end loop runs until the end iterator, not inclusive
|
||||
* of it.
|
||||
*
|
||||
* \param topics_begin the starting iterator for topics.
|
||||
* \param topics_end the final iterator for topics.
|
||||
*/
|
||||
template<typename InputIterator>
|
||||
void subscribe(InputIterator const& topics_begin, InputIterator const& topics_end)
|
||||
{
|
||||
for(InputIterator it = topics_begin; it != topics_end; ++it)
|
||||
{
|
||||
subscribe(*it);
|
||||
}
|
||||
}
|
||||
|
||||
/*!
|
||||
* Unsubscribe from a topic
|
||||
*
|
||||
* Helper function that is equivalent of calling
|
||||
* \code
|
||||
* set(zmqpp::socket_option::unsubscribe, topic);
|
||||
* \endcode
|
||||
*
|
||||
* This method is only useful for subscribe type sockets.
|
||||
*
|
||||
* \param topic the topic to unsubscribe from.
|
||||
*/
|
||||
void unsubscribe(std::string const& topic);
|
||||
|
||||
/*!
|
||||
* Unsubscribe from a topic
|
||||
*
|
||||
* Helper function that is equivalent of a loop calling
|
||||
* \code
|
||||
* set(zmqpp::socket_option::unsubscribe, topic);
|
||||
* \endcode
|
||||
*
|
||||
* This method is only useful for subscribe type sockets.
|
||||
*
|
||||
* Generally this will be used with stl collections using begin() and
|
||||
* end() functions to get the iterators.
|
||||
* For this reason the end loop runs until the end iterator, not inclusive
|
||||
* of it.
|
||||
*
|
||||
* \param topics_begin the starting iterator for topics.
|
||||
* \param topics_end the final iterator for topics.
|
||||
*/
|
||||
template<typename InputIterator>
|
||||
void unsubscribe(InputIterator const& topics_begin, InputIterator const& topics_end)
|
||||
{
|
||||
for(InputIterator it = topics_begin; it != topics_end; ++it)
|
||||
{
|
||||
unsubscribe(*it);
|
||||
}
|
||||
}
|
||||
|
||||
/*!
|
||||
* If the last receive part call to the socket resulted
|
||||
* in a label or a non-terminating part of a multipart
|
||||
* message this will return true.
|
||||
*
|
||||
* \return true if there are more parts
|
||||
*/
|
||||
bool has_more_parts() const;
|
||||
|
||||
/*!
|
||||
* Set the value of an option in the underlaying zmq socket.
|
||||
*
|
||||
* \param option a valid ::socket_option
|
||||
* \param value to set the option to
|
||||
*/
|
||||
void set(socket_option const option, int const value);
|
||||
|
||||
/*!
|
||||
* Set the value of an option in the underlaying zmq socket.
|
||||
*
|
||||
* \since 2.0.0 (built against 0mq version 3.1.x or later)
|
||||
*
|
||||
* \param option a valid ::socket_option
|
||||
* \param value to set the option to
|
||||
*/
|
||||
void set(socket_option const option, bool const value);
|
||||
|
||||
/*!
|
||||
* Set the value of an option in the underlaying zmq socket.
|
||||
*
|
||||
* \param option a valid ::socket_option
|
||||
* \param value to set the option to
|
||||
*/
|
||||
void set(socket_option const option, uint64_t const value);
|
||||
|
||||
/*!
|
||||
* Set the value of an option in the underlaying zmq socket.
|
||||
*
|
||||
* \param option a valid ::socket_option
|
||||
* \param value to set the option to
|
||||
*/
|
||||
void set(socket_option const option, int64_t const value);
|
||||
|
||||
/*!
|
||||
* Set the value of an option in the underlaying zmq socket.
|
||||
*
|
||||
* \param option a valid ::socket_option
|
||||
* \param pointer to raw byte value to set the option to
|
||||
* \param length the size of the raw byte value
|
||||
*/
|
||||
void set(socket_option const option, char const* value, size_t const length);
|
||||
|
||||
/*!
|
||||
* Set the value of an option in the underlaying zmq socket.
|
||||
*
|
||||
* \param option a valid ::socket_option
|
||||
* \param pointer to null terminated cstring value to set the option to
|
||||
*/
|
||||
inline void set(socket_option const option, char const* value) { set(option, value, strlen(value)); }
|
||||
|
||||
/*!
|
||||
* Set the value of an option in the underlaying zmq socket.
|
||||
*
|
||||
* \param option a valid ::socket_option
|
||||
* \param value to set the option to
|
||||
*/
|
||||
inline void set(socket_option const option, std::string const value) { set(option, value.c_str(), value.length()); }
|
||||
|
||||
/*!
|
||||
* Get a socket option from the underlaying zmq socket.
|
||||
*
|
||||
* \param option a valid ::socket_option
|
||||
* \param value referenced int to return value in
|
||||
*/
|
||||
void get(socket_option const option, int& value) const;
|
||||
|
||||
/*!
|
||||
* Get a socket option from the underlaying zmq socket.
|
||||
*
|
||||
* \param option a valid ::socket_option
|
||||
* \param value referenced bool to return value in
|
||||
*/
|
||||
void get(socket_option const option, bool& value) const;
|
||||
|
||||
/*!
|
||||
* Get a socket option from the underlaying zmq socket.
|
||||
*
|
||||
* \param option a valid ::socket_option
|
||||
* \param value referenced uint64_t to return value in
|
||||
*/
|
||||
void get(socket_option const option, uint64_t& value) const;
|
||||
|
||||
/*!
|
||||
* Get a socket option from the underlaying zmq socket.
|
||||
*
|
||||
* \param option a valid ::socket_option
|
||||
* \param value referenced uint64_t to return value in
|
||||
*/
|
||||
void get(socket_option const option, int64_t& value) const;
|
||||
|
||||
/*!
|
||||
* Get a socket option from the underlaying zmq socket.
|
||||
*
|
||||
* \param option a valid ::socket_option
|
||||
* \param value referenced std::string to return value in
|
||||
*/
|
||||
void get(socket_option const option, std::string& value) const;
|
||||
|
||||
/*!
|
||||
* For those that don't want to get into a referenced value this templated method
|
||||
* will return the value instead.
|
||||
*
|
||||
* \param option a valid ::socket_option
|
||||
* \return socket option value
|
||||
*/
|
||||
template<typename Type>
|
||||
Type get(socket_option const option) const
|
||||
{
|
||||
Type value = Type();
|
||||
get(option, value);
|
||||
return value;
|
||||
}
|
||||
|
||||
/*!
|
||||
* Move constructor
|
||||
*
|
||||
* Moves the internals of source to this object, there is no guarantee
|
||||
* that source will be left in a valid state.
|
||||
*
|
||||
* This constructor is noexcept and so will not throw exceptions
|
||||
*
|
||||
* \param source target socket to steal internals from
|
||||
*/
|
||||
socket(socket&& source) NOEXCEPT;
|
||||
|
||||
/*!
|
||||
* Move operator
|
||||
*
|
||||
* Moves the internals of source to this object, there is no guarantee
|
||||
* that source will be left in a valid state.
|
||||
*
|
||||
* This function is noexcept and so will not throw exceptions
|
||||
*
|
||||
* \param source target socket to steal internals from
|
||||
* \return socket reference to this
|
||||
*/
|
||||
socket& operator=(socket&& source) NOEXCEPT;
|
||||
|
||||
/*!
|
||||
* Check the socket is still valid
|
||||
*
|
||||
* This tests the internal state of the socket.
|
||||
* If creation failed for some reason or if the move functions were used
|
||||
* to move the socket internals to another instance this will return false.
|
||||
*
|
||||
* \return true if the socket is valid
|
||||
*/
|
||||
operator bool() const;
|
||||
|
||||
/*!
|
||||
* Access to the raw 0mq context
|
||||
*
|
||||
* \return void pointer to the underlying 0mq socket
|
||||
*/
|
||||
operator void*() const;
|
||||
|
||||
private:
|
||||
void* _socket;
|
||||
socket_type _type;
|
||||
zmq_msg_t _recv_buffer;
|
||||
|
||||
// No copy
|
||||
socket(socket const&) NOEXCEPT ZMQPP_EXPLICITLY_DELETED;
|
||||
socket& operator=(socket const&) NOEXCEPT ZMQPP_EXPLICITLY_DELETED;
|
||||
|
||||
void track_message(message_t const&, uint32_t const, bool&);
|
||||
};
|
||||
|
||||
}
|
||||
|
||||
#endif /* ZMQPP_SOCKET_HPP_ */
|
||||
@@ -0,0 +1,80 @@
|
||||
/**
|
||||
* \file
|
||||
*
|
||||
* \date 23 Sep 2011
|
||||
* \author Ben Gray (\@benjamg)
|
||||
*/
|
||||
|
||||
#ifndef ZMQPP_SOCKET_OPTIONS_HPP_
|
||||
#define ZMQPP_SOCKET_OPTIONS_HPP_
|
||||
|
||||
namespace zmqpp
|
||||
{
|
||||
|
||||
/** \todo Expand the information on the options to make it actually useful. */
|
||||
/*!
|
||||
* \brief possible Socket options in zmq
|
||||
*/
|
||||
|
||||
enum class socket_option {
|
||||
affinity = ZMQ_AFFINITY, /*!< I/O thread affinity */
|
||||
identity = ZMQ_IDENTITY, /*!< Socket identity */
|
||||
subscribe = ZMQ_SUBSCRIBE, /*!< Add topic subscription - set only */
|
||||
unsubscribe = ZMQ_UNSUBSCRIBE, /*!< Remove topic subscription - set only */
|
||||
rate = ZMQ_RATE, /*!< Multicast data rate */
|
||||
send_buffer_size = ZMQ_SNDBUF, /*!< Kernel transmission buffer size */
|
||||
receive_buffer_size = ZMQ_RCVBUF, /*!< Kernel receive buffer size */
|
||||
receive_more = ZMQ_RCVMORE, /*!< Can receive more parts - get only */
|
||||
file_descriptor = ZMQ_FD, /*!< Socket file descriptor - get only */
|
||||
events = ZMQ_EVENTS, /*!< Socket event flags - get only */
|
||||
type = ZMQ_TYPE, /*!< Socket type - get only */
|
||||
linger = ZMQ_LINGER, /*!< Socket linger timeout */
|
||||
backlog = ZMQ_BACKLOG, /*!< Maximum length of outstanding connections - get only */
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
// Note that this is inverse of the zmq names for version 2.x
|
||||
recovery_interval_seconds = ZMQ_RECOVERY_IVL, /*!< Multicast recovery interval in seconds */
|
||||
recovery_interval = ZMQ_RECOVERY_IVL_MSEC, /*!< Multicast recovery interval in milliseconds */
|
||||
#else
|
||||
recovery_interval = ZMQ_RECOVERY_IVL, /*!< Multicast recovery interval in milliseconds */
|
||||
#endif
|
||||
reconnect_interval = ZMQ_RECONNECT_IVL, /*!< Reconnection interval */
|
||||
reconnect_interval_max = ZMQ_RECONNECT_IVL_MAX, /*!< Maximum reconnection interval */
|
||||
receive_timeout = ZMQ_RCVTIMEO, /*!< Maximum inbound block timeout */
|
||||
send_timeout = ZMQ_SNDTIMEO, /*!< Maximum outbound block timeout */
|
||||
#if (ZMQ_VERSION_MAJOR == 2)
|
||||
high_water_mark = ZMQ_HWM, /*!< High-water mark for all messages */
|
||||
swap_size = ZMQ_SWAP, /*!< Maximum socket swap size in bytes */
|
||||
multicast_loopback = ZMQ_MCAST_LOOP, /*!< Allow multicast packet loopback */
|
||||
#else
|
||||
max_messsage_size = ZMQ_MAXMSGSIZE, /*!< Maximum inbound message size */
|
||||
send_high_water_mark = ZMQ_SNDHWM, /*!< High-water mark for outbound messages */
|
||||
receive_high_water_mark = ZMQ_RCVHWM, /*!< High-water mark for inbound messages */
|
||||
multicast_hops = ZMQ_MULTICAST_HOPS, /*!< Maximum number of multicast hops */
|
||||
#endif
|
||||
#if (ZMQ_VERSION_MAJOR > 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR >= 1))
|
||||
ipv4_only = ZMQ_IPV4ONLY,
|
||||
#endif
|
||||
#if (ZMQ_VERSION_MAJOR > 3) || ((ZMQ_VERSION_MAJOR == 3) && (ZMQ_VERSION_MINOR >= 2))
|
||||
#if (ZMQ_VERSION_MINOR == 2)
|
||||
delay_attach_on_connect = ZMQ_DELAY_ATTACH_ON_CONNECT, /*!< Delay buffer attachment until connect complete */
|
||||
#else
|
||||
// ZMQ_DELAY_ATTACH_ON_CONNECT is renamed in ZeroMQ starting 3.3.x
|
||||
immediate = ZMQ_IMMEDIATE, /*!< Block message sending until connect complete */
|
||||
#endif
|
||||
last_endpoint = ZMQ_LAST_ENDPOINT, /*!< Last bound endpoint - get only */
|
||||
router_mandatory = ZMQ_ROUTER_MANDATORY, /*!< Require routable messages - set only */
|
||||
xpub_verbose = ZMQ_XPUB_VERBOSE, /*!< Pass on existing subscriptions - set only */
|
||||
tcp_keepalive = ZMQ_TCP_KEEPALIVE, /*!< Enable TCP keepalives */
|
||||
tcp_keepalive_idle = ZMQ_TCP_KEEPALIVE_IDLE, /*!< TCP keepalive idle count (generally retry count) */
|
||||
tcp_keepalive_count = ZMQ_TCP_KEEPALIVE_CNT, /*!< TCP keepalive retry count */
|
||||
tcp_keepalive_interval = ZMQ_TCP_KEEPALIVE_INTVL, /*!< TCP keepalive interval */
|
||||
tcp_accept_filter = ZMQ_TCP_ACCEPT_FILTER, /*!< Filter inbound connections - set only */
|
||||
#endif
|
||||
#ifdef ZMQ_EXPERIMENTAL_LABELS
|
||||
receive_label = ZMQ_RCVLABEL, /*!< Received label part - get only */
|
||||
#endif
|
||||
};
|
||||
|
||||
}
|
||||
|
||||
#endif /* ZMQPP_SOCKET_OPTIONS_HPP_ */
|
||||
@@ -0,0 +1,148 @@
|
||||
/**
|
||||
* \file
|
||||
*
|
||||
* \date 23 Sep 2011
|
||||
* \author Ben Gray (\@benjamg)
|
||||
*/
|
||||
|
||||
#ifndef ZMQPP_SOCKET_TYPES_HPP_
|
||||
#define ZMQPP_SOCKET_TYPES_HPP_
|
||||
|
||||
namespace zmqpp
|
||||
{
|
||||
|
||||
/*!
|
||||
* \brief Socket types allowed by zmq
|
||||
*
|
||||
* The socket type choose at creation must be one of these types.
|
||||
*
|
||||
* Each is designed for a different use and has different limitations.
|
||||
*/
|
||||
enum class socket_type {
|
||||
/*!
|
||||
* One to one - two way connection.\n
|
||||
* Connect to ::pair.\n
|
||||
* A \c pair socket has to be connected only one other pair socket and allows
|
||||
* two way communication between them.
|
||||
*/
|
||||
pair = ZMQ_PAIR,
|
||||
|
||||
/*!
|
||||
* One to many - fan out.\n
|
||||
* Connect to ::subscribe or ::xsubscribe.\n
|
||||
* Socket is send only.\n
|
||||
* Socket will drop messages and not block.\n
|
||||
* \c publish sockets allow sending of the same message to many subscribers
|
||||
* each subscriber can limit what is sent through the socket_option::subscribe
|
||||
* settings.
|
||||
*/
|
||||
publish = ZMQ_PUB,
|
||||
|
||||
/*!
|
||||
* \note It seems doxygen can't work out which data is for the socket type and
|
||||
* which is for the socket option so both get listed for both.
|
||||
*
|
||||
* One to many - fair-queued.\n
|
||||
* Connect to ::publish or ::xpublish.\n
|
||||
* Socket is receive only.\n
|
||||
* The \c subscribe socket can connection to any number of publishers and will
|
||||
* fairly pull messages from each. The socket_option::subscribe settings can
|
||||
* be use to limit which messages are received and by default none are.
|
||||
*/
|
||||
subscribe = ZMQ_SUB,
|
||||
|
||||
/*!
|
||||
* One to many - fair-queued.\n
|
||||
* Connect to ::push.\n
|
||||
* Socket is receive only.\n
|
||||
* The \c pull socket fairly pulls messages from all pushers it is connected
|
||||
* to.
|
||||
*/
|
||||
pull = ZMQ_PULL,
|
||||
|
||||
/*!
|
||||
* One to many - load-balanced.\n
|
||||
* Connect to ::pull.\n
|
||||
* Socket is send only.\n
|
||||
* Socket will block if unable to send.\n
|
||||
* The \c push socket fairly distributes messages between any connected
|
||||
* puller sockets.
|
||||
*/
|
||||
push = ZMQ_PUSH,
|
||||
|
||||
/*!
|
||||
* One to many - fair-queued outgoing, last peer incoming.\n
|
||||
* Connect to ::reply or ::xreply.\n
|
||||
* Socket flips between send and receive only.\n
|
||||
* Socket will block if unable to send.\n
|
||||
* The \c request socket will fairly balance requests sent out to a
|
||||
* replier and then can only be used to receive until that replier
|
||||
* sends a reply.
|
||||
*/
|
||||
request = ZMQ_REQ,
|
||||
|
||||
/*!
|
||||
* One to many - load-balanced incoming, last peer outgoing.\n
|
||||
* Connect to ::request or ::xrequest.\n
|
||||
* Socket flips between send and receive only.\n
|
||||
* Socket will drop messages and not block.\n
|
||||
* The \c reply socket can only receive until it pulls a message from a
|
||||
* requester at which point it can only send until the reply is sent.
|
||||
*/
|
||||
reply = ZMQ_REP,
|
||||
|
||||
/*!
|
||||
* One to many - fan out.\n
|
||||
* Connect to ::subscribe or ::xsubscribe.\n
|
||||
* Socket is send only with the exception of special subscription messages.\n
|
||||
* Socket will drop messages and not block.\n
|
||||
* \c xpublish act the same as ::publish sockets however also allow special
|
||||
* subscription messages to be received from subscribers.
|
||||
*/
|
||||
xpublish = ZMQ_XPUB,
|
||||
|
||||
/*!
|
||||
* One to many - fair-queued.\n
|
||||
* Connect to ::publish or ::xpublish.\n
|
||||
* Socket is receive only with the exception of special subscription messages\n
|
||||
* \c xsubscribe act the same as ::subscribe sockets however also allow special
|
||||
* subscription messages to be send to connected publishers.
|
||||
*/
|
||||
xsubscribe = ZMQ_XSUB,
|
||||
|
||||
/*!
|
||||
* One to many - fair-queued incoming, load-balanced outgoing.\n
|
||||
* Connect to ::reply or ::xreply.\n
|
||||
* Socket will block if unable to send.\n
|
||||
* An \c xrequest socket balances requests between repliers and pulls replies
|
||||
* back in a fair manner. Each request is expected to have exactly one reply.
|
||||
*/
|
||||
xrequest = ZMQ_XREQ,
|
||||
|
||||
/*!
|
||||
* One to many - fair-queued incoming, targeted outgoing.\n
|
||||
* Connect to ::request or ::xrequest.\n
|
||||
* Socket will drop messages and not block.\n
|
||||
* An \c xreply socket fairly pulls in requests from requesters and will
|
||||
* label requests so it can return replies back to the correct target.
|
||||
*/
|
||||
xreply = ZMQ_XREP,
|
||||
|
||||
// To match for people who prefer the shorter versions
|
||||
pub = ZMQ_PUB, /*!< version of ::publish to match zmq name convention */
|
||||
sub = ZMQ_SUB, /*!< version of ::subscribe to match zmq name convention */
|
||||
req = ZMQ_REQ, /*!< version of ::request to match zmq name convention */
|
||||
rep = ZMQ_REP, /*!< version of ::reply to match zmq name convention */
|
||||
xpub = ZMQ_XPUB, /*!< version of ::xpublish to match zmq name convention */
|
||||
xsub = ZMQ_XSUB, /*!< version of ::xsubscribe to match zmq name convention */
|
||||
xreq = ZMQ_XREQ, /*!< version of ::xrequest to match zmq name convention */
|
||||
xrep = ZMQ_XREP, /*!< version of ::xreply to match zmq name convention */
|
||||
|
||||
// For completion
|
||||
router = ZMQ_ROUTER, /*!< \deprecated Matches zmq 2.x xrep functionality. */
|
||||
dealer = ZMQ_DEALER /*!< \deprecated Matches zmq 2.x xreq functionality. */
|
||||
};
|
||||
|
||||
}
|
||||
|
||||
#endif /* ZMQPP_SOCKET_TYPES_HPP_ */
|
||||
@@ -0,0 +1,30 @@
|
||||
/*
|
||||
* Created on: 18 Aug 2011
|
||||
* Author: Ben Gray (@benjamg)
|
||||
*/
|
||||
|
||||
#include "zmqpp.hpp"
|
||||
|
||||
namespace zmqpp
|
||||
{
|
||||
|
||||
std::string version()
|
||||
{
|
||||
return BUILD_VERSION;
|
||||
}
|
||||
|
||||
void version(uint8_t& major, uint8_t& minor, uint8_t& revision)
|
||||
{
|
||||
major = ZMQPP_VERSION_MAJOR;
|
||||
minor = ZMQPP_VERSION_MINOR;
|
||||
revision = ZMQPP_VERSION_REVISION;
|
||||
}
|
||||
|
||||
void zmq_version(uint8_t& major, uint8_t& minor, uint8_t& patch)
|
||||
{
|
||||
major = ZMQ_VERSION_MAJOR;
|
||||
minor = ZMQ_VERSION_MINOR;
|
||||
patch = ZMQ_VERSION_PATCH;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,111 @@
|
||||
/**
|
||||
* \file
|
||||
*
|
||||
* \date 9 Aug 2011
|
||||
* \author Ben Gray (\@benjamg)
|
||||
*
|
||||
* License: http://www.opensource.org/licenses/MIT
|
||||
*
|
||||
* Copyright (C) 2011 by Ben Gray
|
||||
*
|
||||
* Permission is hereby granted, free of charge, to any person obtaining a copy
|
||||
* of this software and associated documentation files (the "Software"), to deal
|
||||
* in the Software without restriction, including without limitation the rights
|
||||
* to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
||||
* copies of the Software, and to permit persons to whom the Software is
|
||||
* furnished to do so, subject to the following conditions:
|
||||
*
|
||||
* The above copyright notice and this permission notice shall be included in
|
||||
* all copies or substantial portions of the Software.
|
||||
*
|
||||
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
||||
* IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
||||
* FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
||||
* AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
||||
* LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
||||
* OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
||||
* THE SOFTWARE.
|
||||
*/
|
||||
|
||||
#ifndef ZMQPP_ZMQPP_HPP_
|
||||
#define ZMQPP_ZMQPP_HPP_
|
||||
|
||||
/**
|
||||
* \def ZMQPP_VERSION_MAJOR
|
||||
* zmqpp major version number, generated at compile time
|
||||
*/
|
||||
#define ZMQPP_VERSION_MAJOR BUILD_VERSION_MAJOR
|
||||
|
||||
/**
|
||||
* \def ZMQPP_VERSION_MINOR
|
||||
* zmqpp minor version number, generated at compile time
|
||||
*/
|
||||
#define ZMQPP_VERSION_MINOR BUILD_VERSION_MINOR
|
||||
|
||||
/**
|
||||
* \def ZMQPP_VERSION_REVISION
|
||||
* zmqpp version revision number, generated at compile time
|
||||
*/
|
||||
#define ZMQPP_VERSION_REVISION BUILD_VERSION_REVISION
|
||||
|
||||
#include <zmq.h>
|
||||
|
||||
#include "compatibility.hpp"
|
||||
#include "context.hpp"
|
||||
#include "exception.hpp"
|
||||
#include "message.hpp"
|
||||
#include "poller.hpp"
|
||||
#include "socket.hpp"
|
||||
|
||||
/*!
|
||||
* \brief C++ wrapper around zmq
|
||||
*
|
||||
* All zmq++ / zmqpp functions, constants and classes live within this namespace,
|
||||
*/
|
||||
namespace zmqpp
|
||||
{
|
||||
|
||||
/*!
|
||||
* Returns the current major.minor.revision version number as a string.
|
||||
*
|
||||
* \return string version number.
|
||||
*/
|
||||
std::string version();
|
||||
|
||||
/*!
|
||||
* Retrieve the parts of the zmqpp version number.
|
||||
*
|
||||
* Set the three parameters to values representing the zmqpp version number.
|
||||
* These values are generated at library compile time.
|
||||
*
|
||||
* \param major an unsigned 8 bit reference to set to the major version.
|
||||
* \param minor an unsigned 8 bit reference to set to the minor version.
|
||||
* \param revision an unsigned 8 bit reference to set the current revision.
|
||||
*/
|
||||
void version(uint8_t& major, uint8_t& minor, uint8_t& revision);
|
||||
|
||||
/*!
|
||||
* Retrieve the parts of the 0mq version this library was built against.
|
||||
*
|
||||
* Because sections of the library are optionally compiled in or ignored
|
||||
* depending on the version of 0mq it was compiled against this method is
|
||||
* provided to allow sanity checking for usage.
|
||||
*
|
||||
* Set the three parameters to values representing the 0mq version number.
|
||||
* These values are generated at library compile time.
|
||||
*
|
||||
* \param major an unsigned 8 bit reference to set to the major version.
|
||||
* \param minor an unsigned 8 bit reference to set to the minor version.
|
||||
* \param revision an unsigned 8 bit reference to set the current revision.
|
||||
*/
|
||||
void zmq_version(uint8_t& major, uint8_t& minor, uint8_t& patch);
|
||||
|
||||
typedef context context_t; /*!< \brief context type */
|
||||
typedef std::string endpoint_t; /*!< \brief endpoint type */
|
||||
typedef message message_t; /*!< \brief message type */
|
||||
typedef poller poller_t; /*!< \brief poller type */
|
||||
typedef socket socket_t; /*!< \brief socket type */
|
||||
|
||||
}
|
||||
|
||||
#endif /* ZMQPP_ZMQPP_HPP_ */
|
||||
Reference in new issue
Block a user