Add PublisherEventTable/SubscriberEventTable classes

common/Makefile.am

@@ -53,10 +53,12 @@ common_libswsscommon_la_SOURCES = \
     common/notificationproducer.cpp  \
     common/linkcache.cpp             \
     common/portmap.cpp               \
+    common/publishereventtable.cpp     \
     common/pubsub.cpp                \
     common/tokenize.cpp              \
     common/exec.cpp                  \
     common/saiaclschema.cpp          \
+    common/subscribereventtable.cpp  \
     common/subscriberstatetable.cpp  \
     common/timestamp.cpp             \
     common/warm_restart.cpp          \

common/configdb.cpp

@@ -1,9 +1,13 @@
 #include <boost/algorithm/string.hpp>
 #include <map>
+#include <memory>
 #include <vector>
 #include "configdb.h"
+#include "dbconnector.h"
 #include "pubsub.h"
 #include "converter.h"
+#include "table.h"
+#include "publishereventtable.h"
 
 using namespace std;
 using namespace swss;
@@ -19,6 +23,7 @@ void ConfigDBConnector_Native::db_connect(string db_name, bool wait_for_init, bo
 {
     m_db_name = db_name;
     m_key_separator = m_table_name_separator = get_db_separator(db_name);
+    m_event_tables = SonicDBConfig::getEventTables(m_db_name);
     SonicV2Connector_Native::connect(m_db_name, retry_on);
 
     if (wait_for_init)
@@ -62,6 +67,19 @@ void ConfigDBConnector_Native::connect(bool wait_for_init, bool retry_on)
     db_connect("CONFIG_DB", wait_for_init, retry_on);
 }
 
+shared_ptr<Table> ConfigDBConnector_Native::getTableWriter(const string &table_name)
+{
+    auto& client = get_redis_client(m_db_name);
+    if (m_event_tables.find(to_upper(table_name)) != m_event_tables.end())
+    {
+        return make_shared<PublisherEventTable>(&client, to_upper(table_name));
+    }
+    else
+    {
+        return make_shared<Table>(&client, to_upper(table_name));
+    }
+}
+
 // Write a table entry to config db.
 //    Remove extra fields in the db which are not in the data.
 // Args:
@@ -71,23 +89,30 @@ void ConfigDBConnector_Native::connect(bool wait_for_init, bool retry_on)
 //           Pass {} as data will delete the entry.
 void ConfigDBConnector_Native::set_entry(string table, string key, const map<string, string>& data)
 {
-    auto& client = get_redis_client(m_db_name);
+    auto table_writer = getTableWriter(table);
     string _hash = to_upper(table) + m_table_name_separator + key;
     if (data.empty())
     {
-        client.del(_hash);
+        table_writer->del(key);
     }
     else
     {
         auto original = get_entry(table, key);
-        client.hmset(_hash, data.begin(), data.end());
+        // Convert map<string, string> to vector<FieldValueTuple>
+        std::vector<FieldValueTuple> values;
+        values.reserve(data.size());
+        for (const auto& kv : data)
+        {
+            values.emplace_back(kv.first, kv.second);
+        }
+        table_writer->set(key, values);
         for (auto& it: original)
         {
             auto& k = it.first;
             bool found = data.find(k) != data.end();
             if (!found)
             {
-                client.hdel(_hash, k);
+                table_writer->hdel(key, k);
             }
         }
     }
@@ -102,15 +127,21 @@ void ConfigDBConnector_Native::set_entry(string table, string key, const map<str
 //           Pass None as data will delete the entry.
 void ConfigDBConnector_Native::mod_entry(string table, string key, const map<string, string>& data)
 {
-    auto& client = get_redis_client(m_db_name);
-    string _hash = to_upper(table) + m_table_name_separator + key;
+    auto table_writer = getTableWriter(table);
     if (data.empty())
     {
-        client.del(_hash);
+        table_writer->del(key);
     }
     else
     {
-        client.hmset(_hash, data.begin(), data.end());
+        // Convert map<string, string> to vector<FieldValueTuple>
+        std::vector<FieldValueTuple> values;
+        values.reserve(data.size());
+        for (const auto& kv : data)
+        {
+            values.emplace_back(kv.first, kv.second);
+        }
+        table_writer->set(key, values);
     }
 }
 
@@ -195,12 +226,12 @@ map<string, map<string, string>> ConfigDBConnector_Native::get_table(string tabl
 //     table: Table name.
 void ConfigDBConnector_Native::delete_table(string table)
 {
-    auto& client = get_redis_client(m_db_name);
-    string pattern = to_upper(table) + m_table_name_separator + "*";
-    const auto& keys = client.keys(pattern);
+    auto table_writer = getTableWriter(table);
+    vector<string> keys;
+    table_writer->getKeys(keys);
     for (auto& key: keys)
     {
-        client.del(key);
+        table_writer->del(key);
     }
 }
 

common/configdb.h

@@ -2,8 +2,10 @@
 
 #include <string>
 #include <map>
+#include <memory>
 #include "sonicv2connector.h"
 #include "redistran.h"
+#include "table.h"
 
 namespace swss {
 
@@ -31,8 +33,11 @@ class ConfigDBConnector_Native : public SonicV2Connector_Native
     std::string getDbName() const;
 
 protected:
+    std::shared_ptr<swss::Table> getTableWriter(const std::string &table_name);
+
     std::string m_table_name_separator = "|";
     std::string m_key_separator = "|";
+    std::set<std::string> m_event_tables;
 
     std::string m_db_name;
 };

common/database_config.json

@@ -35,7 +35,8 @@
         "CONFIG_DB" : {
             "id" : 4,
             "separator": "|",
-            "instance" : "redis"
+            "instance" : "redis",
+            "event_tables" : [ "PORT" ]
         },
         "PFC_WD_DB" : {
             "id" : 5,
@@ -50,7 +51,8 @@
         "STATE_DB" : {
             "id" : 6,
             "separator": "|",
-            "instance" : "redis"
+            "instance" : "redis",
+            "event_tables" : [ "PORT_TABLE", "TRANSCEIVER_INFO" ]
         },
         "SNMP_OVERLAY_DB" : {
             "id" : 7,

common/dbconnector.cpp

@@ -33,27 +33,42 @@
             i >> j;
             for (auto it = j["INSTANCES"].begin(); it!= j["INSTANCES"].end(); it++)
             {
-               string instName = it.key();
-               string socket;
-               auto path = it.value().find("unix_socket_path");
-               if (path != it.value().end())
-               {
+                string instName = it.key();
+                string socket;
+                auto path = it.value().find("unix_socket_path");
+                if (path != it.value().end())
+                {
                     socket = *path;
-               }
-               string hostname = it.value().at("hostname");
-               int port = it.value().at("port");
-               inst_entry[instName] = {socket, hostname, port};
+                }
+                string hostname = it.value().at("hostname");
+                int port = it.value().at("port");
+                inst_entry[instName] = {socket, hostname, port};
             }
 
             for (auto it = j["DATABASES"].begin(); it!= j["DATABASES"].end(); it++)
             {
-               string dbName = it.key();
-               string instName = it.value().at("instance");
-               int dbId = it.value().at("id");
-               string separator = it.value().at("separator");
-               db_entry[dbName] = {instName, dbId, separator};
+                string dbName = it.key();
+                string instName = it.value().at("instance");
+                int dbId = it.value().at("id");
+                string separator = it.value().at("separator");
+
+                set<string> eventTables;
+                auto path = it.value().find("event_tables");
+                if (path != it.value().end())
+                {
+                    for (auto& eventTable: it.value()["event_tables"])
+                    {
+                        eventTables.emplace(eventTable);
+                        // printf("Event table: %s %s\n", dbName.c_str(), eventTable.get<string>().c_str());
+                    }
+                }
+                else
+                {
+                    // printf("Warning: event_tables not found for database %s, using empty set\n", dbName.c_str());
+                }
+                db_entry[dbName] = {instName, dbId, separator, eventTables};
 
-               separator_entry.emplace(dbId, separator);
+                separator_entry.emplace(dbId, separator);
             }
         }
 
@@ -179,8 +194,11 @@
 
     SonicDBKey empty_key;
     parseDatabaseConfig(file, inst_entry, db_entry, separator_entry);
+    // printf("SonicDBConfig::initialize: parsed %zu Redis instances, %zu databases, %zu separators, %lu\n",
+           // inst_entry.size(), db_entry.size(), separator_entry.size(), db_entry["CONFIG_DB"].eventTables.size());
     m_inst_info.emplace(empty_key, std::move(inst_entry));
     m_db_info.emplace(empty_key, std::move(db_entry));
+    // printf("m_db_info %lu\n", m_db_info[empty_key]["CONFIG_DB"].eventTables.size());
     m_db_separator.emplace(empty_key, std::move(separator_entry));
 
     // Set it as the config file is already parsed and init done.
@@ -392,6 +410,41 @@
     }
 }
 
+set<string> SonicDBConfig::getEventTables(const string &dbName, const string &netns, const std::string &containerName)
+{
+    SonicDBKey key;
+    // printf("SonicDBConfig::getEventTables: dbName=%s, netns=%s, containerName=%s, %lu\n",
+    //        dbName.c_str(), netns.c_str(), containerName.c_str(), m_db_info[key][dbName].eventTables.size());
+    key.netns = netns;
+    key.containerName = containerName;
+    // printf("SonicDBConfig::getEventTables: dbName=%s, netns=%s, containerName=%s, %lu\n",
+    //        dbName.c_str(), netns.c_str(), containerName.c_str(), m_db_info[key][dbName].eventTables.size());
+    auto ret = getEventTables(dbName, key);
+    // printf("SonicDBConfig::getEventTables: %lu\n", ret.size());
+    return ret;
+}
+
+set<string> SonicDBConfig::getEventTables(const string &dbName, const SonicDBKey &key)
+{
+    std::lock_guard<std::recursive_mutex> guard(m_db_info_mutex);
+
+    if (!m_init)
+        initialize(DEFAULT_SONIC_DB_CONFIG_FILE);
+
+    if (!key.isEmpty())
+    {
+        if (!m_global_init)
+        {
+            SWSS_LOG_THROW("Initialize global DB config using API SonicDBConfig::initializeGlobalConfig");
+        }
+    }
+    // printf("SonicDBConfig::getEventTables2: dbName=%s, %lu\n",
+    //        dbName.c_str(), m_db_info[key][dbName].eventTables.size());
+    auto ret = getDbInfo(dbName, key).eventTables;
+    // printf("SonicDBConfig::getEventTables2: %lu\n", ret.size());
+    return ret;
+}
+
 string SonicDBConfig::getDbSock(const string &dbName, const string &netns, const std::string &containerName)
 {
     SonicDBKey key;

common/dbconnector.h

@@ -5,6 +5,7 @@
 #include <vector>
 #include <unordered_map>
 #include <utility>
+#include <set>
 #include <map>
 #include <memory>
 #include <mutex>
@@ -36,6 +37,7 @@ class SonicDBInfo
     std::string instName;
     int dbId;
     std::string separator;
+    std::set<std::string> eventTables;
 };
 
 struct SonicDBKey
@@ -120,6 +122,8 @@ class SonicDBConfig
     static std::string getSeparator(int dbId, const std::string &netns = EMPTY_NAMESPACE, const std::string &containerName=EMPTY_CONTAINERNAME);
     static std::string getSeparator(int dbId, const SonicDBKey &key);
     static std::string getSeparator(const DBConnector* db);
+    static std::set<std::string> getEventTables(const std::string &dbName, const std::string &netns = EMPTY_NAMESPACE, const std::string &containerName=EMPTY_CONTAINERNAME);
+    static std::set<std::string> getEventTables(const std::string &dbName, const SonicDBKey &key);
     static std::string getDbSock(const std::string &dbName, const std::string &netns = EMPTY_NAMESPACE, const std::string &containerName=EMPTY_CONTAINERNAME);
     static std::string getDbSock(const std::string &dbName, const SonicDBKey &key);
     static std::string getDbHostname(const std::string &dbName, const std::string &netns = EMPTY_NAMESPACE, const std::string &containerName=EMPTY_CONTAINERNAME);