[PINS] Add ConsumerNotifier for PINS response path

azure-pipelines.yml

@@ -39,7 +39,7 @@ stages:
         sudo apt-get install -y python3-pip
         sudo pip3 install pytest
         sudo apt-get install -y python
-        sudo apt-get install cmake libgtest-dev
+        sudo apt-get install cmake libgtest-dev libgmock-dev
         cd /usr/src/gtest && sudo cmake . && sudo make
       displayName: "Install dependencies"
     - script: |

common/consumernotifier.cpp

@@ -0,0 +1,41 @@
+#include "consumernotifier.h"
+
+#include "select.h"
+#include "selectable.h"
+
+namespace swss
+{
+
+ConsumerNotifier::ConsumerNotifier(const std::string &notifier_channel_name, DBConnector *db_connector)
+{
+    notifier_channel_name_ = std::string{notifier_channel_name};
+    notification_consumer_ = std::make_unique<NotificationConsumer>(db_connector, notifier_channel_name_);
+}
+
+bool ConsumerNotifier::WaitForNotificationAndPop(std::string &op, std::string &data,
+                                                 std::vector<FieldValueTuple> &values, int64_t timeout_ms)
+{
+    int message_available = notification_consumer_->peek();
+    if (message_available == -1)
+    {
+        return false;
+    }
+
+    // Wait for notification only when the queue is empty.
+    if (message_available == 0)
+    {
+        Select s;
+        s.addSelectable(notification_consumer_.get());
+        Selectable *sel;
+
+        int error = s.select(&sel, timeout_ms);
+        if (error != Select::OBJECT)
+        {
+            return false;
+        }
+    }
+    notification_consumer_->pop(op, data, values);
+    return true;
+}
+
+} // namespace swss

common/consumernotifier.h

@@ -0,0 +1,27 @@
+#pragma once
+
+#include <memory>
+#include <vector>
+
+#include "consumernotifierinterface.h"
+#include "dbconnector.h"
+#include "notificationconsumer.h"
+
+namespace swss
+{
+
+// Wrapper class to use swss::NotificationConsumer.
+class ConsumerNotifier : public ConsumerNotifierInterface
+{
+  public:
+    ConsumerNotifier(const std::string &notifier_channel_name, DBConnector *db_connector);
+
+    bool WaitForNotificationAndPop(std::string &op, std::string &data, std::vector<FieldValueTuple> &values,
+                                   int64_t timeout_ms = 60000LL) override;
+
+  private:
+    std::unique_ptr<NotificationConsumer> notification_consumer_;
+    std::string notifier_channel_name_;
+};
+
+} // namespace swss

common/consumernotifierinterface.h

@@ -0,0 +1,23 @@
+#pragma once
+
+#include <vector>
+
+#include "table.h"
+
+namespace swss
+{
+
+// Wrapper class to use swss::NotificationConsumer.
+class ConsumerNotifierInterface
+{
+  public:
+    virtual ~ConsumerNotifierInterface() = default;
+
+    // Waits for notifications to be posted on the response channel,
+    // this goes into a blocking wait (default timeout of 60 seconds) until a
+    // response is received and then uses swss::pop to get the response values.
+    virtual bool WaitForNotificationAndPop(std::string &op, std::string &data, std::vector<FieldValueTuple> &values,
+                                           int64_t timeout_ms = 60000LL) = 0;
+};
+
+} // namespace swss

common/dbconnector.cpp

@@ -1,5 +1,6 @@
 #include <string.h>
 #include <stdint.h>
+#include <iterator>
 #include <vector>
 #include <unistd.h>
 #include <errno.h>
@@ -750,6 +751,19 @@ int64_t DBConnector::decr(const string &key)
     return r.getContext()->integer;
 }
 
+#ifndef SWIG
+std::unordered_map<std::string, std::string> DBConnector::hgetall(
+    const std::string &key){
+  return hgetall<std::unordered_map<std::string, std::string>>(key);
+}
+#endif
+
+void DBConnector::hmset(
+    const std::string &key,
+    const std::vector<std::pair<std::string, std::string>> &values) {
+  hmset(key, values.begin(), values.end());
+}
+
 shared_ptr<string> DBConnector::get(const string &key)
 {
     RedisCommand sget;

common/dbconnector.h

@@ -9,6 +9,7 @@
 #include <mutex>
 
 #include <hiredis/hiredis.h>
+#include "dbconnectorinterface.h"
 #include "rediscommand.h"
 #include "redisreply.h"
 #define EMPTY_NAMESPACE std::string()
@@ -137,7 +138,7 @@ class RedisContext
     redisContext *m_conn;
 };
 
-class DBConnector : public RedisContext
+class DBConnector : public RedisContext, public DBConnectorInterface
 {
 public:
     static constexpr const char *DEFAULT_UNIXSOCKET = "/var/run/redis/redis.sock";
@@ -174,7 +175,7 @@ class DBConnector : public RedisContext
 
     PubSub *pubsub();
 
-    int64_t del(const std::string &key);
+    int64_t del(const std::string &key) override;
 
 #ifdef SWIG
     // SWIG interface file (.i) globally rename map C++ `del` to python `delete`,
@@ -187,7 +188,7 @@ class DBConnector : public RedisContext
     %}
 #endif
 
-    bool exists(const std::string &key);
+    bool exists(const std::string &key) override;
 
     int64_t hdel(const std::string &key, const std::string &field);
 
@@ -199,11 +200,14 @@ class DBConnector : public RedisContext
     ReturnType hgetall(const std::string &key);
 
 #ifndef SWIG
+    std::unordered_map<std::string, std::string> hgetall(
+        const std::string &key) override;
+
     template <typename OutputIterator>
     void hgetall(const std::string &key, OutputIterator result);
 #endif
 
-    std::vector<std::string> keys(const std::string &key);
+    std::vector<std::string> keys(const std::string &key) override;
 
     std::pair<int, std::vector<std::string>> scan(int cursor = 0, const char *match = "", uint32_t count = 10);
 
@@ -217,9 +221,14 @@ class DBConnector : public RedisContext
 
     void hmset(const std::unordered_map<std::string, std::vector<std::pair<std::string, std::string>>>& multiHash);
 
+    void hmset(const std::string &key,
+               const std::vector<std::pair<std::string, std::string>> &values)
+        override;
+
     std::shared_ptr<std::string> get(const std::string &key);
 
-    std::shared_ptr<std::string> hget(const std::string &key, const std::string &field);
+    std::shared_ptr<std::string> hget(const std::string &key,
+                                      const std::string &field) override;
 
     bool hexists(const std::string &key, const std::string &field);
 

common/dbconnectorinterface.h

@@ -0,0 +1,30 @@
+#pragma once
+
+#include <map>
+#include <memory>
+#include <vector>
+
+namespace swss
+{
+
+class DBConnectorInterface
+{
+  public:
+    virtual ~DBConnectorInterface() = default;
+
+    virtual int64_t del(const std::string &key) = 0;
+
+    virtual void del(const std::vector<std::string> &keys) = 0;
+
+    virtual bool exists(const std::string &key) = 0;
+
+    virtual std::unordered_map<std::string, std::string> hgetall(const std::string &glob) = 0;
+
+    virtual std::vector<std::string> keys(const std::string &glob) = 0;
+
+    virtual void hmset(const std::string &key, const std::vector<std::pair<std::string, std::string>> &values) = 0;
+
+    virtual std::shared_ptr<std::string> hget(const std::string &key, const std::string &field) = 0;
+};
+
+} // namespace swss

common/producerstatetable.cpp

@@ -358,4 +358,9 @@ void ProducerStateTable::apply_temp_view()
     m_tempViewActive = false;
 }
 
+std::string ProducerStateTable::get_table_name() const
+{
+  return getTableName();
+}
+
 }

common/producerstatetable.h

@@ -1,12 +1,14 @@
 #pragma once
 
 #include <memory>
+
+#include "producerstatetableinterface.h"
 #include "table.h"
 #include "redispipeline.h"
 
 namespace swss {
 
-class ProducerStateTable : public TableBase, public TableName_KeySet
+class ProducerStateTable : public TableBase, public TableName_KeySet, public ProducerStateTableInterface
 {
 public:
     ProducerStateTable(DBConnector *db, const std::string &tableName);
@@ -18,11 +20,11 @@ class ProducerStateTable : public TableBase, public TableName_KeySet
     virtual void set(const std::string &key,
                      const std::vector<FieldValueTuple> &values,
                      const std::string &op = SET_COMMAND,
-                     const std::string &prefix = EMPTY_PREFIX);
+                     const std::string &prefix = EMPTY_PREFIX) override;
 
     virtual void del(const std::string &key,
                      const std::string &op = DEL_COMMAND,
-                     const std::string &prefix = EMPTY_PREFIX);
+                     const std::string &prefix = EMPTY_PREFIX) override;
 
 #ifdef SWIG
     // SWIG interface file (.i) globally rename map C++ `del` to python `delete`,
@@ -44,6 +46,9 @@ class ProducerStateTable : public TableBase, public TableName_KeySet
     void create_temp_view();
 
     void apply_temp_view();
+
+    std::string get_table_name() const override;
+
 private:
     bool m_buffered;
     bool m_pipeowned;

common/producerstatetableinterface.h

@@ -0,0 +1,25 @@
+#pragma once
+
+#include <memory>
+#include <vector>
+
+#include "table.h"
+
+namespace swss
+{
+
+class ProducerStateTableInterface
+{
+  public:
+    virtual ~ProducerStateTableInterface() = default;
+
+    virtual void set(const std::string &key, const std::vector<FieldValueTuple> &values,
+                     const std::string &op = SET_COMMAND, const std::string &prefix = EMPTY_PREFIX) = 0;
+
+    virtual void del(const std::string &key, const std::string &op = DEL_COMMAND,
+                     const std::string &prefix = EMPTY_PREFIX) = 0;
+
+    virtual std::string get_table_name() const = 0;
+};
+
+} // namespace swss

tests/Makefile.am

@@ -9,7 +9,7 @@ DBGFLAGS = -g -DNDEBUG
 endif
 
 CFLAGS_GTEST =
-LDADD_GTEST = -L/usr/src/gtest -lgtest -lgtest_main
+LDADD_GTEST = -L/usr/src/gtest -lgtest -lgtest_main -lgmock -lgmock_main
 
 tests_SOURCES = redis_ut.cpp                \
                 redis_piped_ut.cpp          \

tests/fakes/fake_consumer_notifier.cpp

@@ -0,0 +1,20 @@
+#include "swss/fakes/fake_consumer_notifier.h"
+
+#include "glog/logging.h"
+
+namespace swss
+{
+
+FakeConsumerNotifier::FakeConsumerNotifier(FakeSonicDbTable *app_db_table) : app_db_table_(app_db_table)
+{
+    LOG_IF(FATAL, app_db_table == nullptr) << "FakeSonicDbTable cannot be nullptr.";
+}
+
+bool FakeConsumerNotifier::WaitForNotificationAndPop(std::string &op, std::string &data, SonicDbEntry &values,
+                                                     int64_t timeout_ms)
+{
+    app_db_table_->GetNextNotification(op, data, values);
+    return true;
+}
+
+} // namespace swss

tests/fakes/fake_consumer_notifier.h

@@ -0,0 +1,36 @@
+#pragma once
+
+#include <vector>
+
+#include "absl/status/status.h"
+#include "swss/consumernotifierinterface.h"
+#include "swss/fakes/fake_sonic_db_table.h"
+
+namespace swss
+{
+
+// Fakes the OrchAgent response path behavior for AppDb table entries.
+//
+// Every write into an AppDb table is handled by the OrchAgent. The write can
+// either succeed or fail. In the latter case a failed StatusCode should be
+// returned by the fake.
+class FakeConsumerNotifier final : public ConsumerNotifierInterface
+{
+  public:
+    explicit FakeConsumerNotifier(FakeSonicDbTable *app_db_table);
+
+    // Not copyable or moveable.
+    FakeConsumerNotifier(const FakeConsumerNotifier &) = delete;
+    FakeConsumerNotifier &operator=(const FakeConsumerNotifier &) = delete;
+
+    // Faked methods.
+    bool WaitForNotificationAndPop(std::string &op, std::string &data, SonicDbEntry &values,
+                                   int64_t timeout_ms = 60000LL) override;
+
+  private:
+    // The AppDb table maintains a list of notifications that this fake can
+    // request.
+    FakeSonicDbTable *app_db_table_; // No ownership.
+};
+
+} // namespace swss