17 namespace serialization
40 template <
typename ZMQ_SOCKET_TYPE>
43 const size_t max_packet_len = 0)
46 if (!buf)
throw std::bad_alloc();
48 buf->WriteObject(&obj);
51 throw std::runtime_error(
52 "[mrpt_send_to_zmq] Serialized object has 0 bytes, which probably "
53 "means something went wrong...");
57 :
static_cast<unsigned int>(ceil(
double(nBytes) / max_packet_len));
58 for (
unsigned int iPkt = 0; iPkt < nPkts; ++iPkt)
61 mrpt::serialization::internal::TFreeFnDataForZMQ* fd =
62 new mrpt::serialization::internal::TFreeFnDataForZMQ();
63 if (!fd)
throw std::bad_alloc();
68 void* pkt_data =
reinterpret_cast<char*
>(fd->buf->getRawBufferData()) +
69 max_packet_len * iPkt;
70 size_t nBytesThisPkt = nBytes - max_packet_len * iPkt;
71 if (max_packet_len != 0 && nBytesThisPkt > max_packet_len)
72 nBytesThisPkt = max_packet_len;
75 if (0 != zmq_msg_init_data(
76 &message, pkt_data, nBytesThisPkt,
78 throw std::runtime_error(
79 "[mrpt_send_to_zmq] Error in zmq_msg_init_data()");
82 zmq_msg_send(&message, zmq_socket, fd->do_free ? 0 : ZMQ_SNDMORE);
83 if (0 != zmq_msg_close(&message))
84 throw std::runtime_error(
85 "[mrpt_send_to_zmq] Error in zmq_msg_close()");
86 if (sent_size !=
static_cast<int>(nBytesThisPkt))
87 throw std::runtime_error(
88 "[mrpt_send_to_zmq] Error in zmq_msg_send()");
96 template <
typename ZMQ_SOCKET_TYPE,
typename VECTOR_MSG_T>
98 ZMQ_SOCKET_TYPE zmq_socket, VECTOR_MSG_T& out_lst_msgs,
100 size_t* rx_obj_length_in_bytes)
102 if (rx_obj_length_in_bytes) *rx_obj_length_in_bytes = 0;
103 out_lst_msgs.clear();
106 size_t more_size =
sizeof(more);
110 zmq_msg_t* msg =
new zmq_msg_t();
111 if (0 != zmq_msg_init(msg))
return false;
112 out_lst_msgs.push_back(msg);
114 int rc = zmq_msg_recv(msg, zmq_socket, dont_wait ? ZMQ_DONTWAIT : 0);
115 if (rc == -1)
return false;
117 rc = zmq_getsockopt(zmq_socket, ZMQ_RCVMORE, &more, &more_size);
118 if (rc != 0)
return false;
120 if (out_lst_msgs.size() == 1 && !more)
123 if (rx_obj_length_in_bytes)
124 *rx_obj_length_in_bytes = zmq_msg_size(msg);
128 if (out_lst_msgs.size() > 1)
130 for (
size_t i = 0; i < out_lst_msgs.size(); i++)
132 target_buf.WriteBuffer(
133 zmq_msg_data(out_lst_msgs[i]), zmq_msg_size(out_lst_msgs[i]));
135 if (rx_obj_length_in_bytes)
144 template <
typename VECTOR_MSG_T>
147 for (
size_t i = 0; i < lst_msgs.size(); ++i)
149 zmq_msg_close(lst_msgs[i]);
175 template <
typename ZMQ_SOCKET_TYPE>
177 ZMQ_SOCKET_TYPE zmq_socket,
bool dont_wait =
false,
178 size_t* rx_obj_length_in_bytes =
nullptr)
182 std::vector<zmq_msg_t*> lst_msgs_to_close;
184 zmq_socket, lst_msgs_to_close, target_buf, dont_wait,
185 rx_obj_length_in_bytes))
188 obj = target_buf.ReadObject();
203 template <
typename ZMQ_SOCKET_TYPE>
205 ZMQ_SOCKET_TYPE zmq_socket,
207 size_t* rx_obj_length_in_bytes =
nullptr)
210 std::vector<zmq_msg_t*> lst_msgs_to_close;
212 zmq_socket, lst_msgs_to_close, target_buf, dont_wait,
213 rx_obj_length_in_bytes))
216 target_buf.ReadObject(&target_object);