Overview Tutorial API reference Examples Build Download

ZmqMessage 0.1 - 21 Oct 2011

---

include/ZmqMessage.hpp

Go to the documentation of this file.

#include <cstddef>
#include <vector>
#include <tr1/array>
#include <string>
#include <memory>
#include <zmq.hpp>

#include <ZmqMessageFwd.hpp>

#include <zmqmessage/Config.hpp>
#include <zmqmessage/DelObj.hpp>
#include <zmqmessage/NonCopyable.hpp>
#include <zmqmessage/MsgPtrVec.hpp>
#include <zmqmessage/RawMessage.hpp>

#include <ZmqTools.hpp>

#ifndef ZMQMESSAGE_HPP_
#define ZMQMESSAGE_HPP_

namespace ZmqMessage
{
  ZMQMESSAGE_EXCEPTION_MACRO(MessageFormatError)
  ;

  ZMQMESSAGE_EXCEPTION_MACRO(NoSuchPartError)
  ;

  void
  send(zmq::socket_t& sock, Multipart& multipart, bool nonblock)
    throw(ZmqErrorType);


  // routing policies

  class SimpleRouting
  {
  protected:
    inline void
    receive_routing(zmq::socket_t& sock) {}

    inline
    MsgPtrVec*
    get_routing()
    {
      return 0;
    }

    inline
    void
    log_routing_received() const {}
  };

  class XRouting
  {
  private:
    MsgPtrVec routing_; 

  protected:
    void
    receive_routing(zmq::socket_t& sock)
      throw (MessageFormatError, ZmqErrorType);

    inline
    MsgPtrVec*
    get_routing()
    {
      return &routing_;
    }

    void
    log_routing_received() const;

    ~XRouting();
  };

  class Multipart : private Private::NonCopyable
  {
  protected:
    MsgPtrVec parts_;

    void
    check_has_part(size_t n) const throw(NoSuchPartError);

  private:

    void
    clear();

    friend class Sink;

    friend void
    send(zmq::socket_t& sock, Multipart& multipart, bool nonblock)
      throw(ZmqErrorType);

  public:
    template <typename T>
    class iterator
    {
    public:
      typedef T value_type;
      typedef T& reference;
      typedef T* pointer;
      typedef std::ptrdiff_t difference_type;
      typedef std::input_iterator_tag iterator_category;

    private:
      explicit
      iterator(const MsgPtrVec& messages, bool binary_mode, bool end = false) :
          messages_(messages), idx_(end ? messages.size() : 0),
          binary_mode_(binary_mode)
      {
        set_cur();
      }

      iterator(const MsgPtrVec& messages, size_t idx, bool binary_mode) :
          messages_(messages), idx_(idx), binary_mode_(binary_mode)
      {
        set_cur();
      }

      friend class Multipart;

    public:
      iterator
      operator++()
      {
        ++idx_;
        set_cur();
        return *this;
      }

      iterator
      operator++(int)
      {
        iterator<T> ret_val(*this);
        ++(*this);
        return ret_val;
      }

      const T
      operator*()
      {
        return cur_;
      }

      bool
      operator==(const iterator<T>& rhs)
      {
        return equal(rhs);
      }

      bool
      operator!=(const iterator<T>& rhs)
      {
        return !equal(rhs);
      }

    private:
      const MsgPtrVec& messages_;
      size_t idx_;
      T cur_;
      const bool binary_mode_;

      inline
      bool
      end() const
      {
        return idx_ >= messages_.size();
      }

      bool
      equal(const iterator<T>& rhs) const
      {
        return end() && rhs.end();
      }

      void
      set_cur();
    };
  public:
    virtual ~Multipart()
    {
      clear();
    }

    Multipart*
    detach();

    inline
    bool
    has_part(size_t idx)
    {
      return (parts_.size() > idx && parts_[idx] != 0);
    }

    inline
    void
    reserve(size_t sz)
    {
      parts_.reserve(sz);
    }

    template <typename T>
    inline
    iterator<T>
    begin(bool binary_mode = false) const
    {
      return iterator<T>(parts_, binary_mode);
    }

    inline
    iterator<ZMQMESSAGE_STRING_CLASS>
    beginstr() const
    {
      return begin<ZMQMESSAGE_STRING_CLASS>();
    }

    template <typename T>
    inline
    iterator<T>
    iter_at(size_t pos, bool binary_mode = false) const
    {
      return iterator<T>(parts_, pos, binary_mode);
    }

    template <typename T>
    inline
    iterator<T>
    end() const
    {
      return iterator<T>(parts_, false, true);
    }

    inline
    iterator<ZMQMESSAGE_STRING_CLASS>
    endstr() const
    {
      return end<ZMQMESSAGE_STRING_CLASS>();
    }

    inline
    size_t
    size() const
    {
      return parts_.size();
    }

    zmq::message_t&
    operator[](size_t i) throw (NoSuchPartError);

    zmq::message_t*
    release(size_t i);

    inline
    std::auto_ptr<zmq::message_t>
    release_ptr(size_t i)
    {
      return std::auto_ptr<zmq::message_t>(release(i));
    }
  };

  Sink&
  NullMessage(Sink& out);

  template<typename RoutingPolicy>
  Incoming<RoutingPolicy>&
  Skip(Incoming<RoutingPolicy>& in)
  {
    in.check_has_part(in.cur_extract_idx_);
    ++in.cur_extract_idx_;
    return in;
  }

  Sink&
  Flush(Sink& out);

  template<typename StreamAlike>
  StreamAlike&
  Binary(StreamAlike& out)
  {
    out.set_binary();
    return out;
  }

  template<typename StreamAlike>
  StreamAlike&
  Text(StreamAlike& out)
  {
    out.set_text();
    return out;
  }

  template <class RoutingPolicy>
  class Incoming :
    public Multipart, private RoutingPolicy
  {
  private:

    zmq::socket_t& src_; 
    bool is_terminal_; 
    size_t cur_extract_idx_;
    bool binary_mode_; 

    void
    append_message_data(
      zmq::message_t& message, std::vector<char>& area) const;

    bool
    receive_one() throw(ZmqErrorType);

    template <class OutRoutingPolicy>
    friend class Outgoing;

    inline
    MsgPtrVec*
    get_routing()
    {
      return RoutingPolicy::get_routing();
    }

    friend
    Incoming<RoutingPolicy>&
    Skip<RoutingPolicy>(Incoming<RoutingPolicy>&);

  public:
    typedef RoutingPolicy RoutingPolicyType;

    explicit Incoming(zmq::socket_t& sock)
        : src_(sock), is_terminal_(false),
          cur_extract_idx_(0), binary_mode_(false)
    {
    }

    inline
    zmq::socket_t&
    src()
    {
      return src_;
    }

    inline
    bool
    is_terminal() const
    {
      return is_terminal_;
    }

    void
    check_is_terminal() const throw(MessageFormatError);

    void
    validate(
      const char* part_names[],
      size_t part_names_length, bool strict)
      throw (MessageFormatError);

    Incoming <RoutingPolicy>&
    receive(
      size_t parts, const char* part_names[],
      size_t part_names_length, bool check_terminal)
      throw (MessageFormatError, ZmqErrorType);

    template <size_t N>
    inline
    Incoming <RoutingPolicy>&
    receive(
      std::tr1::array<const char*, N> part_names, bool check_terminal)
      throw (MessageFormatError)
    {
      return receive(N, part_names.data(), check_terminal);
    }

    inline
    Incoming <RoutingPolicy>&
    receive(size_t parts, bool check_terminal)
      throw (MessageFormatError, ZmqErrorType)
    {
      return receive(parts, 0, 0, check_terminal);
    }

    inline
    Incoming <RoutingPolicy>&
    receive(
      size_t parts, const char* part_names[], bool check_terminal)
      throw (MessageFormatError, ZmqErrorType)
    {
      return receive(parts, part_names, parts, check_terminal);
    }

    Incoming <RoutingPolicy>&
    receive_all(
      const size_t min_parts,
      const char* part_names[], size_t part_names_length)
      throw (MessageFormatError, ZmqErrorType);

    inline
    Incoming <RoutingPolicy>&
    receive_all(const size_t min_parts, const char* part_names[])
      throw (MessageFormatError, ZmqErrorType)
    {
      return receive_all(min_parts, part_names, min_parts);
    }

    inline
    Incoming <RoutingPolicy>&
    receive_all()
      throw (MessageFormatError, ZmqErrorType)
    {
      return receive_all(0, 0, 0);
    }

    inline
    Incoming <RoutingPolicy>&
    receive_all(const size_t min_parts)
      throw (MessageFormatError, ZmqErrorType)
    {
      return receive_all(min_parts, 0, 0);
    }

    Incoming <RoutingPolicy>&
    receive_up_to(size_t min_parts, const char* part_names[],
      size_t max_parts) throw (MessageFormatError, ZmqErrorType);

    int
    fetch_tail(std::vector<char>& area, const char* delimiter = 0)
      throw (ZmqErrorType);

    int
    drop_tail() throw(ZmqErrorType);

    template <typename T>
    Incoming<RoutingPolicy>&
    operator>> (T& t) throw(NoSuchPartError);

    Incoming<RoutingPolicy>&
    operator>> (zmq::message_t& msg) throw(NoSuchPartError);

    Incoming<RoutingPolicy>&
    operator>> (Incoming<RoutingPolicy>& (*f)(Incoming<RoutingPolicy>&))
    {
      return f(*this);
    }

    void
    set_binary()
    {
      binary_mode_ = true;
    }

    void
    set_text()
    {
      binary_mode_ = false;
    }
  };

  struct OutOptions
  {
    static const unsigned NONBLOCK = 0x1;
    static const unsigned CACHE_ON_BLOCK = 0x2;
    static const unsigned DROP_ON_BLOCK = 0x4;
    static const unsigned COPY_INCOMING = 0x8;

    static const unsigned EMULATE_BLOCK_SENDS = 0x10;

    static const unsigned BINARY_MODE = 0x20;

    zmq::socket_t& sock;
    unsigned options;

    inline OutOptions(zmq::socket_t& sock_p, unsigned options_p) :
      sock(sock_p), options(options_p)
    {}
  };

  class Sink : private Private::NonCopyable
  {
  public:
    template <typename T>
    class iterator
    {
    public:
      typedef T value_type;
      typedef T& reference;
      typedef T* pointer;
      typedef std::ptrdiff_t difference_type;
      typedef std::output_iterator_tag iterator_category;

    private:
      class AssignProxy
      {
      public:
        AssignProxy(Sink& outgoing)
            : outgoing_(outgoing)
        {
        }

        void
        operator=(const T& output_object)
        {
          outgoing_ << output_object;
        }

      private:
        Sink& outgoing_;
      };

    public:
      explicit
      iterator(Sink& outgoing)
          : outgoing_(outgoing)
      {
      }

      iterator
      operator++()
      {
        return *this;
      }

      iterator
      operator++(int)
      {
        return *this;
      }

      AssignProxy
      operator*()
      {
        return AssignProxy(outgoing_);
      }

      bool
      operator==(const iterator<T>& rhs)
      {
        return equal(rhs);
      }

      bool
      operator!=(const iterator<T>& rhs)
      {
        return !equal(rhs);
      }

    private:
      Sink& outgoing_;

      bool
      equal(const iterator<T>& rhs) const
      {
        return false;
      }
    };

  private:
    typedef std::auto_ptr<zmq::message_t> MsgPtr;

    zmq::socket_t& dst_;

    unsigned options_;

    Multipart* incoming_;

    std::auto_ptr<Multipart> outgoing_queue_;

    MsgPtr cached_;

    enum State
    {
      NOTSENT, //<! we haven't sent messages
      SENDING, //<! we have sent at least one message, it's OK
      QUEUEING,  //<! we failed to send, queueing to @c outgoing_queue_ instead
      DROPPING,  //<! we failed to send, dropped. Terminal state
      FLUSHED  //<! Message is flushed, all messages are sent or queued, no more messages accepted
    };

    State state_;

  protected:
    Sink(zmq::socket_t& dst, unsigned options, Multipart* incoming = 0) :
      dst_(dst), options_(options), incoming_(incoming),
      outgoing_queue_(0), cached_(0), state_(NOTSENT)
    {}

    inline
    unsigned
    options() const
    {
      return options_;
    }

    void
    send_one(
      zmq::message_t* msg, bool use_copy = false)
      throw(ZmqErrorType);

  private:
    void
    send_owned(zmq::message_t* owned) throw(ZmqErrorType);

    void
    do_send_one(
      zmq::message_t* msg, bool last) throw(ZmqErrorType);

    bool
    try_send_first_cached(
      bool last) throw(ZmqErrorType);

    void
    add_to_queue(zmq::message_t* msg);

  public:
    virtual
    ~Sink();

     const Multipart*
     incoming() const
     {
       return incoming_;
     }

     inline
     Multipart*
     detach()
     {
       return outgoing_queue_.release();
     }

     inline
     bool
     is_queued() const
     {
       return (outgoing_queue_.get() != 0);
     }

     inline
     bool
     is_dropping() const
     {
       return (state_ == DROPPING);
     }

     inline
     zmq::socket_t&
     dst()
     {
       return dst_;
     }

     void
     flush() throw(ZmqErrorType);

     void
     set_binary()
     {
       options_ |= OutOptions::BINARY_MODE;
     }

     void
     set_text()
     {
       options_ &= ~OutOptions::BINARY_MODE;
     }

     void
     send_incoming_messages(size_t idx_from = 0, size_t idx_to = UINT_MAX)
       throw(ZmqErrorType);

     void
     send_incoming_messages(Multipart& multipart,
       bool copy, size_t idx_from = 0, size_t idx_to = UINT_MAX)
       throw(ZmqErrorType);

     void
     relay_from(zmq::socket_t& relay_src) throw(ZmqErrorType);

     template <class OccupationAccumulator>
     void
     relay_from(
       zmq::socket_t& relay_src, OccupationAccumulator acc)
       throw (ZmqErrorType);

     template <typename T>
     Sink&
     operator<< (const T& t) throw (ZmqErrorType);

     Sink&
     operator<< (zmq::message_t& msg) throw (ZmqErrorType);

     inline Sink&
     operator<< (MsgPtr msg) throw (ZmqErrorType)
     {
       send_owned(msg.get() ? msg.release() : new zmq::message_t(0));
       return *this;
     }

     Sink&
     operator<< (const RawMessage& m) throw (ZmqErrorType);

     inline Sink&
     operator<< (Sink& (*f)(Sink&))
     {
       return f(*this);
     }
  };

  template <class RoutingPolicy>
  class Outgoing : public Sink
  {
  private:

    void
    send_routing(
      MsgPtrVec* routing) throw(ZmqErrorType);

  public:

    using Sink::iterator;

    typedef RoutingPolicy RoutingPolicyType;

    Outgoing(zmq::socket_t& dst, unsigned options) :
      Sink(dst, options)
    {
      send_routing(0);
    }

    explicit Outgoing(OutOptions out_opts) :
      Sink(out_opts.sock, out_opts.options)
    {
      send_routing(0);
    }

    template <typename InRoutingPolicy>
    Outgoing(zmq::socket_t& dst,
      Incoming<InRoutingPolicy>& incoming,
      unsigned options) throw(ZmqErrorType) :
      Sink(dst, options, &incoming)
    {
      send_routing(incoming.get_routing());
    }

    template <typename InRoutingPolicy>
    Outgoing(OutOptions out_opts,
      Incoming<InRoutingPolicy>& incoming) throw(ZmqErrorType) :
      Sink(out_opts.sock, out_opts.options, &incoming)
    {
      send_routing(incoming.get_routing());
    }

    Outgoing(zmq::socket_t& dst,
      Multipart& incoming,
      unsigned options) throw(ZmqErrorType) :
      Sink(dst, options, &incoming)
    {
      send_routing(0);
    }

    Outgoing(OutOptions out_opts,
      Multipart& incoming) throw(ZmqErrorType) :
      Sink(out_opts.sock, out_opts.options, &incoming)
    {
      send_routing(0);
    }
  };
}

#endif /* ZMQMESSAGE_HPP_ */

#include "zmqmessage/ZmqMessageTemplateImpl.hpp"

---

ZmqMessage 0.1 — open source software, support@zmqmessage.org