[NotificationProducer] add pipeline support

BUILD

@@ -10,6 +10,7 @@ cc_library(
         "common/*.hpp",
     ]),
     copts = [
+        "-std=c++14",
         "-I/usr/include/libnl3", # Expected location in the SONiC build container"
     ],
     includes = [

common/notificationproducer.cpp

@@ -1,7 +1,14 @@
 #include "notificationproducer.h"
 
+#define NON_BUFFERED_COMMAND_BUFFER_SIZE 1
+
 swss::NotificationProducer::NotificationProducer(swss::DBConnector *db, const std::string &channel):
-    m_db(db), m_channel(channel)
+    m_ownedpipe(std::make_unique<swss::RedisPipeline>(db, NON_BUFFERED_COMMAND_BUFFER_SIZE)), m_pipe(m_ownedpipe.get()), m_channel(channel), m_buffered(false)
+{
+}
+
+swss::NotificationProducer::NotificationProducer(swss::RedisPipeline *pipeline, const std::string &channel, bool buffered):
+     m_pipe(pipeline), m_channel(channel), m_buffered(buffered)
 {
 }
 
@@ -19,5 +26,19 @@ int64_t swss::NotificationProducer::send(const std::string &op, const std::strin
 
     SWSS_LOG_DEBUG("channel %s, publish: %s", m_channel.c_str(), msg.c_str());
 
-    return m_db->publish(m_channel, msg);
+    RedisCommand command;
+    command.format("PUBLISH %s %s", m_channel.c_str(), msg.c_str());
+
+    if (m_buffered)
+    {
+        m_pipe->push(command, REDIS_REPLY_INTEGER);
+
+        // if operating in buffered mode return -1 as we can't know the number
+        // of client's that have read the message immediately
+        return -1;
+    }
+
+    RedisReply reply = m_pipe->push(command);
+    reply.checkReplyType(REDIS_REPLY_INTEGER);
+    return reply.getReply<long long int>();
 }

common/notificationproducer.h

@@ -9,6 +9,7 @@
 #include "logger.h"
 #include "table.h"
 #include "redisreply.h"
+#include "redispipeline.h"
 #include "json.h"
 
 namespace swss {
@@ -18,6 +19,14 @@ class NotificationProducer
 public:
     NotificationProducer(swss::DBConnector *db, const std::string &channel);
 
+    /**
+     * @brief Create NotificationProducer using RedisPipeline
+     * @param pipeline Pointer to RedisPipeline
+     * @param channel Channel name
+     * @param buffered Whether NotificationProducer will work in buffered mode
+     */
+    NotificationProducer(RedisPipeline *pipeline, const std::string &channel, bool buffered = false);
+
     // Returns: the number of clients that received the message
     int64_t send(const std::string &op, const std::string &data, std::vector<FieldValueTuple> &values);
 
@@ -26,8 +35,10 @@ class NotificationProducer
     NotificationProducer(const NotificationProducer &other);
     NotificationProducer& operator = (const NotificationProducer &other);
 
-    swss::DBConnector *m_db;
+    std::unique_ptr<RedisPipeline> m_ownedpipe{};
+    RedisPipeline *m_pipe;
     std::string m_channel;
+    bool m_buffered{false};
 };
 
 }

configure.ac

@@ -50,7 +50,7 @@ AC_PATH_PROGS(SWIG, [swig4.0 swig3.0 swig])
 CFLAGS_COMMON=""
 CFLAGS_COMMON+=" -ansi"
 CFLAGS_COMMON+=" -fPIC"
-CFLAGS_COMMON+=" -std=c++11"
+CFLAGS_COMMON+=" -std=c++14"
 CFLAGS_COMMON+=" -Wall"
 CFLAGS_COMMON+=" -Wcast-align"
 CFLAGS_COMMON+=" -Wcast-qual"

tests/ntf_ut.cpp

@@ -164,3 +164,54 @@ TEST(Notifications, peek)
     int rc = nc.peek();
     EXPECT_EQ(rc, 0);
 }
+
+TEST(Notifications, pipelineProducer)
+{
+    SWSS_LOG_ENTER();
+
+    swss::DBConnector dbNtf("ASIC_DB", 0, true);
+    swss::RedisPipeline pipeline{&dbNtf};
+    swss::NotificationConsumer nc(&dbNtf, "NOTIFICATIONS", 100, (size_t)10);
+    const bool buffered = true;
+    swss::NotificationProducer notifications(&pipeline, "NOTIFICATIONS", buffered);
+
+    std::vector<swss::FieldValueTuple> entry;
+    for(int i = 0; i < messages; i++)
+    {
+        auto s = std::to_string(i+1);
+        auto sentClients = notifications.send("ntf", s, entry);
+        // In buffered mode we get -1 in return
+        EXPECT_EQ(sentClients, -1);
+    }
+
+    // Flush the pipeline
+    pipeline.flush();
+
+    // Pop all the notifications
+    std::deque<swss::KeyOpFieldsValuesTuple> vkco;
+    size_t popped = 0;
+    size_t npop = 10000;
+    int collected = 0;
+    while(nc.peek() > 0 && popped < npop)
+    {
+        nc.pops(vkco);
+        popped += vkco.size();
+
+        for (auto& kco : vkco)
+        {
+            collected++;
+            auto data = kfvKey(kco);
+            auto op = kfvOp(kco);
+
+            EXPECT_EQ(op, "ntf");
+            int i = stoi(data);
+            EXPECT_EQ(i, collected);
+        }
+    }
+    EXPECT_EQ(popped, (size_t)messages);
+
+    // Peek and get nothing more
+    int rc = nc.peek();
+    EXPECT_EQ(rc, 0);
+}
+