Make sure all Redis tasks are executed on the WQ threads

refs #4991

diff --git a/lib/redis/rediswriter-config.cpp b/lib/redis/rediswriter-config.cpp
index 56d17a5672c112d0d71141c1c0eca0dd781503e0..1a68b748c9976914da69f50b3bddd1ef188a6064 100644 (file)
--- a/lib/redis/rediswriter-config.cpp
+++ b/lib/redis/rediswriter-config.cpp
@@ -222,7 +222,7 @@ void RedisWriter::StateChangedHandler(const ConfigObject::Ptr& object)
        Type::Ptr type = object->GetReflectionType();
 
        for (const RedisWriter::Ptr& rw : ConfigType::GetObjectsByType<RedisWriter>()) {
-               rw->SendStatusUpdate(object, type->GetName());
+               rw->m_WorkQueue.Enqueue(boost::bind(&RedisWriter::SendStatusUpdate, rw.get(), object, type->GetName()));
        }
 }
 
@@ -231,7 +231,7 @@ void RedisWriter::VarsChangedHandler(const ConfigObject::Ptr& object)
        Type::Ptr type = object->GetReflectionType();
 
        for (const RedisWriter::Ptr& rw : ConfigType::GetObjectsByType<RedisWriter>()) {
-               rw->SendConfigUpdate(object, type->GetName());
+               rw->m_WorkQueue.Enqueue(boost::bind(&RedisWriter::SendConfigUpdate, rw.get(), object, type->GetName()));
        }
 }
 
@@ -240,6 +240,6 @@ void RedisWriter::VersionChangedHandler(const ConfigObject::Ptr& object)
        Type::Ptr type = object->GetReflectionType();
 
        for (const RedisWriter::Ptr& rw : ConfigType::GetObjectsByType<RedisWriter>()) {
-               rw->SendConfigUpdate(object, type->GetName());
+               rw->m_WorkQueue.Enqueue(boost::bind(&RedisWriter::SendConfigUpdate, rw.get(), object, type->GetName()));
        }
 }

diff --git a/lib/redis/rediswriter.cpp b/lib/redis/rediswriter.cpp
index f5f7e3ad581787e5c7bf301a8e5c4f1af55131d4..6fba06c29d6df4d2f28698026a67481d5d482c34 100644 (file)
--- a/lib/redis/rediswriter.cpp
+++ b/lib/redis/rediswriter.cpp
@@ -141,6 +141,8 @@ void RedisWriter::UpdateSubscriptions(void)
                if (reply->type == REDIS_REPLY_STATUS || reply->type == REDIS_REPLY_ERROR) {
                        Log(LogInformation, "RedisWriter")
                            << "SCAN " << cursor << " MATCH icinga:subscription:* COUNT 1000: " << reply->str;
+
+                       return;
                }
 
                VERIFY(reply->type == REDIS_REPLY_ARRAY);
@@ -167,6 +169,8 @@ void RedisWriter::UpdateSubscriptions(void)
                        if (vreply->type == REDIS_REPLY_STATUS || vreply->type == REDIS_REPLY_ERROR) {
                                Log(LogInformation, "RedisWriter")
                                    << "GET " << keyReply->str << ": " << vreply->str;
+
+                               continue;
                        }
 
                        subscriptions[keyReply->str] = vreply->str;
@@ -271,6 +275,8 @@ void RedisWriter::HandleEvent(const Dictionary::Ptr& event)
                if (reply->type == REDIS_REPLY_STATUS || reply->type == REDIS_REPLY_ERROR) {
                        Log(LogInformation, "RedisWriter")
                            << "LPUSH icinga:event:" << kv.first << " " << body << ": " << reply->str;
+
+                       continue;
                }
 
                if (reply->type == REDIS_REPLY_ERROR) {