Merge pull request #113 from nickbroon/zmq_minor_fix

split zsock_new_sub() into zsock_new() and zsock_connect()
tumi8 · Sep 10, 2018 · d6d61c5 · d6d61c5
2 parents aceda69 + 187082a
commit d6d61c5
Showing 1 changed file with 13 additions and 5 deletions.
diff --git a/src/modules/ipfix/IpfixReceiverZmq.cpp b/src/modules/ipfix/IpfixReceiverZmq.cpp
@@ -56,17 +56,25 @@ IpfixReceiverZmq::IpfixReceiverZmq(std::vector<std::string> endpoints,
 	}
 
 	for (std::vector<std::string>::iterator i = endpoints.begin(); i != endpoints.end(); i++) {
-		// If no channel is passed down, listen on everything (empty string)
-		zsock_t *sock = zsock_new_sub((*i).c_str(), channels.empty() ? "" : NULL);
+		zsock_t *sock = zsock_new(ZMQ_SUB);
 		if (!sock) {
-			THROWEXCEPTION("Could not connect ZMQ socket, cannot start ZMQ Receiver");
+			THROWEXCEPTION("Could not create ZMQ socket");
 		}
 
 		zsock_set_sndhwm(sock, zmq_high_watermark);
 		zsock_set_rcvhwm(sock, zmq_high_watermark);
 
-		for (std::vector<std::string>::iterator j = channels.begin(); j != channels.end(); j++) {
-			zsock_set_subscribe(sock, (*j).c_str());
+		if (zsock_connect(sock, "%s", (*i).c_str())) {
+	                THROWEXCEPTION("Could not connect ZMQ socket");
+	        }
+
+		// If no channel is passed down, listen on everything (empty string)
+		if (channels.empty()) {
+			zsock_set_subscribe(sock, "");
+		} else {
+			for (std::vector<std::string>::iterator j = channels.begin(); j != channels.end(); j++) {
+				zsock_set_subscribe(sock, (*j).c_str());
+			}
 		}
 
 		if (zpoller_add(zpoller, sock)) {