ExternalVendorCode/ZeroTierOne
1
0
Fork 0
mirror of https://github.com/zerotier/ZeroTierOne.git synced 2024-12-21 05:53:09 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

amqp_consume_message now has a timeout

Browse Source 
RabbitMQ::consume() will return an empty string if the call to amqp_consume_message times out
This commit is contained in:
Grant Limberg 2019-04-18 14:57:06 -07:00
parent 6014df2847
commit 6a027c9c0a
2 changed files with 17 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
controller/PostgreSQL.cpp
View File

@ -666,6 +666,9 @@ void PostgreSQL::_membersWatcher_RabbitMQ() {
try { try {
std::string msg = rmq.consume(); std::string msg = rmq.consume();
// fprintf(stderr, "Got Member Update: %s\n", msg.c_str()); // fprintf(stderr, "Got Member Update: %s\n", msg.c_str());
if (msg.empty()) {
continue;
}
json tmp(json::parse(msg)); json tmp(json::parse(msg));
json &ov = tmp["old_val"]; json &ov = tmp["old_val"];
json &nv = tmp["new_val"]; json &nv = tmp["new_val"];
@ -766,6 +769,9 @@ void PostgreSQL::_networksWatcher_RabbitMQ() {
while (_run == 1) { while (_run == 1) {
try { try {
std::string msg = rmq.consume(); std::string msg = rmq.consume();
if (msg.empty()) {
continue;
}
// fprintf(stderr, "Got network update: %s\n", msg.c_str()); // fprintf(stderr, "Got network update: %s\n", msg.c_str());
json tmp(json::parse(msg)); json tmp(json::parse(msg));
json &ov = tmp["old_val"]; json &ov = tmp["old_val"];

11
controller/RabbitMQ.cpp
View File

@ -80,10 +80,19 @@ std::string RabbitMQ::consume()
amqp_envelope_t envelope; amqp_envelope_t envelope;
amqp_maybe_release_buffers(_conn); amqp_maybe_release_buffers(_conn);
res = amqp_consume_message(_conn, &envelope, NULL, 0); struct timeval timeout;
timeout.tv_sec = 1;
timeout.tv_usec = 0;
res = amqp_consume_message(_conn, &envelope, &timeout, 0);
if (res.reply_type != AMQP_RESPONSE_NORMAL) { if (res.reply_type != AMQP_RESPONSE_NORMAL) {
if (res.reply_type == AMQP_RESPONSE_LIBRARY_EXCEPTION && res.library_error == AMQP_STATUS_TIMEOUT) {
// timeout waiting for message. Return empty string
return "";
} else {
throw std::runtime_error("Error getting message"); throw std::runtime_error("Error getting message");
} }
}
std::string msg( std::string msg(
(const char*)envelope.message.body.bytes, (const char*)envelope.message.body.bytes,