Re: [Boost-users] Boost-users Digest, Vol 3643, Issue 1

2 Dec 2013

      On Dec 1, 2013, at 11:00 AM, boost-users-request@lists.boost.org wrote:
...
Date: Sun, 1 Dec 2013 20:38:32 +0530
From: Rahul Mathur <srivmuk@gmail.com>
Subject: [Boost-users] ZeroMQ integration with Boost-v1.54
Message-ID:
      <CALtf4sBn6Dh0WVOgAfaXvSj4obMiMw335w88xDj5vguPOoHMaw@mail.gmail.com>
[…]
BTW, How do I proceed to have ZeroMQ integrated with Boost-v1.54 with
GCC on Linux platform?
Here is some code that I've been using to integrate Boost.Asio and
0MQ.  The OP question asked for GNU/Linux only but this code has been
working for me on both Windows and POSIX.  Most of this comes from
comments on various Internet forums but integrates together a few
things which I personally found less than obvious on my first few
attempts.  For example, most of the code snippets I found only
mentioned POSIX and I had to beat my head against a few walls to get
something that worked on both, though YMMV.

==> config.hpp.in <==
#ifndef INCLUDE_SHOR_DM_CONFIG_HPP
#define INCLUDE_SHOR_DM_CONFIG_HPP

// Boost C++ headers
#include <boost/filesystem.hpp>
// C++ headers
#include <string>

#ifndef _WIN32_WINNT
#  define _WIN32_WINNT 0x0601
#endif

#ifdef _MSC_VER
#  define SHOR_DM_CONSTEXPR
#  define SHOR_DM_DEFAULT_CTOR(expr) expr { }
#  define SHOR_DM_DEFAULT_COPY(expr)
#  define SHOR_DM_DEFAULT_DTOR(expr) expr { }
#  define SHOR_DM_DEFAULT_ASSIGN(expr)
#  define SHOR_DM_DELETE(expr) expr
#  define SHOR_DM_NOEXCEPT throw()
#else
#  define SHOR_DM_CONSTEXPR constexpr
#  define SHOR_DM_DEFAULT_CTOR(expr) expr = default
#  define SHOR_DM_DEFAULT_COPY(expr) expr = default
#  define SHOR_DM_DEFAULT_DTOR(expr) expr = default
#  define SHOR_DM_DEFAULT_ASSIGN(expr) expr = default
#  define SHOR_DM_DELETE(expr) expr = delete
#  define SHOR_DM_NOEXCEPT noexcept
#endif

#endif

==> zee/basic_zmq_fd.hpp <==
#ifndef INCLUDE_SHOR_DM_ZEE_BASIC_ZMQ_FD_HPP
#define INCLUDE_SHOR_DM_ZEE_BASIC_ZMQ_FD_HPP

// 0MQ headers
#include <zmq.hpp>

namespace shor {
namespace dm {
namespace zee {
template<typename Native_handle>
struct Basic_zmq_fd {
    typedef Native_handle native_handle_type;
    typedef native_handle_type value_type;

private:
    zmq::socket_t& zmq_socket_;
    Native_handle zmq_fd_;

public:
    explicit Basic_zmq_fd(zmq::socket_t& zmq_socket)
        : zmq_socket_(zmq_socket),
          zmq_fd_()
        {
            std::size_t zmq_fd_size = sizeof(zmq_fd_);
            zmq_socket_.getsockopt(ZMQ_FD, &zmq_fd_, &zmq_fd_size);
        }
    value_type operator () () const { return zmq_fd_; }
};

} // namespace zee
} // namespace dm
} // namespace shor

#endif

==> zee/handler.hpp <==
#ifndef INCLUDE_SHOR_DM_ZEE_HANDLER_HPP
#define INCLUDE_SHOR_DM_ZEE_HANDLER_HPP

// DM headers
#include "dm/zee/msg.hpp"
// Boost C++ headers
#include <boost/system/error_code.hpp>
#include <boost/function.hpp>
// C++ headers
#include <functional>

namespace shor {
namespace dm {
namespace zee {
typedef std::function<
    void (const boost::system::error_code& err, shor::dm::zee::Msg& msg)
    > Async_recv_handler;

} // namespace zee
} // namespace dm
} // namespace shor

#endif

==> zee/lossy_service.hpp <==
#ifndef INCLUDE_SHOR_DM_ZEE_LOSSY_SERVICE_HPP
#define INCLUDE_SHOR_DM_ZEE_LOSSY_SERVICE_HPP

// Boost C++ headers
#include <boost/asio.hpp>

namespace shor {
namespace dm {
namespace zee {
template<typename Service>
struct Lossy_service : public Service {
    explicit Lossy_service(boost::asio::io_service& ios)
        : Service(ios)
        { }

    void destory(typename Service::implementation_type& impl)
        { }
};

} // namespace zee
} // namespace dm
} // namespace shor

#endif

==> zee/msg.cpp <==
#include "dm/zee/msg.hpp"

shor::dm::zee::detail::Msg::Frame::Frame(
    Frame&& that)
    : that_(std::move(that.that_))
{ }

shor::dm::zee::detail::Msg::Frame::Frame(
    zmq::message_t&& that)
    : that_(std::move(that))
{ }

shor::dm::zee::detail::Msg::Frame::Frame(
    const std::string& the_other)
    : that_(the_other.size())
{
    std::copy(the_other.begin(), the_other.end(),
              static_cast<char*>(that_.data()));
}

shor::dm::zee::detail::Msg::Frame&
shor::dm::zee::detail::Msg::Frame::operator = (
    Frame&& that)
{
    that_ = std::move(that.that_);
    return *this;
}

shor::dm::zee::detail::Msg::Frame&
shor::dm::zee::detail::Msg::Frame::operator = (
    const std::string& the_other)
{
    that_.rebuild(the_other.size());
    std::copy(the_other.begin(), the_other.end(),
              static_cast<char*>(that_.data()));
    return *this;
}

shor::dm::zee::detail::Msg::Msg(
    Msg&& that)
    : frames_(std::move(that.frames_))
{ }

shor::dm::zee::detail::Msg&
shor::dm::zee::detail::Msg::operator = (
    Msg&& that)
{
    frames_ = std::move(that.frames_);
    return *this;
}

==> zee/msg.hpp <==
#ifndef INCLUDE_SHOR_DM_ZEE_MSG_HPP
#define INCLUDE_SHOR_DM_ZEE_MSG_HPP

// My header
#include "dm/zee/msg_fwd.hpp"
// DM headers
#include "dm/config.hpp"
// 0MQ headers
#include <zmq.hpp>
// C++ headers
#include <deque>

namespace shor {
namespace dm {
namespace zee {
namespace detail {
struct Msg {
    class Frame {
        zmq::message_t that_;

    public:
        SHOR_DM_DEFAULT_CTOR(Frame());
        Frame(Frame&& that);
        Frame(zmq::message_t&& that);
        Frame(const std::string& the_other);
        SHOR_DM_DEFAULT_DTOR(~Frame());

        Frame& operator = (Frame&& that);
        Frame& operator = (const std::string& the_other);
        operator zmq::message_t& ();
        operator const zmq::message_t& () const;

        zmq::message_t& get();

        const zmq::message_t& cget() const;
        const zmq::message_t& get() const;

    private:
        SHOR_DM_DELETE(Frame(const Frame& that));
        SHOR_DM_DELETE(Frame& operator = (const Frame& that));
    };

    typedef std::deque<Frame> Frames;

private:
    Frames frames_;

public:
    SHOR_DM_DEFAULT_CTOR(Msg());
    Msg(Msg&& that);
    SHOR_DM_DEFAULT_DTOR(~Msg());

    Msg& operator = (Msg&& that);

    Frames& frames();

    const Frames& frames() const;

private:
    SHOR_DM_DELETE(Msg(const Msg& that));
    SHOR_DM_DELETE(Msg& operator = (const Msg& that));
};

} // namespace detail
typedef detail::Msg Msg;

} // namespace zee
} // namespace dm
} // namespace shor

#include "dm/zee/msg.ipp"

#endif

==> zee/msg.ipp <==
#ifndef INCLUDE_SHOR_DM_ZEE_MSG_IPP
#define INCLUDE_SHOR_DM_ZEE_MSG_IPP

// My header
#include "dm/zee/msg.hpp"

inline shor::dm::zee::detail::Msg::Frame::operator
zmq::message_t& ()
{
    return that_;
}

inline shor::dm::zee::detail::Msg::Frame::operator
const zmq::message_t& () const
{
    return that_;
}

inline zmq::message_t&
shor::dm::zee::detail::Msg::Frame::get()
{
    return that_;
}

inline const zmq::message_t&
shor::dm::zee::detail::Msg::Frame::cget()
    const
{
    return that_;
}

inline const zmq::message_t&
shor::dm::zee::detail::Msg::Frame::get()
    const
{
    return that_;
}

inline shor::dm::zee::detail::Msg::Frames&
shor::dm::zee::detail::Msg::frames()
{
    return frames_;
}

inline const shor::dm::zee::detail::Msg::Frames&
shor::dm::zee::detail::Msg::frames()
    const
{
    return frames_;
}

#endif

==> zee/msg_fwd.hpp <==
#ifndef INCLUDE_SHOR_DM_ZEE_MSG_FWD_HPP
#define INCLUDE_SHOR_DM_ZEE_MSG_FWD_HPP

namespace shor {
namespace dm {
namespace zee {
namespace detail {
struct Msg;

} // namespace detail
typedef detail::Msg Msg;

} // namespace zee
} // namespace dm
} // namespace shor

#endif

==> zee/posix_zmq_socket.cpp <==
// My header
#include "dm/zee/posix_zmq_socket.hpp"
// DM headers
#include "dm/zee/handler.hpp"
// Boost C++ headers
#include <boost/bind.hpp>

struct shor::dm::zee::detail::Posix_zmq_socket::Read_handler {
    typedef Posix_zmq_socket::implementation_type Io_type;
    typedef Posix_zmq_socket::Base_read_handler Base_read_handler;

private:
    Io_type& io_;
    Base_read_handler read_handler_;
    Async_recv_handler recv_handler_;

public:
    Read_handler(
        Io_type& io,
        Base_read_handler read_handler,
        Async_recv_handler recv_handler)
        : io_(io),
          read_handler_(read_handler),
          recv_handler_(recv_handler)
        { }

    void operator () (
        const boost::system::error_code& err,
        std::size_t /*bytes_transferred*/)
        {
            read_handler_(*this, err, recv_handler_);
        }

    void operator () ()
        {
            io_.async_read_some(boost::asio::null_buffers(), *this);
        }
};

shor::dm::zee::detail::Posix_zmq_socket::Posix_zmq_socket(
    boost::asio::io_service& ios,
    zmq::context_t& zenv,
    const int zmq_socket_type)
    : Zmq_socket_base(ios, zenv, zmq_socket_type),
      zmq_fd_(Zmq_socket_base::zmq_socket()),
      io_(ios, zmq_fd_()),
      io_release_(&io_, Release())
{ }

void
shor::dm::zee::detail::Posix_zmq_socket::async_recv(
    Async_recv_handler recv_handler)
{
    const auto base_read_handler_fn = boost::bind(
        &Posix_zmq_socket::read_handler, this, _1, _2, _3);
    Read_handler read_handler(
        std::ref(io_), base_read_handler_fn, recv_handler);
    read_handler();
}

void
shor::dm::zee::detail::Posix_zmq_socket::close_ext()
{
    this->cancel();
    // TODO: reset zmq_fd_
    io_.release();
}

==> zee/posix_zmq_socket.hpp <==
#ifndef INCLUDE_SHOR_DM_ZEE_POSIX_ZMQ_SOCKET_HPP
#define INCLUDE_SHOR_DM_ZEE_POSIX_ZMQ_SOCKET_HPP

// Base class header
#include "dm/zee/zmq_socket_base.hpp"
// DM headers
#include "dm/zee/handler.hpp"
#include "dm/zee/lossy_service.hpp"
#include "dm/zee/basic_zmq_fd.hpp"
// Boost C++ headers
#include <boost/asio/posix/basic_stream_descriptor.hpp>
#include <boost/asio/posix/stream_descriptor_service.hpp>
#include <boost/asio.hpp>

namespace shor {
namespace dm {
namespace zee {
namespace detail {
class Posix_zmq_socket : public Zmq_socket_base {
    using Stream_descriptor_service =
        boost::asio::posix::stream_descriptor_service;

    template<typename Service> using Basic_stream_descriptor =
        boost::asio::posix::basic_stream_descriptor<Service>;

    // Borrowed from <http://preview.tinyurl.com/n6jh2ne> but it
    // doesn't seem to work; i.e. I was still finding that the file
    // descriptor was being closed, causing a crash at shutdown.
    // Because of this, I (re)introduced Release (see below).
    typedef Lossy_service<Stream_descriptor_service> service_type;
    typedef Basic_stream_descriptor<service_type> implementation_type;

    struct Read_handler;
    struct Release {
        void operator () (implementation_type* ptr)
            { if (ptr->is_open()) { ptr->release(); } }
    };

    Basic_zmq_fd<int> zmq_fd_;
    implementation_type io_;
    std::unique_ptr<implementation_type, Release> io_release_;

public:
    Posix_zmq_socket(
        boost::asio::io_service& ios, zmq::context_t& zenv,
        int zmq_socket_type);
    SHOR_DM_DEFAULT_DTOR(~Posix_zmq_socket());

    void async_recv(Async_recv_handler handler);
    void cancel();

private:
    SHOR_DM_DELETE(Posix_zmq_socket(const Posix_zmq_socket& that));
    SHOR_DM_DELETE(
        Posix_zmq_socket& operator = (const Posix_zmq_socket& that));

    virtual void close_ext() override;
};

} // namespace detail
typedef detail::Posix_zmq_socket Posix_zmq_socket;

} // namespace zee
} // namespace dm
} // namespace shor

#include "dm/zee/posix_zmq_socket.ipp"

#endif

==> zee/posix_zmq_socket.ipp <==
#ifndef INCLUDE_SHOR_DM_ZEE_POSIX_ZMQ_SOCKET_IPP
#define INCLUDE_SHOR_DM_ZEE_POSIX_ZMQ_SOCKET_IPP

// My header
#include "dm/zee/posix_zmq_socket.hpp"

inline void
shor::dm::zee::detail::Posix_zmq_socket::cancel()
{
    io_.cancel();
}

#endif

==> zee/windows_zmq_socket.cpp <==
// My header
#include "dm/zee/windows_zmq_socket.hpp"
// DM headers
#include "dm/zee/handler.hpp"
// Boost C++ headers
#include <boost/system/error_code.hpp>
#include <boost/system/system_error.hpp>
#include <boost/asio/error.hpp>
#include <boost/bind.hpp>

struct shor::dm::zee::detail::Windows_zmq_socket::Read_handler {
    typedef Windows_zmq_socket::implementation_type Io_type;
    typedef Windows_zmq_socket::Base_read_handler Base_read_handler;

private:
    Io_type& io_;
    Base_read_handler read_handler_;
    Async_recv_handler recv_handler_;

public:
    Read_handler(
        Io_type& io,
        Base_read_handler read_handler,
        Async_recv_handler recv_handler)
        : io_(io),
          read_handler_(read_handler),
          recv_handler_(recv_handler)
        { }

    void operator () (const boost::system::error_code& err)
        {
            read_handler_(*this, err, recv_handler_);
        }

    void operator () ()
        {
            io_.async_wait(*this);
        }
};

shor::dm::zee::detail::Windows_zmq_socket::Windows_zmq_socket(
    boost::asio::io_service& ios,
    zmq::context_t& zenv,
    const int zmq_socket_type)
    : Zmq_socket_base(ios, zenv, zmq_socket_type),
      zmq_fd_(Zmq_socket_base::zmq_socket()),
      io_(ios)
{
    auto zfd_event = CreateEvent(
        nullptr,                // default security settings
        false,                  // auto reset
        false,                  // initial state: unsignalled
        nullptr);               // anonymous object
    if (!zfd_event) {
        const boost::system::error_code an_error_code(
            GetLastError(),
            boost::asio::error::get_system_category());
        throw boost::system::system_error(an_error_code, "CreateEvent");
    }
    io_.assign(zfd_event);
    if (WSAEventSelect(zmq_fd_(), zfd_event, FD_READ) == SOCKET_ERROR) {
        const boost::system::error_code an_error_code(
            WSAGetLastError(),
            boost::asio::error::get_system_category());
        throw boost::system::system_error(
            an_error_code, "WSAEventSelect");
    }
}

void
shor::dm::zee::detail::Windows_zmq_socket::async_recv(
    Async_recv_handler recv_handler)
{
    const auto base_read_handler_fn = boost::bind(
        &Windows_zmq_socket::read_handler, this, _1, _2, _3);
    Read_handler read_handler(
        std::ref(io_), base_read_handler_fn, recv_handler);
    read_handler();
}

void
shor::dm::zee::detail::Windows_zmq_socket::close_ext()
{
    this->cancel();
    // TODO: reset zmq_fd_
    io_.close();
}

==> zee/windows_zmq_socket.hpp <==
#ifndef INCLUDE_SHOR_DM_ZEE_WINDOWS_ZMQ_SOCKET_HPP
#define INCLUDE_SHOR_DM_ZEE_WINDOWS_ZMQ_SOCKET_HPP

// Base class header
#include "dm/zee/zmq_socket_base.hpp"
// DM headers
#include "dm/zee/basic_zmq_fd.hpp"
#include "dm/zee/handler.hpp"
// Boost C++ headers
#include <boost/asio/windows/object_handle.hpp>
#include <boost/asio.hpp>

namespace shor {
namespace dm {
namespace zee {
namespace detail {
class Windows_zmq_socket : public Zmq_socket_base {
    typedef boost::asio::windows::object_handle implementation_type;

    struct Read_handler;

    Basic_zmq_fd<SOCKET> zmq_fd_;
    implementation_type io_;

public:
    Windows_zmq_socket(
        boost::asio::io_service& ios, zmq::context_t& zenv,
        int zmq_socket_type);
    SHOR_DM_DEFAULT_DTOR(~Windows_zmq_socket());

    void async_recv(Async_recv_handler handler);
    void cancel();

private:
    SHOR_DM_DELETE(Windows_zmq_socket(const Windows_zmq_socket& that));
    SHOR_DM_DELETE(
        Windows_zmq_socket& operator = (const Windows_zmq_socket& that));

    virtual void close_ext() override;
};

} // namespace detail
typedef detail::Windows_zmq_socket Windows_zmq_socket;

} // namespace zee
} // namespace dm
} // namespace shor

#include "dm/zee/windows_zmq_socket.ipp"

#endif

==> zee/windows_zmq_socket.ipp <==
#ifndef INCLUDE_SHOR_DM_ZEE_WINDOWS_ZMQ_SOCKET_IPP
#define INCLUDE_SHOR_DM_ZEE_WINDOWS_ZMQ_SOCKET_IPP

// My header
#include "dm/zee/windows_zmq_socket.hpp"

inline void
shor::dm::zee::detail::Windows_zmq_socket::cancel()
{
    io_.cancel();
}

#endif

==> zee/zmq_socket.hpp <==
#ifndef INCLUDE_SHOR_DM_ZEE_ZMQ_SOCKET_HPP
#define INCLUDE_SHOR_DM_ZEE_ZMQ_SOCKET_HPP

// My header
#include "dm/zee/zmq_socket_fwd.hpp"
// Conditional headers
#ifndef _MSC_VER
#  include "dm/zee/posix_zmq_socket.hpp"
#else
#  include "dm/zee/windows_zmq_socket.hpp"
#endif
// Boost C++ headers
#include <boost/asio.hpp>

namespace shor {
namespace dm {
namespace zee {
#ifndef _MSC_VER
typedef Posix_zmq_socket Zmq_socket;
#else
typedef Windows_zmq_socket Zmq_socket;
#endif

} // namespace zee
} // namespace dm
} // namespace shor

#endif

==> zee/zmq_socket_base.cpp <==
// My headers
#include "dm/zee/zmq_socket_base.hpp"
// C++ headers
#include <cassert>

shor::dm::zee::detail::Zmq_socket_base::Opts::Opts(
    zmq::socket_t& zmq_socket)
    : zmq_socket_(zmq_socket)
{ }

shor::dm::zee::detail::Zmq_socket_base::Zmq_socket_base(
    boost::asio::io_service& ios,
    zmq::context_t& zenv,
    const int zmq_socket_type)
    : ios_(ios),
      zenv_(zenv),
      zmq_socket_(zenv, zmq_socket_type),
      opts_(zmq_socket_)
{
    opts_.linger(0);
}

bool
shor::dm::zee::detail::Zmq_socket_base::send(
    Msg msg)
{
    if (msg.frames().empty()) {
        auto frame = Msg::Frame();
        const auto send_p = zmq_socket_.send(frame);
        static_cast<void>(opts_.events());
        return send_p;
    }
    const auto first = msg.frames().begin();
    const auto last = std::prev(msg.frames().end());
    for (auto pos = first; pos != last; ++pos) {
        const auto send_p = zmq_socket_.send(*pos, ZMQ_SNDMORE);
        static_cast<void>(opts_.events());
        if (!send_p) {
            return false;
        }
    }
    const auto send_p = zmq_socket_.send(msg.frames().back(), 0);
    static_cast<void>(opts_.events());
    return send_p;
}

shor::dm::zee::Msg
shor::dm::zee::detail::Zmq_socket_base::recv()
{
    Msg my_value;
    int more_p = 0;
    do {
        zmq::message_t frame;
        const int flags = 0;
        zmq_socket_.recv(&frame, flags);
        static_cast<void>(opts_.events());
        my_value.frames().push_back(std::move(frame));
        more_p = opts_.rcvmore();
    } while (more_p);
    return my_value;
}

void
shor::dm::zee::detail::Zmq_socket_base::read_handler(
    std::function<void ()> io_fn,
    const boost::system::error_code& err,
    const Async_recv_handler recv_handler)
{
    if (err == boost::asio::error::operation_aborted) {
        auto msg = Msg();
        recv_handler(err, msg);
    } else if (opts_.events() & ZMQ_POLLIN) {
        do {
            auto msg = this->recv();
            recv_handler(err, msg);
        } while (opts_.events() & ZMQ_POLLIN);
    } else {
        io_fn();
    }
}

==> zee/zmq_socket_base.hpp <==
#ifndef INCLUDE_SHOR_DM_ZEE_ZMQ_SOCKET_BASE_HPP
#define INCLUDE_SHOR_DM_ZEE_ZMQ_SOCKET_BASE_HPP

// DM headers
#include "dm/zee/handler.hpp"
#include "dm/zee/msg.hpp"
// 0MQ headers
#include <zmq.hpp>
// Boost C++ headers
#include <boost/asio.hpp>
// C++ headers
#include <string>

namespace shor {
namespace dm {
namespace zee {
namespace detail {
struct Zmq_socket_base {
    class Opts {
        zmq::socket_t& zmq_socket_;

    public:
        explicit Opts(zmq::socket_t& zmq_socket);
        SHOR_DM_DEFAULT_DTOR(~Opts());

        void linger(int value);
        void subscribe(const std::string& value);

        int events() const;
        int linger() const;
        int rcvmore() const;

    private:
        template<typename Opt_value>
        Opt_value getsockopt(int opt_name) const;

        void setsockopt(int opt_name, const std::string& opt_value);

        template<typename Opt_value>
        void setsockopt(int opt_name, const Opt_value& opt_value);

        SHOR_DM_DELETE(Opts(const Opts& that));
        SHOR_DM_DELETE(Opts& operator = (const Opts& that));
    };

private:
    boost::asio::io_service& ios_;
    zmq::context_t& zenv_;
    zmq::socket_t zmq_socket_;
    Opts opts_;

public:
    void bind(const char* addr);
    void bind(const std::string& addr);
    void close();
    void connect(const char* addr);
    void connect(const std::string& addr);
    boost::asio::io_service& get_io_service();
    Opts& opts();
    bool send(Msg msg);
    Msg recv();

    bool connected() const;
    const boost::asio::io_service& get_const_io_service() const;
    const boost::asio::io_service& get_io_service() const;
    const Opts& opts() const;

protected:
    typedef std::function<void ()> Void_fn;
    typedef std::function<
        void (Void_fn, const boost::system::error_code& err,
              Async_recv_handler recv_handler)
        > Base_read_handler;

    Zmq_socket_base(
        boost::asio::io_service& ios, zmq::context_t& zenv,
        int zmq_socket_type);
    SHOR_DM_DEFAULT_DTOR(virtual ~Zmq_socket_base());

    void read_handler(
        std::function<void ()> io_fn,
        const boost::system::error_code& err,
        Async_recv_handler handler);
    zmq::context_t& zenv();
    zmq::socket_t& zmq_socket();

    const zmq::context_t& const_zenv() const;
    const zmq::socket_t& const_zmq_socket() const;
    const zmq::context_t& zenv() const;
    const zmq::socket_t& zmq_socket() const;

private:
    SHOR_DM_DELETE(Zmq_socket_base(const Zmq_socket_base& that));
    SHOR_DM_DELETE(
        Zmq_socket_base& operator = (const Zmq_socket_base& that));

    virtual void close_ext() = 0;
};

} // namespace detail
typedef detail::Zmq_socket_base Zmq_socket_base;

} // namespace zee
} // namespace dm
} // namespace shor

#include "dm/zee/zmq_socket_base.ipp"

#endif

==> zee/zmq_socket_base.ipp <==
#ifndef INCLUDE_SHOR_DM_ZEE_ZMQ_SOCKET_BASE_IPP
#define INCLUDE_SHOR_DM_ZEE_ZMQ_SOCKET_BASE_IPP

// My header
#include "dm/zee/zmq_socket_base.hpp"

inline void
shor::dm::zee::detail::Zmq_socket_base::Opts::linger(
    const int value)
{
    this->setsockopt(ZMQ_LINGER, value);
}

inline void
shor::dm::zee::detail::Zmq_socket_base::Opts::subscribe(
    const std::string& value)
{
    this->setsockopt(ZMQ_SUBSCRIBE, value);
}

inline int
shor::dm::zee::detail::Zmq_socket_base::Opts::events()
    const
{
    return this->getsockopt<int>(ZMQ_EVENTS);
}

inline int
shor::dm::zee::detail::Zmq_socket_base::Opts::linger()
    const
{
    return this->getsockopt<int>(ZMQ_LINGER);
}

inline int
shor::dm::zee::detail::Zmq_socket_base::Opts::rcvmore()
    const
{
    return this->getsockopt<int>(ZMQ_RCVMORE);
}

template<typename Opt_value>
Opt_value
shor::dm::zee::detail::Zmq_socket_base::Opts::getsockopt(
    const int opt_name)
    const
{
    Opt_value opt_value;
    std::size_t opt_size = sizeof(Opt_value);
    zmq_socket_.getsockopt(opt_name, &opt_value, &opt_size);
    return opt_value;
}

inline void
shor::dm::zee::detail::Zmq_socket_base::Opts::setsockopt(
    const int opt_name,
    const std::string& opt_value)
{
    zmq_socket_.setsockopt(opt_name, opt_value.c_str(), opt_value.size());
}

template<typename Opt_value>
inline void
shor::dm::zee::detail::Zmq_socket_base::Opts::setsockopt(
    const int opt_name,
    const Opt_value& opt_value)
{
    const std::size_t opt_size = sizeof(Opt_value);
    zmq_socket_.setsockopt(opt_name, &opt_value, opt_size);
}

inline void
shor::dm::zee::detail::Zmq_socket_base::bind(
    const char* addr)
{
    zmq_socket_.bind(addr);
}

inline void
shor::dm::zee::detail::Zmq_socket_base::bind(
    const std::string& addr)
{
    zmq_socket_.bind(addr.c_str());
}

inline void
shor::dm::zee::detail::Zmq_socket_base::close()
{
    this->close_ext();
    zmq_socket_.close();
}

inline void
shor::dm::zee::detail::Zmq_socket_base::connect(
    const char* addr)
{
    zmq_socket_.connect(addr);
}

inline void
shor::dm::zee::detail::Zmq_socket_base::connect(
    const std::string& addr)
{
    zmq_socket_.connect(addr.c_str());
}

inline boost::asio::io_service&
shor::dm::zee::detail::Zmq_socket_base::get_io_service()
{
    return ios_;
}

inline shor::dm::zee::detail::Zmq_socket_base::Opts&
shor::dm::zee::detail::Zmq_socket_base::opts()
{
    return opts_;
}

inline bool
shor::dm::zee::detail::Zmq_socket_base::connected()
    const
{
    // This function should have been const in zmq::socket_t
    return const_cast<zmq::socket_t&>(zmq_socket_).connected();
}

inline const boost::asio::io_service&
shor::dm::zee::detail::Zmq_socket_base::get_const_io_service()
    const
{
    return ios_;
}

inline const boost::asio::io_service&
shor::dm::zee::detail::Zmq_socket_base::get_io_service()
    const
{
    return ios_;
}

inline zmq::context_t&
shor::dm::zee::detail::Zmq_socket_base::zenv()
{
    return zenv_;
}

inline zmq::socket_t&
shor::dm::zee::detail::Zmq_socket_base::zmq_socket()
{
    return zmq_socket_;
}

inline const zmq::context_t&
shor::dm::zee::detail::Zmq_socket_base::const_zenv()
    const
{
    return zenv_;
}

inline const zmq::socket_t&
shor::dm::zee::detail::Zmq_socket_base::const_zmq_socket()
    const
{
    return zmq_socket_;
}

inline const zmq::context_t&
shor::dm::zee::detail::Zmq_socket_base::zenv()
    const
{
    return zenv_;
}

inline const zmq::socket_t&
shor::dm::zee::detail::Zmq_socket_base::zmq_socket()
    const
{
    return zmq_socket_;
}

#endif

==> zee/zmq_socket_fwd.hpp <==
#ifndef INCLUDE_SHOR_DM_ZEE_ZMQ_SOCKET_FWD_HPP
#define INCLUDE_SHOR_DM_ZEE_ZMQ_SOCKET_FWD_HPP

namespace shor {
namespace dm {
namespace zee {
namespace detail {
#ifndef _MSC_VER
class Posix_zmq_socket;
#else
class Windows_zmq_socket;
#endif

} // namespace detail
#ifndef _MSC_VER
typedef detail::Posix_zmq_socket Posix_zmq_socket;
typedef Posix_zmq_socket Zmq_socket;
#else
typedef detail::Windows_zmq_socket Windows_zmq_socket;
typedef Windows_zmq_socket Zmq_socket;
#endif

} // namespace zee
} // namespace dm
} // namespace shor

#endif

________________________________

This e-mail and any attachments are confidential. If it is not intended for you, please notify the sender, and please erase and ignore the contents.

Re: [Boost-users] Boost-users Digest, Vol 3643, Issue 1

Damien Kick