zmq_addon.hpp 26 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848
  1. /*
  2. Copyright (c) 2016-2017 ZeroMQ community
  3. Copyright (c) 2016 VOCA AS / Harald Nøkland
  4. Permission is hereby granted, free of charge, to any person obtaining a copy
  5. of this software and associated documentation files (the "Software"), to
  6. deal in the Software without restriction, including without limitation the
  7. rights to use, copy, modify, merge, publish, distribute, sublicense, and/or
  8. sell copies of the Software, and to permit persons to whom the Software is
  9. furnished to do so, subject to the following conditions:
  10. The above copyright notice and this permission notice shall be included in
  11. all copies or substantial portions of the Software.
  12. THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  13. IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  14. FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
  15. AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  16. LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
  17. FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
  18. IN THE SOFTWARE.
  19. */
  20. #ifndef __ZMQ_ADDON_HPP_INCLUDED__
  21. #define __ZMQ_ADDON_HPP_INCLUDED__
  22. #include "zmq.hpp"
  23. #include <deque>
  24. #include <iomanip>
  25. #include <sstream>
  26. #include <stdexcept>
  27. #ifdef ZMQ_CPP11
  28. #include <limits>
  29. #include <functional>
  30. #include <unordered_map>
  31. namespace zmq
  32. {
  33. // socket ref or native file descriptor for poller
  34. class poller_ref_t
  35. {
  36. public:
  37. enum RefType
  38. {
  39. RT_SOCKET,
  40. RT_FD
  41. };
  42. poller_ref_t() : poller_ref_t(socket_ref{})
  43. {}
  44. poller_ref_t(const zmq::socket_ref& socket) : data{RT_SOCKET, socket, {}}
  45. {}
  46. poller_ref_t(zmq::fd_t fd) : data{RT_FD, {}, fd}
  47. {}
  48. size_t hash() const ZMQ_NOTHROW
  49. {
  50. std::size_t h = 0;
  51. hash_combine(h, std::get<0>(data));
  52. hash_combine(h, std::get<1>(data));
  53. hash_combine(h, std::get<2>(data));
  54. return h;
  55. }
  56. bool operator == (const poller_ref_t& o) const ZMQ_NOTHROW
  57. {
  58. return data == o.data;
  59. }
  60. private:
  61. template <class T>
  62. static void hash_combine(std::size_t& seed, const T& v) ZMQ_NOTHROW
  63. {
  64. std::hash<T> hasher;
  65. seed ^= hasher(v) + 0x9e3779b9 + (seed<<6) + (seed>>2);
  66. }
  67. std::tuple<int, zmq::socket_ref, zmq::fd_t> data;
  68. }; // class poller_ref_t
  69. } // namespace zmq
  70. // std::hash<> specialization for std::unordered_map
  71. template <> struct std::hash<zmq::poller_ref_t>
  72. {
  73. size_t operator()(const zmq::poller_ref_t& ref) const ZMQ_NOTHROW
  74. {
  75. return ref.hash();
  76. }
  77. };
  78. #endif // ZMQ_CPP11
  79. namespace zmq
  80. {
  81. #ifdef ZMQ_CPP11
  82. namespace detail
  83. {
  84. template<bool CheckN, class OutputIt>
  85. recv_result_t
  86. recv_multipart_n(socket_ref s, OutputIt out, size_t n, recv_flags flags)
  87. {
  88. size_t msg_count = 0;
  89. message_t msg;
  90. while (true) {
  91. if ZMQ_CONSTEXPR_IF (CheckN) {
  92. if (msg_count >= n)
  93. throw std::runtime_error(
  94. "Too many message parts in recv_multipart_n");
  95. }
  96. if (!s.recv(msg, flags)) {
  97. // zmq ensures atomic delivery of messages
  98. assert(msg_count == 0);
  99. return {};
  100. }
  101. ++msg_count;
  102. const bool more = msg.more();
  103. *out++ = std::move(msg);
  104. if (!more)
  105. break;
  106. }
  107. return msg_count;
  108. }
  109. inline bool is_little_endian()
  110. {
  111. const uint16_t i = 0x01;
  112. return *reinterpret_cast<const uint8_t *>(&i) == 0x01;
  113. }
  114. inline void write_network_order(unsigned char *buf, const uint32_t value)
  115. {
  116. if (is_little_endian()) {
  117. ZMQ_CONSTEXPR_VAR uint32_t mask = (std::numeric_limits<std::uint8_t>::max)();
  118. *buf++ = static_cast<unsigned char>((value >> 24) & mask);
  119. *buf++ = static_cast<unsigned char>((value >> 16) & mask);
  120. *buf++ = static_cast<unsigned char>((value >> 8) & mask);
  121. *buf++ = static_cast<unsigned char>(value & mask);
  122. } else {
  123. std::memcpy(buf, &value, sizeof(value));
  124. }
  125. }
  126. inline uint32_t read_u32_network_order(const unsigned char *buf)
  127. {
  128. if (is_little_endian()) {
  129. return (static_cast<uint32_t>(buf[0]) << 24)
  130. + (static_cast<uint32_t>(buf[1]) << 16)
  131. + (static_cast<uint32_t>(buf[2]) << 8)
  132. + static_cast<uint32_t>(buf[3]);
  133. } else {
  134. uint32_t value;
  135. std::memcpy(&value, buf, sizeof(value));
  136. return value;
  137. }
  138. }
  139. } // namespace detail
  140. /* Receive a multipart message.
  141. Writes the zmq::message_t objects to OutputIterator out.
  142. The out iterator must handle an unspecified number of writes,
  143. e.g. by using std::back_inserter.
  144. Returns: the number of messages received or nullopt (on EAGAIN).
  145. Throws: if recv throws. Any exceptions thrown
  146. by the out iterator will be propagated and the message
  147. may have been only partially received with pending
  148. message parts. It is adviced to close this socket in that event.
  149. */
  150. template<class OutputIt>
  151. ZMQ_NODISCARD recv_result_t recv_multipart(socket_ref s,
  152. OutputIt out,
  153. recv_flags flags = recv_flags::none)
  154. {
  155. return detail::recv_multipart_n<false>(s, std::move(out), 0, flags);
  156. }
  157. /* Receive a multipart message.
  158. Writes at most n zmq::message_t objects to OutputIterator out.
  159. If the number of message parts of the incoming message exceeds n
  160. then an exception will be thrown.
  161. Returns: the number of messages received or nullopt (on EAGAIN).
  162. Throws: if recv throws. Throws std::runtime_error if the number
  163. of message parts exceeds n (exactly n messages will have been written
  164. to out). Any exceptions thrown
  165. by the out iterator will be propagated and the message
  166. may have been only partially received with pending
  167. message parts. It is adviced to close this socket in that event.
  168. */
  169. template<class OutputIt>
  170. ZMQ_NODISCARD recv_result_t recv_multipart_n(socket_ref s,
  171. OutputIt out,
  172. size_t n,
  173. recv_flags flags = recv_flags::none)
  174. {
  175. return detail::recv_multipart_n<true>(s, std::move(out), n, flags);
  176. }
  177. /* Send a multipart message.
  178. The range must be a ForwardRange of zmq::message_t,
  179. zmq::const_buffer or zmq::mutable_buffer.
  180. The flags may be zmq::send_flags::sndmore if there are
  181. more message parts to be sent after the call to this function.
  182. Returns: the number of messages sent (exactly msgs.size()) or nullopt (on EAGAIN).
  183. Throws: if send throws. Any exceptions thrown
  184. by the msgs range will be propagated and the message
  185. may have been only partially sent. It is adviced to close this socket in that event.
  186. */
  187. template<class Range
  188. #ifndef ZMQ_CPP11_PARTIAL
  189. ,
  190. typename = typename std::enable_if<
  191. detail::is_range<Range>::value
  192. && (std::is_same<detail::range_value_t<Range>, message_t>::value
  193. || detail::is_buffer<detail::range_value_t<Range>>::value)>::type
  194. #endif
  195. >
  196. send_result_t
  197. send_multipart(socket_ref s, Range &&msgs, send_flags flags = send_flags::none)
  198. {
  199. using std::begin;
  200. using std::end;
  201. auto it = begin(msgs);
  202. const auto end_it = end(msgs);
  203. size_t msg_count = 0;
  204. while (it != end_it) {
  205. const auto next = std::next(it);
  206. const auto msg_flags =
  207. flags | (next == end_it ? send_flags::none : send_flags::sndmore);
  208. if (!s.send(*it, msg_flags)) {
  209. // zmq ensures atomic delivery of messages
  210. assert(it == begin(msgs));
  211. return {};
  212. }
  213. ++msg_count;
  214. it = next;
  215. }
  216. return msg_count;
  217. }
  218. /* Encode a multipart message.
  219. The range must be a ForwardRange of zmq::message_t. A
  220. zmq::multipart_t or STL container may be passed for encoding.
  221. Returns: a zmq::message_t holding the encoded multipart data.
  222. Throws: std::range_error is thrown if the size of any single part
  223. can not fit in an unsigned 32 bit integer.
  224. The encoding is compatible with that used by the CZMQ function
  225. zmsg_encode(), see https://rfc.zeromq.org/spec/50/.
  226. Each part consists of a size followed by the data.
  227. These are placed contiguously into the output message. A part of
  228. size less than 255 bytes will have a single byte size value.
  229. Larger parts will have a five byte size value with the first byte
  230. set to 0xFF and the remaining four bytes holding the size of the
  231. part's data.
  232. */
  233. template<class Range
  234. #ifndef ZMQ_CPP11_PARTIAL
  235. ,
  236. typename = typename std::enable_if<
  237. detail::is_range<Range>::value
  238. && (std::is_same<detail::range_value_t<Range>, message_t>::value
  239. || detail::is_buffer<detail::range_value_t<Range>>::value)>::type
  240. #endif
  241. >
  242. message_t encode(const Range &parts)
  243. {
  244. size_t mmsg_size = 0;
  245. // First pass check sizes
  246. for (const auto &part : parts) {
  247. const size_t part_size = part.size();
  248. if (part_size > (std::numeric_limits<std::uint32_t>::max)()) {
  249. // Size value must fit into uint32_t.
  250. throw std::range_error("Invalid size, message part too large");
  251. }
  252. const size_t count_size =
  253. part_size < (std::numeric_limits<std::uint8_t>::max)() ? 1 : 5;
  254. mmsg_size += part_size + count_size;
  255. }
  256. message_t encoded(mmsg_size);
  257. unsigned char *buf = encoded.data<unsigned char>();
  258. for (const auto &part : parts) {
  259. const uint32_t part_size = static_cast<uint32_t>(part.size());
  260. const unsigned char *part_data =
  261. static_cast<const unsigned char *>(part.data());
  262. if (part_size < (std::numeric_limits<std::uint8_t>::max)()) {
  263. // small part
  264. *buf++ = (unsigned char) part_size;
  265. } else {
  266. // big part
  267. *buf++ = (std::numeric_limits<uint8_t>::max)();
  268. detail::write_network_order(buf, part_size);
  269. buf += sizeof(part_size);
  270. }
  271. std::memcpy(buf, part_data, part_size);
  272. buf += part_size;
  273. }
  274. assert(static_cast<size_t>(buf - encoded.data<unsigned char>()) == mmsg_size);
  275. return encoded;
  276. }
  277. /* Decode an encoded message to multiple parts.
  278. The given output iterator must be a ForwardIterator to a container
  279. holding zmq::message_t such as a zmq::multipart_t or various STL
  280. containers.
  281. Returns the ForwardIterator advanced once past the last decoded
  282. part.
  283. Throws: a std::out_of_range is thrown if the encoded part sizes
  284. lead to exceeding the message data bounds.
  285. The decoding assumes the message is encoded in the manner
  286. performed by zmq::encode(), see https://rfc.zeromq.org/spec/50/.
  287. */
  288. template<class OutputIt> OutputIt decode(const message_t &encoded, OutputIt out)
  289. {
  290. const unsigned char *source = encoded.data<unsigned char>();
  291. const unsigned char *const limit = source + encoded.size();
  292. while (source < limit) {
  293. size_t part_size = *source++;
  294. if (part_size == (std::numeric_limits<std::uint8_t>::max)()) {
  295. if (static_cast<size_t>(limit - source) < sizeof(uint32_t)) {
  296. throw std::out_of_range(
  297. "Malformed encoding, overflow in reading size");
  298. }
  299. part_size = detail::read_u32_network_order(source);
  300. // the part size is allowed to be less than 0xFF
  301. source += sizeof(uint32_t);
  302. }
  303. if (static_cast<size_t>(limit - source) < part_size) {
  304. throw std::out_of_range("Malformed encoding, overflow in reading part");
  305. }
  306. *out = message_t(source, part_size);
  307. ++out;
  308. source += part_size;
  309. }
  310. assert(source == limit);
  311. return out;
  312. }
  313. #endif
  314. #ifdef ZMQ_HAS_RVALUE_REFS
  315. /*
  316. This class handles multipart messaging. It is the C++ equivalent of zmsg.h,
  317. which is part of CZMQ (the high-level C binding). Furthermore, it is a major
  318. improvement compared to zmsg.hpp, which is part of the examples in the ØMQ
  319. Guide. Unnecessary copying is avoided by using move semantics to efficiently
  320. add/remove parts.
  321. */
  322. class multipart_t
  323. {
  324. private:
  325. std::deque<message_t> m_parts;
  326. public:
  327. typedef std::deque<message_t>::value_type value_type;
  328. typedef std::deque<message_t>::iterator iterator;
  329. typedef std::deque<message_t>::const_iterator const_iterator;
  330. typedef std::deque<message_t>::reverse_iterator reverse_iterator;
  331. typedef std::deque<message_t>::const_reverse_iterator const_reverse_iterator;
  332. // Default constructor
  333. multipart_t() {}
  334. // Construct from socket receive
  335. multipart_t(socket_ref socket) { recv(socket); }
  336. // Construct from memory block
  337. multipart_t(const void *src, size_t size) { addmem(src, size); }
  338. // Construct from string
  339. multipart_t(const std::string &string) { addstr(string); }
  340. // Construct from message part
  341. multipart_t(message_t &&message) { add(std::move(message)); }
  342. // Move constructor
  343. multipart_t(multipart_t &&other) ZMQ_NOTHROW { m_parts = std::move(other.m_parts); }
  344. // Move assignment operator
  345. multipart_t &operator=(multipart_t &&other) ZMQ_NOTHROW
  346. {
  347. m_parts = std::move(other.m_parts);
  348. return *this;
  349. }
  350. // Destructor
  351. virtual ~multipart_t() { clear(); }
  352. message_t &operator[](size_t n) { return m_parts[n]; }
  353. const message_t &operator[](size_t n) const { return m_parts[n]; }
  354. message_t &at(size_t n) { return m_parts.at(n); }
  355. const message_t &at(size_t n) const { return m_parts.at(n); }
  356. iterator begin() { return m_parts.begin(); }
  357. const_iterator begin() const { return m_parts.begin(); }
  358. const_iterator cbegin() const { return m_parts.cbegin(); }
  359. reverse_iterator rbegin() { return m_parts.rbegin(); }
  360. const_reverse_iterator rbegin() const { return m_parts.rbegin(); }
  361. iterator end() { return m_parts.end(); }
  362. const_iterator end() const { return m_parts.end(); }
  363. const_iterator cend() const { return m_parts.cend(); }
  364. reverse_iterator rend() { return m_parts.rend(); }
  365. const_reverse_iterator rend() const { return m_parts.rend(); }
  366. // Delete all parts
  367. void clear() { m_parts.clear(); }
  368. // Get number of parts
  369. size_t size() const { return m_parts.size(); }
  370. // Check if number of parts is zero
  371. bool empty() const { return m_parts.empty(); }
  372. // Receive multipart message from socket
  373. bool recv(socket_ref socket, int flags = 0)
  374. {
  375. clear();
  376. bool more = true;
  377. while (more) {
  378. message_t message;
  379. #ifdef ZMQ_CPP11
  380. if (!socket.recv(message, static_cast<recv_flags>(flags)))
  381. return false;
  382. #else
  383. if (!socket.recv(&message, flags))
  384. return false;
  385. #endif
  386. more = message.more();
  387. add(std::move(message));
  388. }
  389. return true;
  390. }
  391. // Send multipart message to socket
  392. bool send(socket_ref socket, int flags = 0)
  393. {
  394. flags &= ~(ZMQ_SNDMORE);
  395. bool more = size() > 0;
  396. while (more) {
  397. message_t message = pop();
  398. more = size() > 0;
  399. #ifdef ZMQ_CPP11
  400. if (!socket.send(message, static_cast<send_flags>(
  401. (more ? ZMQ_SNDMORE : 0) | flags)))
  402. return false;
  403. #else
  404. if (!socket.send(message, (more ? ZMQ_SNDMORE : 0) | flags))
  405. return false;
  406. #endif
  407. }
  408. clear();
  409. return true;
  410. }
  411. // Concatenate other multipart to front
  412. void prepend(multipart_t &&other)
  413. {
  414. while (!other.empty())
  415. push(other.remove());
  416. }
  417. // Concatenate other multipart to back
  418. void append(multipart_t &&other)
  419. {
  420. while (!other.empty())
  421. add(other.pop());
  422. }
  423. // Push memory block to front
  424. void pushmem(const void *src, size_t size)
  425. {
  426. m_parts.push_front(message_t(src, size));
  427. }
  428. // Push memory block to back
  429. void addmem(const void *src, size_t size)
  430. {
  431. m_parts.push_back(message_t(src, size));
  432. }
  433. // Push string to front
  434. void pushstr(const std::string &string)
  435. {
  436. m_parts.push_front(message_t(string.data(), string.size()));
  437. }
  438. // Push string to back
  439. void addstr(const std::string &string)
  440. {
  441. m_parts.push_back(message_t(string.data(), string.size()));
  442. }
  443. // Push type (fixed-size) to front
  444. template<typename T> void pushtyp(const T &type)
  445. {
  446. static_assert(!std::is_same<T, std::string>::value,
  447. "Use pushstr() instead of pushtyp<std::string>()");
  448. m_parts.push_front(message_t(&type, sizeof(type)));
  449. }
  450. // Push type (fixed-size) to back
  451. template<typename T> void addtyp(const T &type)
  452. {
  453. static_assert(!std::is_same<T, std::string>::value,
  454. "Use addstr() instead of addtyp<std::string>()");
  455. m_parts.push_back(message_t(&type, sizeof(type)));
  456. }
  457. // Push message part to front
  458. void push(message_t &&message) { m_parts.push_front(std::move(message)); }
  459. // Push message part to back
  460. void add(message_t &&message) { m_parts.push_back(std::move(message)); }
  461. // Alias to allow std::back_inserter()
  462. void push_back(message_t &&message) { m_parts.push_back(std::move(message)); }
  463. // Pop string from front
  464. std::string popstr()
  465. {
  466. std::string string(m_parts.front().data<char>(), m_parts.front().size());
  467. m_parts.pop_front();
  468. return string;
  469. }
  470. // Pop type (fixed-size) from front
  471. template<typename T> T poptyp()
  472. {
  473. static_assert(!std::is_same<T, std::string>::value,
  474. "Use popstr() instead of poptyp<std::string>()");
  475. if (sizeof(T) != m_parts.front().size())
  476. throw std::runtime_error(
  477. "Invalid type, size does not match the message size");
  478. T type = *m_parts.front().data<T>();
  479. m_parts.pop_front();
  480. return type;
  481. }
  482. // Pop message part from front
  483. message_t pop()
  484. {
  485. message_t message = std::move(m_parts.front());
  486. m_parts.pop_front();
  487. return message;
  488. }
  489. // Pop message part from back
  490. message_t remove()
  491. {
  492. message_t message = std::move(m_parts.back());
  493. m_parts.pop_back();
  494. return message;
  495. }
  496. // get message part from front
  497. const message_t &front() { return m_parts.front(); }
  498. // get message part from back
  499. const message_t &back() { return m_parts.back(); }
  500. // Get pointer to a specific message part
  501. const message_t *peek(size_t index) const { return &m_parts[index]; }
  502. // Get a string copy of a specific message part
  503. std::string peekstr(size_t index) const
  504. {
  505. std::string string(m_parts[index].data<char>(), m_parts[index].size());
  506. return string;
  507. }
  508. // Peek type (fixed-size) from front
  509. template<typename T> T peektyp(size_t index) const
  510. {
  511. static_assert(!std::is_same<T, std::string>::value,
  512. "Use peekstr() instead of peektyp<std::string>()");
  513. if (sizeof(T) != m_parts[index].size())
  514. throw std::runtime_error(
  515. "Invalid type, size does not match the message size");
  516. T type = *m_parts[index].data<T>();
  517. return type;
  518. }
  519. // Create multipart from type (fixed-size)
  520. template<typename T> static multipart_t create(const T &type)
  521. {
  522. multipart_t multipart;
  523. multipart.addtyp(type);
  524. return multipart;
  525. }
  526. // Copy multipart
  527. multipart_t clone() const
  528. {
  529. multipart_t multipart;
  530. for (size_t i = 0; i < size(); i++)
  531. multipart.addmem(m_parts[i].data(), m_parts[i].size());
  532. return multipart;
  533. }
  534. // Dump content to string
  535. std::string str() const
  536. {
  537. std::stringstream ss;
  538. for (size_t i = 0; i < m_parts.size(); i++) {
  539. const unsigned char *data = m_parts[i].data<unsigned char>();
  540. size_t size = m_parts[i].size();
  541. // Dump the message as text or binary
  542. bool isText = true;
  543. for (size_t j = 0; j < size; j++) {
  544. if (data[j] < 32 || data[j] > 127) {
  545. isText = false;
  546. break;
  547. }
  548. }
  549. ss << "\n[" << std::dec << std::setw(3) << std::setfill('0') << size
  550. << "] ";
  551. if (size >= 1000) {
  552. ss << "... (too big to print)";
  553. continue;
  554. }
  555. for (size_t j = 0; j < size; j++) {
  556. if (isText)
  557. ss << static_cast<char>(data[j]);
  558. else
  559. ss << std::hex << std::setw(2) << std::setfill('0')
  560. << static_cast<short>(data[j]);
  561. }
  562. }
  563. return ss.str();
  564. }
  565. // Check if equal to other multipart
  566. bool equal(const multipart_t *other) const ZMQ_NOTHROW
  567. {
  568. return *this == *other;
  569. }
  570. bool operator==(const multipart_t &other) const ZMQ_NOTHROW
  571. {
  572. if (size() != other.size())
  573. return false;
  574. for (size_t i = 0; i < size(); i++)
  575. if (at(i) != other.at(i))
  576. return false;
  577. return true;
  578. }
  579. bool operator!=(const multipart_t &other) const ZMQ_NOTHROW
  580. {
  581. return !(*this == other);
  582. }
  583. #ifdef ZMQ_CPP11
  584. // Return single part message_t encoded from this multipart_t.
  585. message_t encode() const { return zmq::encode(*this); }
  586. // Decode encoded message into multiple parts and append to self.
  587. void decode_append(const message_t &encoded)
  588. {
  589. zmq::decode(encoded, std::back_inserter(*this));
  590. }
  591. // Return a new multipart_t containing the decoded message_t.
  592. static multipart_t decode(const message_t &encoded)
  593. {
  594. multipart_t tmp;
  595. zmq::decode(encoded, std::back_inserter(tmp));
  596. return tmp;
  597. }
  598. #endif
  599. private:
  600. // Disable implicit copying (moving is more efficient)
  601. multipart_t(const multipart_t &other) ZMQ_DELETED_FUNCTION;
  602. void operator=(const multipart_t &other) ZMQ_DELETED_FUNCTION;
  603. }; // class multipart_t
  604. inline std::ostream &operator<<(std::ostream &os, const multipart_t &msg)
  605. {
  606. return os << msg.str();
  607. }
  608. #endif // ZMQ_HAS_RVALUE_REFS
  609. #if defined(ZMQ_BUILD_DRAFT_API) && defined(ZMQ_CPP11) && defined(ZMQ_HAVE_POLLER)
  610. class active_poller_t
  611. {
  612. public:
  613. active_poller_t() = default;
  614. ~active_poller_t() = default;
  615. active_poller_t(const active_poller_t &) = delete;
  616. active_poller_t &operator=(const active_poller_t &) = delete;
  617. active_poller_t(active_poller_t &&src) = default;
  618. active_poller_t &operator=(active_poller_t &&src) = default;
  619. using handler_type = std::function<void(event_flags)>;
  620. void add(zmq::socket_ref socket, event_flags events, handler_type handler)
  621. {
  622. const poller_ref_t ref{socket};
  623. if (!handler)
  624. throw std::invalid_argument("null handler in active_poller_t::add (socket)");
  625. auto ret = handlers.emplace(
  626. ref, std::make_shared<handler_type>(std::move(handler)));
  627. if (!ret.second)
  628. throw error_t(EINVAL); // already added
  629. try {
  630. base_poller.add(socket, events, ret.first->second.get());
  631. need_rebuild = true;
  632. }
  633. catch (...) {
  634. // rollback
  635. handlers.erase(ref);
  636. throw;
  637. }
  638. }
  639. void add(fd_t fd, event_flags events, handler_type handler)
  640. {
  641. const poller_ref_t ref{fd};
  642. if (!handler)
  643. throw std::invalid_argument("null handler in active_poller_t::add (fd)");
  644. auto ret = handlers.emplace(
  645. ref, std::make_shared<handler_type>(std::move(handler)));
  646. if (!ret.second)
  647. throw error_t(EINVAL); // already added
  648. try {
  649. base_poller.add(fd, events, ret.first->second.get());
  650. need_rebuild = true;
  651. }
  652. catch (...) {
  653. // rollback
  654. handlers.erase(ref);
  655. throw;
  656. }
  657. }
  658. void remove(zmq::socket_ref socket)
  659. {
  660. base_poller.remove(socket);
  661. handlers.erase(socket);
  662. need_rebuild = true;
  663. }
  664. void remove(fd_t fd)
  665. {
  666. base_poller.remove(fd);
  667. handlers.erase(fd);
  668. need_rebuild = true;
  669. }
  670. void modify(zmq::socket_ref socket, event_flags events)
  671. {
  672. base_poller.modify(socket, events);
  673. }
  674. void modify(fd_t fd, event_flags events)
  675. {
  676. base_poller.modify(fd, events);
  677. }
  678. size_t wait(std::chrono::milliseconds timeout)
  679. {
  680. if (need_rebuild) {
  681. poller_events.resize(handlers.size());
  682. poller_handlers.clear();
  683. poller_handlers.reserve(handlers.size());
  684. for (const auto &handler : handlers) {
  685. poller_handlers.push_back(handler.second);
  686. }
  687. need_rebuild = false;
  688. }
  689. const auto count = base_poller.wait_all(poller_events, timeout);
  690. std::for_each(poller_events.begin(),
  691. poller_events.begin() + static_cast<ptrdiff_t>(count),
  692. [](decltype(base_poller)::event_type &event) {
  693. assert(event.user_data != nullptr);
  694. (*event.user_data)(event.events);
  695. });
  696. return count;
  697. }
  698. ZMQ_NODISCARD bool empty() const noexcept { return handlers.empty(); }
  699. size_t size() const noexcept { return handlers.size(); }
  700. private:
  701. bool need_rebuild{false};
  702. poller_t<handler_type> base_poller{};
  703. std::unordered_map<zmq::poller_ref_t, std::shared_ptr<handler_type>> handlers{};
  704. std::vector<decltype(base_poller)::event_type> poller_events{};
  705. std::vector<std::shared_ptr<handler_type>> poller_handlers{};
  706. }; // class active_poller_t
  707. #endif // defined(ZMQ_BUILD_DRAFT_API) && defined(ZMQ_CPP11) && defined(ZMQ_HAVE_POLLER)
  708. } // namespace zmq
  709. #endif // __ZMQ_ADDON_HPP_INCLUDED__