Line data Source code
1 : /**
2 : *
3 : * @file ZmqPublisher.cpp ZmqPublisher messaging class definitions
4 : *
5 : * This is part of the DUNE DAQ Application Framework, copyright 2020.
6 : * Licensing/copyright details are in the COPYING file that you should have
7 : * received with this code.
8 : */
9 :
10 : #include "ipm/Sender.hpp"
11 : #include "ipm/ZmqContext.hpp"
12 :
13 : #include "logging/Logging.hpp"
14 : #include "utilities/ZmqUri.hpp"
15 : #include "zmq.hpp"
16 :
17 : #include <string>
18 : #include <vector>
19 :
20 : namespace dunedaq::ipm {
21 : class ZmqPublisher : public Sender
22 : {
23 : public:
24 12 : ZmqPublisher()
25 12 : : m_socket(ZmqContext::instance().GetContext(), zmq::socket_type::pub)
26 : {
27 12 : }
28 :
29 24 : ~ZmqPublisher()
30 12 : {
31 : // Probably (cpp)zmq does this in the socket dtor anyway, but I guess it doesn't hurt to be explicit
32 12 : if (m_connection_info.connection_string != "" && m_socket_connected) {
33 10 : try {
34 20 : TLOG_DEBUG(TLVL_ZMQPUBLISHER_DESTRUCTOR)
35 10 : << m_connection_info.connection_name << ": Disconnecting socket from " << m_connection_info.connection_string;
36 :
37 10 : m_socket.unbind(m_connection_info.connection_string);
38 10 : m_socket_connected = false;
39 0 : } catch (zmq::error_t const& err) {
40 0 : ers::error(ZmqOperationError(ERS_HERE,
41 0 : m_connection_info.connection_name,
42 : "unbind",
43 : "send",
44 0 : err.what(),
45 0 : m_connection_info.connection_string));
46 0 : }
47 : }
48 12 : m_socket.close();
49 24 : }
50 :
51 29 : bool can_send() const noexcept override { return m_socket_connected; }
52 13 : std::string connect_for_sends(const ConnectionInfo& connection_info) override
53 : {
54 13 : m_connection_info = connection_info;
55 13 : try {
56 13 : m_socket.set(zmq::sockopt::sndtimeo, 0); // Return immediately if we can't send
57 0 : } catch (zmq::error_t const& err) {
58 0 : throw ZmqOperationError(ERS_HERE,
59 0 : m_connection_info.connection_name,
60 : "set timeout",
61 : "send",
62 0 : err.what(),
63 0 : m_connection_info.connection_string);
64 0 : }
65 :
66 13 : auto hwm = connection_info.capacity;
67 13 : if (hwm > 0) {
68 0 : try {
69 0 : m_socket.set(zmq::sockopt::sndhwm, hwm);
70 0 : } catch (zmq::error_t const& err) {
71 0 : throw ZmqOperationError(ERS_HERE,
72 0 : m_connection_info.connection_name,
73 : "set hwm",
74 : "send",
75 0 : err.what(),
76 0 : m_connection_info.connection_string);
77 0 : }
78 : }
79 :
80 13 : std::vector<std::string> resolved;
81 13 : try {
82 13 : utilities::ZmqUri this_uri(m_connection_info.connection_string);
83 12 : resolved = this_uri.get_uri_ip_addresses();
84 13 : } catch (utilities::InvalidUri const& err) {
85 1 : throw ZmqOperationError(ERS_HERE,
86 1 : m_connection_info.connection_name,
87 : "resolve connection_string",
88 : "send",
89 : "An invalid URI was passed",
90 : m_connection_info.connection_string,
91 2 : err);
92 1 : }
93 :
94 12 : if (resolved.size() == 0) {
95 1 : throw ZmqOperationError(ERS_HERE,
96 1 : m_connection_info.connection_name,
97 : "resolve connection_string",
98 : "send",
99 : "Unable to resolve connection_string",
100 2 : m_connection_info.connection_string);
101 : }
102 12 : for (auto& connection_string : resolved) {
103 11 : TLOG_DEBUG(TLVL_CONNECTIONSTRING) << m_connection_info.connection_name << ": Connection String is "
104 11 : << connection_string;
105 11 : try {
106 11 : m_socket.bind(connection_string);
107 10 : m_connection_info.connection_string = m_socket.get(zmq::sockopt::last_endpoint);
108 10 : m_socket_connected = true;
109 10 : break;
110 1 : } catch (zmq::error_t const& err) {
111 1 : ers::error(ZmqOperationError(
112 2 : ERS_HERE, m_connection_info.connection_name, "bind", "send", err.what(), connection_string));
113 1 : }
114 : }
115 11 : if (!m_socket_connected) {
116 1 : throw ZmqOperationError(ERS_HERE,
117 1 : m_connection_info.connection_name,
118 : "bind",
119 : "send",
120 : "Bind failed for all resolved connection strings",
121 2 : "");
122 : }
123 :
124 20 : return m_connection_info.connection_string;
125 13 : }
126 :
127 : protected:
128 20 : bool send_(const void* message,
129 : int N,
130 : const duration_t& timeout,
131 : std::string const& topic,
132 : bool no_tmoexcept_mode) override
133 : {
134 20 : TLOG_DEBUG(TLVL_ZMQPUBLISHER_SEND_START)
135 20 : << m_connection_info.connection_name << ": Starting send of " << N << " bytes";
136 20 : auto start_time = std::chrono::steady_clock::now();
137 20 : zmq::send_result_t res{};
138 20 : do {
139 :
140 20 : zmq::message_t topic_msg(topic.c_str(), topic.size());
141 20 : try {
142 20 : res = m_socket.send(topic_msg, zmq::send_flags::sndmore);
143 0 : } catch (zmq::error_t const& err) {
144 0 : throw ZmqSendError(ERS_HERE, m_connection_info.connection_name, err.what(), topic.size(), topic);
145 0 : }
146 :
147 20 : if (!res || res != topic.size()) {
148 0 : TLOG_DEBUG(TLVL_ZMQPUBLISHER_SEND_ERR) << m_connection_info.connection_name << ": Unable to send message";
149 0 : continue;
150 0 : }
151 :
152 20 : zmq::message_t msg(message, N);
153 20 : try {
154 20 : res = m_socket.send(msg, zmq::send_flags::none);
155 0 : } catch (zmq::error_t const& err) {
156 0 : throw ZmqSendError(ERS_HERE, m_connection_info.connection_name, err.what(), N, topic);
157 0 : }
158 :
159 20 : if (!res && timeout > duration_t::zero()) {
160 0 : usleep(1000);
161 : }
162 60 : } while (std::chrono::duration_cast<duration_t>(std::chrono::steady_clock::now() - start_time) < timeout && !res);
163 :
164 20 : if (!res && !no_tmoexcept_mode) {
165 0 : throw SendTimeoutExpired(ERS_HERE, m_connection_info.connection_name, timeout.count());
166 : }
167 :
168 20 : TLOG_DEBUG(TLVL_ZMQPUBLISHER_SEND_END)
169 20 : << m_connection_info.connection_name << ": Completed send of " << N << " bytes";
170 20 : return res && res == N;
171 : }
172 :
173 : private:
174 : zmq::socket_t m_socket;
175 : bool m_socket_connected{ false };
176 : };
177 :
178 : } // namespace dunedaq::ipm
179 :
180 12 : DEFINE_DUNE_IPM_SENDER(dunedaq::ipm::ZmqPublisher)
|
