GCC Code Coverage Report

Directory: ./
File: src/phoenix_zmq.cpp
Date: 2026-06-23 10:44:39
Exec Total Coverage
Lines: 61 71 85.9%
Functions: 10 11 90.9%
Branches: 37 66 56.1%
Line Branch Exec Source
1 /***************************************
2 Auteur : Pierre Aubert
3 Mail : pierre.aubert@lapp.in2p3.fr
4 Licence : CeCILL-C
5 ****************************************/
6
7 #include <sstream>
8 #include "phoenix_zmq.h"
9
10
11 ///Create a client socket to be used by the SocketManagerZMQ
12 /** @param context : zeromq context which defines the number of thread to be used in the data transfert
13 * @param type : type of the socket (ZMQ_PULL, ZMQ_PUSH, etc)
14 * @param address : address of the socket (example localhost or 127.0.0.1)
15 * @param port : port to be used
16 * @return create socket
17 */
18 7 zmq::socket_t* pzmq_createClientSocket(zmq::context_t & context, int type, const std::string & address, size_t port, int immediate){
19
1/1
✓ Branch 0 (2→3) taken 7 times.
 7 std::stringstream socketAddressData;
20
4/4
✓ Branch 0 (3→4) taken 7 times.
✓ Branch 2 (4→5) taken 7 times.
✓ Branch 4 (5→6) taken 7 times.
✓ Branch 6 (6→7) taken 7 times.
 7 socketAddressData << "tcp://" << address <<":" << port;
21
3/6
✓ Branch 0 (7→8) taken 7 times.
✓ Branch 2 (8→9) taken 7 times.
✗ Branch 4 (9→10) not taken.
✓ Branch 5 (9→11) taken 7 times.
✗ Branch 6 (26→27) not taken.
✗ Branch 7 (26→28) not taken.
 7 zmq::socket_t *socket = new zmq::socket_t(context, type);
22
1/2
✗ Branch 0 (11→12) not taken.
✓ Branch 1 (11→13) taken 7 times.
 7 if(socket == NULL){return NULL;}
23
24 #if (CPPZMQ_VERSION_MAJOR*100 + CPPZMQ_VERSION_MINOR*10 + CPPZMQ_VERSION_PATCH) >= 471
25
1/1
✓ Branch 0 (13→14) taken 7 times.
 7 socket->set(zmq::sockopt::immediate, immediate);
26
1/1
✓ Branch 0 (14→15) taken 7 times.
 7 socket->set(zmq::sockopt::linger, -1); //1 ms to stop
27 #else
28 int immediateVal(immediate);
29 socket->setsockopt(ZMQ_IMMEDIATE, &immediateVal, sizeof(int));
30 socket->setsockopt(ZMQ_LINGER, -1); //1 ms to stop
31 #endif
32
2/2
✓ Branch 0 (15→16) taken 7 times.
✓ Branch 2 (16→17) taken 7 times.
 7 socket->connect(socketAddressData.str()); // ← connect APRÈS les options
33
34
1/2
✗ Branch 0 (18→19) not taken.
✓ Branch 1 (18→22) taken 7 times.
 7 if(type == ZMQ_SUB){
35 #if (CPPZMQ_VERSION_MAJOR*100 + CPPZMQ_VERSION_MINOR*10 + CPPZMQ_VERSION_PATCH) >= 471
36 ✗ socket->set(zmq::sockopt::subscribe, "");
37 ✗ socket->set(zmq::sockopt::conflate, 1);
38 #else
39 socket->setsockopt(ZMQ_SUBSCRIBE, "", 0);
40 int conflate(1);
41 socket->setsockopt(ZMQ_CONFLATE, &conflate, sizeof(int));
42 #endif
43 }
44 7 return socket;
45 7 }
46 ///Create a server socket to be used by the SocketManagerZMQ
47 /** @param context : zeromq context which defines the number of thread to be used in the data transfert
48 * @param type : type of the socket (ZMQ_PULL, ZMQ_PUSH, etc)
49 * @param port : port to be used
50 * @return create socket
51 */
52 6 zmq::socket_t* pzmq_createServerSocket(zmq::context_t & context, int type, size_t port){
53
1/1
✓ Branch 0 (2→3) taken 6 times.
 6 std::stringstream socketAddressData;
54
2/2
✓ Branch 0 (3→4) taken 6 times.
✓ Branch 2 (4→5) taken 6 times.
 6 socketAddressData << "tcp://127.0.0.1:" << port;
55
3/6
✓ Branch 0 (5→6) taken 6 times.
✓ Branch 2 (6→7) taken 6 times.
✗ Branch 4 (7→8) not taken.
✓ Branch 5 (7→9) taken 6 times.
✗ Branch 6 (18→19) not taken.
✗ Branch 7 (18→20) not taken.
 6 zmq::socket_t *socket = new zmq::socket_t(context, type);
56
1/2
✓ Branch 0 (9→10) taken 6 times.
✗ Branch 1 (9→15) not taken.
 6 if(socket != NULL){
57
2/2
✓ Branch 0 (10→11) taken 6 times.
✓ Branch 2 (11→12) taken 6 times.
 6 socket->bind(socketAddressData.str());
58 #if (CPPZMQ_VERSION_MAJOR*100 + CPPZMQ_VERSION_MINOR*10 + CPPZMQ_VERSION_PATCH) >= 471
59
1/1
✓ Branch 0 (13→14) taken 6 times.
 6 socket->set(zmq::sockopt::linger, -1); //1 ms to stop
60 #else
61 socket->setsockopt(ZMQ_LINGER, -1); //1 ms to stop
62 #endif
63 }
64 6 return socket;
65 6 }
66
67 ///Add a server socket to the manager
68 /** @param context : zmq context where to create socket
69 * @param port : port to be used
70 * @param type : type of the connection (ZMQ_PULL, ZMQ_PUSH, etc)
71 * @param nbBufferMessage : number of messages to be buffered
72 * @param bufferSizeByte : size of the zmq buffer in bytes
73 * @param threadAffinity : bit mask which determines which threads from the 0MQ I/O thread pool associated with the socket's context shall handle newly created connections (1 : means first, 2 : means second, 3 : means first and second, etc)
74 * @param dataRate : expected data rate (in kilobytes per second)
75 * @return zmq socket
76 */
77 6 zmq::socket_t* pzmq_createServerSocket(zmq::context_t & context, size_t port, int type, int nbBufferMessage, int bufferSizeByte,
78 size_t threadAffinity, ssize_t dataRate)
79 {
80 6 zmq::socket_t* socket = pzmq_createServerSocket(context, type, port);
81 6 bool b(socket != NULL);
82
1/2
✓ Branch 0 (3→4) taken 6 times.
✗ Branch 1 (3→6) not taken.
 6 if(b){
83 6 pzmq_setBufferSize(socket, type, nbBufferMessage, dataRate, bufferSizeByte);
84 6 pzmq_setThreadAffinity(socket, threadAffinity);
85 }
86 6 return socket;
87 }
88
89
90 ///Add a client socket to the manager
91 /** @param context : zmq context where to create socket
92 * @param address : address of the server to be connected to
93 * @param port : port to be used
94 * @param type : type of the connection (ZMQ_PULL, ZMQ_PUSH, etc)
95 * @param nbBufferMessage : number of messages to be buffered
96 * @param bufferSizeByte : size of the zmq buffer in bytes
97 * @param threadAffinity : bit mask which determines which threads from the 0MQ I/O thread pool associated with the socket's context shall handle newly created connections (1 : means first, 2 : means second, 3 : means first and second, etc)
98 * @param dataRate : expected data rate (in kilobytes per second)
99 * @return zmq socket
100 */
101 7 zmq::socket_t* pzmq_createClientSocket(zmq::context_t & context, const std::string & address, size_t port, int type, int nbBufferMessage,
102 int bufferSizeByte, size_t threadAffinity, ssize_t dataRate, int immediate)
103 {
104 7 zmq::socket_t* socket = pzmq_createClientSocket(context, type, address, port, immediate);
105 7 bool b(socket != NULL);
106
1/2
✓ Branch 0 (3→4) taken 7 times.
✗ Branch 1 (3→6) not taken.
 7 if(b){
107 7 pzmq_setBufferSize(socket, type, nbBufferMessage, dataRate, bufferSizeByte);
108 7 pzmq_setThreadAffinity(socket, threadAffinity);
109 }
110 7 return socket;
111 }
112
113
114 ///Close the given server socket
115 /** @param[out] socket : pointer to the server socket to be closed (will be set to NULL at then end of the function)
116 */
117 ✗ void pzmq_closeServerSocket(zmq::socket_t *& socket){
118 ✗ if(socket != NULL){
119 ✗ socket->close();
120 ✗ delete socket;
121 ✗ socket = NULL;
122 }
123 ✗ }
124
125 ///Set the number of messages in the messages buffer
126 /** @param[out] socket : socket to be modified
127 * @param nbBufferMessage : number of messages to be buffered
128 */
129 13 void pzmq_setNbMessageBuffer(zmq::socket_t* socket, int nbBufferMessage){
130
1/2
✓ Branch 0 (2→3) taken 13 times.
✗ Branch 1 (2→4) not taken.
 13 if(socket != NULL){
131 #if (CPPZMQ_VERSION_MAJOR*100 + CPPZMQ_VERSION_MINOR*10 + CPPZMQ_VERSION_PATCH) >= 471
132 13 socket->set(zmq::sockopt::rcvhwm, nbBufferMessage);
133 #else
134 //See doc at http://api.zeromq.org/3-1:zmq-setsockopt
135 socket->setsockopt(ZMQ_RCVHWM, &nbBufferMessage, sizeof(int));
136 socket->setsockopt(ZMQ_SNDHWM, &nbBufferMessage, sizeof(int));
137 #endif
138 }
139 13 }
140
141 ///Set the data rate of the socket
142 /** @param[out] socket : socket to be modified
143 * @param type : type of the socket to be used
144 * @param dataRate : expected data rate (in kilobytes per second)
145 */
146 13 void pzmq_setDataRate(zmq::socket_t* socket, int type, int dataRate){
147
3/6
✓ Branch 0 (2→3) taken 13 times.
✗ Branch 1 (2→7) not taken.
✓ Branch 2 (3→4) taken 13 times.
✗ Branch 3 (3→5) not taken.
✗ Branch 4 (4→5) not taken.
✓ Branch 5 (4→7) taken 13 times.
 13 if(socket != NULL && (type == ZMQ_PUB || type == ZMQ_SUB)){
148 ✗ int dataRateKbit(dataRate*8l);
149 #if (CPPZMQ_VERSION_MAJOR*100 + CPPZMQ_VERSION_MINOR*10 + CPPZMQ_VERSION_PATCH) >= 471
150 ✗ socket->set(zmq::sockopt::rate, dataRateKbit);
151 #else
152 //See doc at http://api.zeromq.org/3-1:zmq-setsockopt
153 socket->setsockopt(ZMQ_RATE, dataRateKbit);
154 #endif
155 }
156 13 }
157
158 ///Set the size of the buffer to received messages
159 /** @param[out] socket : socket to be modified
160 * @param bufferSizeByte : size of the zmq buffer in bytes
161 */
162 6 void pzmq_setRecvBufferSize(zmq::socket_t* socket, int bufferSizeByte){
163
1/2
✓ Branch 0 (2→3) taken 6 times.
✗ Branch 1 (2→4) not taken.
 6 if(socket != NULL){
164 #if (CPPZMQ_VERSION_MAJOR*100 + CPPZMQ_VERSION_MINOR*10 + CPPZMQ_VERSION_PATCH) >= 471
165 6 socket->set(zmq::sockopt::rcvbuf, bufferSizeByte);
166 #else
167 //See doc at http://api.zeromq.org/3-1:zmq-setsockopt
168 socket->setsockopt(ZMQ_RCVBUF, bufferSizeByte);
169 #endif
170 }
171 6 }
172
173 ///Set the size of the buffer to send messages
174 /** @param[out] socket : socket to be modified
175 * @param bufferSizeByte : size of the zmq buffer in bytes
176 */
177 7 void pzmq_setSendBufferSize(zmq::socket_t* socket, int bufferSizeByte){
178
1/2
✓ Branch 0 (2→3) taken 7 times.
✗ Branch 1 (2→4) not taken.
 7 if(socket != NULL){
179 #if (CPPZMQ_VERSION_MAJOR*100 + CPPZMQ_VERSION_MINOR*10 + CPPZMQ_VERSION_PATCH) >= 471
180 7 socket->set(zmq::sockopt::sndbuf, bufferSizeByte);
181 #else
182 //See doc at http://api.zeromq.org/3-1:zmq-setsockopt
183 socket->setsockopt(ZMQ_SNDBUF, bufferSizeByte);
184 #endif
185 }
186 7 }
187
188 ///Set the thread affinity of zmq
189 /** @param socket : socket to be modified
190 * @param threadAffinity : bit mask which determines which threads from the 0MQ I/O thread pool associated with the socket's context shall handle newly created connections (1 : means first, 2 : means second, 3 : means first and second, etc)
191 */
192 13 void pzmq_setThreadAffinity(zmq::socket_t* socket, size_t threadAffinity){
193
1/2
✓ Branch 0 (2→3) taken 13 times.
✗ Branch 1 (2→4) not taken.
 13 if(socket != NULL){
194 #if (CPPZMQ_VERSION_MAJOR*100 + CPPZMQ_VERSION_MINOR*10 + CPPZMQ_VERSION_PATCH) >= 471
195 13 socket->set(zmq::sockopt::affinity, threadAffinity);
196 #else
197 //See doc at http://api.zeromq.org/3-1:zmq-setsockopt
198 socket->setsockopt(ZMQ_AFFINITY, threadAffinity);
199 #endif
200 }
201 13 }
202
203 ///Set the size of the buffer to send messages
204 /** @param[out] socket : socket to be modified
205 * @param type : type of the socket to be used
206 * @param nbBufferMessage : number of messages to be buffered
207 * @param dataRate : expected data rate (in kilobytes per second)
208 * @param bufferSizeByte : size of the zmq buffer in bytes
209 */
210 13 void pzmq_setBufferSize(zmq::socket_t* socket, int type, int nbBufferMessage, int dataRate, size_t bufferSizeByte){
211 13 pzmq_setNbMessageBuffer(socket, nbBufferMessage);
212 13 pzmq_setDataRate(socket, type, dataRate);
213
3/4
✓ Branch 0 (4→5) taken 7 times.
✓ Branch 1 (4→6) taken 6 times.
✗ Branch 2 (5→6) not taken.
✓ Branch 3 (5→7) taken 7 times.
 13 if(type == ZMQ_PULL || type == ZMQ_SUB){
214 6 pzmq_setRecvBufferSize(socket, bufferSizeByte);
215
1/4
✗ Branch 0 (7→8) not taken.
✓ Branch 1 (7→9) taken 7 times.
✗ Branch 2 (8→9) not taken.
✗ Branch 3 (8→10) not taken.
 7 }else if(type == ZMQ_PUSH || type == ZMQ_PUB){
216 7 pzmq_setSendBufferSize(socket, bufferSizeByte);
217 }
218 13 }
219
220