Line data Source code
1 :
2 : /**
3 : *
4 : * @file ZmqSubscriber.cpp ZmqSubscriber messaging class definitions
5 : *
6 : * This is part of the DUNE DAQ Application Framework, copyright 2020.
7 : * Licensing/copyright details are in the COPYING file that you should have
8 : * received with this code.
9 : */
10 :
11 : #include "CallbackAdapter.hpp"
12 : #include "ipm/Subscriber.hpp"
13 : #include "ipm/ZmqContext.hpp"
14 :
15 : #include "logging/Logging.hpp"
16 :
17 : #include <set>
18 : #include <string>
19 : #include <vector>
20 :
21 : namespace dunedaq::ipm {
22 :
23 : class ZmqSubscriber : public Subscriber
24 : {
25 : public:
26 13 : ZmqSubscriber()
27 13 : : m_socket(ZmqContext::instance().GetContext(), zmq::socket_type::sub)
28 : {
29 13 : }
30 :
31 26 : ~ZmqSubscriber()
32 13 : {
33 13 : unregister_callback();
34 : // Probably (cpp)zmq does this in the socket dtor anyway, but I guess it doesn't hurt to be explicit
35 13 : if (!m_connection_strings.empty() && m_socket_connected) {
36 10 : m_socket_connected = false;
37 25 : for (auto& conn_string : m_connection_strings) {
38 15 : try {
39 15 : m_socket.disconnect(conn_string);
40 0 : } catch (zmq::error_t const& err) {
41 0 : ers::error(ZmqOperationError(
42 0 : ERS_HERE, m_connection_info.connection_name, "disconnect", "receive", err.what(), conn_string));
43 0 : }
44 : }
45 : }
46 13 : m_socket.close();
47 26 : }
48 :
49 13 : std::string connect_for_receives(const ConnectionInfo& connection_info) override
50 : {
51 13 : m_connection_info = connection_info;
52 13 : std::set<std::string> new_connection_strings;
53 13 : if (connection_info.connection_string != "") {
54 5 : new_connection_strings.insert(connection_info.connection_string);
55 : }
56 :
57 26 : for (auto& conn_string : connection_info.connection_strings) {
58 13 : if (m_connection_strings.count(conn_string) == 0)
59 13 : new_connection_strings.insert(conn_string);
60 : }
61 :
62 13 : if (m_connection_strings.size() == 0 && new_connection_strings.size() == 0) {
63 0 : throw ZmqOperationError(ERS_HERE,
64 0 : m_connection_info.connection_name,
65 : "resolve connections",
66 : "receive",
67 : "No valid connection strings passed",
68 0 : "");
69 : }
70 :
71 13 : if (!m_socket_connected) {
72 11 : try {
73 11 : m_socket.set(zmq::sockopt::rcvtimeo, 0); // Return immediately if we can't receive
74 0 : } catch (zmq::error_t const& err) {
75 0 : throw ZmqOperationError(ERS_HERE,
76 0 : m_connection_info.connection_name,
77 : "set timeout",
78 : "receive",
79 0 : err.what(),
80 0 : *m_connection_strings.begin());
81 0 : }
82 : }
83 31 : for (auto& conn_string : new_connection_strings) {
84 18 : try {
85 18 : TLOG_DEBUG(TLVL_CONNECTIONSTRING)
86 18 : << m_connection_info.connection_name << ": Connection String is " << conn_string;
87 18 : m_socket.connect(conn_string);
88 15 : m_connection_strings.insert(conn_string);
89 3 : } catch (zmq::error_t const& err) {
90 3 : ers::error(ZmqOperationError(
91 6 : ERS_HERE, m_connection_info.connection_name, "connect", "receive", err.what(), conn_string));
92 3 : }
93 : }
94 13 : m_socket_connected = true;
95 13 : m_callback_adapter.set_receiver(this);
96 :
97 13 : if (m_connection_strings.size() > 0) {
98 10 : return *m_connection_strings.begin();
99 : }
100 3 : return {};
101 13 : }
102 :
103 63 : bool can_receive() const noexcept override { return m_socket_connected; }
104 :
105 12 : void subscribe(std::string const& topic) override
106 : {
107 12 : try {
108 12 : m_socket.set(zmq::sockopt::subscribe, topic);
109 0 : } catch (zmq::error_t const& err) {
110 0 : throw ZmqSubscribeError(ERS_HERE, m_connection_info.connection_name, err.what(), topic);
111 0 : }
112 12 : }
113 2 : void unsubscribe(std::string const& topic) override
114 : {
115 2 : try {
116 2 : m_socket.set(zmq::sockopt::unsubscribe, topic);
117 0 : } catch (zmq::error_t const& err) {
118 0 : throw ZmqUnsubscribeError(ERS_HERE, m_connection_info.connection_name, err.what(), topic);
119 0 : }
120 2 : }
121 :
122 6 : bool data_pending() override
123 : {
124 6 : try {
125 6 : auto events = m_socket.get(zmq::sockopt::events);
126 6 : return (events & ZMQ_POLLIN) != 0;
127 0 : } catch (zmq::error_t const& err) {
128 0 : ers::error(ZmqOperationError(ERS_HERE,
129 0 : m_connection_info.connection_name,
130 : "get events sockopt",
131 : "data_pending",
132 0 : err.what(),
133 0 : *m_connection_strings.begin()));
134 0 : }
135 0 : return false;
136 : }
137 :
138 1 : void register_callback(std::function<void(Response&)> callback) override
139 : {
140 1 : m_callback_adapter.set_callback(callback);
141 1 : }
142 14 : void unregister_callback() override { m_callback_adapter.clear_callback(); }
143 :
144 : protected:
145 54 : Receiver::Response receive_(const duration_t& timeout, bool no_tmoexcept_mode) override
146 : {
147 54 : Receiver::Response output;
148 54 : zmq::message_t hdr;
149 54 : zmq::message_t msg;
150 54 : zmq::recv_result_t res{};
151 :
152 54 : auto start_time = std::chrono::steady_clock::now();
153 2165 : do {
154 :
155 2165 : try {
156 2165 : TLOG_DEBUG(TLVL_ZMQSUBSCRIBER_RECV_HDR) << m_connection_info.connection_name << ": Going to receive header";
157 2165 : res = m_socket.recv(hdr);
158 2165 : TLOG_DEBUG(TLVL_ZMQSUBSCRIBER_RECV_HDR_2)
159 0 : << m_connection_info.connection_name << ": Recv res=" << res.value_or(0)
160 2165 : << " for header (hdr.size() == " << hdr.size() << ")";
161 0 : } catch (zmq::error_t const& err) {
162 0 : throw ZmqReceiveError(ERS_HERE, m_connection_info.connection_name, err.what(), "header");
163 0 : }
164 2165 : if (res || hdr.more()) {
165 16 : TLOG_DEBUG(TLVL_ZMQSUBSCRIBER_RECV_DATA) << m_connection_info.connection_name << ": Going to receive data";
166 16 : output.metadata.resize(hdr.size());
167 16 : memcpy(&output.metadata[0], hdr.data(), hdr.size());
168 :
169 : // ZMQ guarantees that the entire message has arrived
170 :
171 16 : try {
172 16 : res = m_socket.recv(msg);
173 0 : } catch (zmq::error_t const& err) {
174 0 : throw ZmqReceiveError(ERS_HERE, m_connection_info.connection_name, err.what(), "data");
175 0 : }
176 16 : TLOG_DEBUG(TLVL_ZMQSUBSCRIBER_RECV_DATA_2)
177 0 : << m_connection_info.connection_name << ": Recv res=" << res.value_or(0)
178 16 : << " for data (msg.size() == " << msg.size() << ")";
179 16 : output.data.resize(msg.size());
180 16 : memcpy(&output.data[0], msg.data(), msg.size());
181 2149 : } else if (timeout > duration_t::zero()) {
182 2127 : usleep(1000);
183 : }
184 2165 : } while (std::chrono::duration_cast<duration_t>(std::chrono::steady_clock::now() - start_time) < timeout &&
185 2180 : res.value_or(0) == 0);
186 :
187 54 : if (res.value_or(0) == 0 && !no_tmoexcept_mode) {
188 38 : throw ReceiveTimeoutExpired(ERS_HERE, m_connection_info.connection_name, timeout.count());
189 : }
190 :
191 16 : TLOG_DEBUG(TLVL_ZMQSUBSCRIBER_RECV_END)
192 0 : << m_connection_info.connection_name << ": Returning output with metadata size " << output.metadata.size()
193 16 : << " and data size " << output.data.size();
194 16 : return output;
195 130 : }
196 :
197 : private:
198 : zmq::socket_t m_socket;
199 : std::set<std::string> m_connection_strings{};
200 : bool m_socket_connected{ false };
201 : CallbackAdapter m_callback_adapter;
202 : };
203 : } // namespace dunedaq::ipm
204 :
205 13 : DEFINE_DUNE_IPM_RECEIVER(dunedaq::ipm::ZmqSubscriber)
|
