Improve Consumer interface to support ZMQ based Producer/Consumer table. (sonic-net#2562)

**What I did**
Improve Consumer interface to support ZMQ based Producer/Consumer table.

**Why I did it**
To improve route create performance, swsscommon lib will add ZMQ based Producer/Consumer table.
Because currently Consumer interface only support Redis based Producer/Consumer table, so improve this interface to support ZMQ based Producer/Consumer table.

ZMQ based Producer/Consumer table can find in this PR: sonic-net/sonic-swss-common#715

**How I verified it**
Pass all UT.

**Details if related**

Author: liuh-80

orchagent/orch.cpp

@@ -70,7 +70,7 @@ vector<Selectable *> Orch::getSelectables()
     return selectables;
 }
 
-void Consumer::addToSync(const KeyOpFieldsValuesTuple &entry)
+void ConsumerBase::addToSync(const KeyOpFieldsValuesTuple &entry)
 {
     SWSS_LOG_ENTER();
 
@@ -157,7 +157,7 @@ void Consumer::addToSync(const KeyOpFieldsValuesTuple &entry)
 
 }
 
-size_t Consumer::addToSync(const std::deque<KeyOpFieldsValuesTuple> &entries)
+size_t ConsumerBase::addToSync(const std::deque<KeyOpFieldsValuesTuple> &entries)
 {
     SWSS_LOG_ENTER();
 
@@ -194,9 +194,7 @@ size_t Consumer::refillToSync(Table* table)
 
 size_t Consumer::refillToSync()
 {
-    ConsumerTableBase *consumerTable = getConsumerTable();
-
-    auto subTable = dynamic_cast<SubscriberStateTable *>(consumerTable);
+    auto subTable = dynamic_cast<SubscriberStateTable *>(getSelectable());
     if (subTable != NULL)
     {
         size_t update_size = 0;
@@ -213,35 +211,14 @@ size_t Consumer::refillToSync()
     else
     {
         // consumerTable is either ConsumerStateTable or ConsumerTable
-        auto db = consumerTable->getDbConnector();
-        string tableName = consumerTable->getTableName();
+        auto db = getDbConnector();
+        string tableName = getTableName();
         auto table = Table(db, tableName);
         return refillToSync(&table);
     }
 }
 
-void Consumer::execute()
-{
-    SWSS_LOG_ENTER();
-
-    size_t update_size = 0;
-    do
-    {
-        std::deque<KeyOpFieldsValuesTuple> entries;
-        getConsumerTable()->pops(entries);
-        update_size = addToSync(entries);
-    } while (update_size != 0);
-
-    drain();
-}
-
-void Consumer::drain()
-{
-    if (!m_toSync.empty())
-        m_orch->doTask(*this);
-}
-
-string Consumer::dumpTuple(const KeyOpFieldsValuesTuple &tuple)
+string ConsumerBase::dumpTuple(const KeyOpFieldsValuesTuple &tuple)
 {
     string s = getTableName() + getConsumerTable()->getTableNameSeparator() + kfvKey(tuple)
                + "|" + kfvOp(tuple);
@@ -253,7 +230,7 @@ string Consumer::dumpTuple(const KeyOpFieldsValuesTuple &tuple)
     return s;
 }
 
-void Consumer::dumpPendingTasks(vector<string> &ts)
+void ConsumerBase::dumpPendingTasks(vector<string> &ts)
 {
     for (auto &tm : m_toSync)
     {
@@ -265,6 +242,28 @@ void Consumer::dumpPendingTasks(vector<string> &ts)
     }
 }
 
+void Consumer::execute()
+{
+    SWSS_LOG_ENTER();
+
+    size_t update_size = 0;
+    auto table = static_cast<swss::ConsumerTableBase *>(getSelectable());
+    do
+    {
+        std::deque<KeyOpFieldsValuesTuple> entries;
+        table->pops(entries);
+        update_size = addToSync(entries);
+    } while (update_size != 0);
+
+    drain();
+}
+
+void Consumer::drain()
+{
+    if (!m_toSync.empty())
+        m_orch->doTask(*this);
+}
+
 size_t Orch::addExistingData(const string& tableName)
 {
     auto consumer = dynamic_cast<Consumer *>(getExecutor(tableName));
@@ -586,7 +585,7 @@ void Orch::logfileReopen()
     }
 }
 
-void Orch::recordTuple(Consumer &consumer, const KeyOpFieldsValuesTuple &tuple)
+void Orch::recordTuple(ConsumerBase &consumer, const KeyOpFieldsValuesTuple &tuple)
 {
     string s = consumer.dumpTuple(tuple);
 

orchagent/orch.h

@@ -132,49 +132,66 @@ class Executor : public swss::Selectable
     swss::Selectable *getSelectable() const { return m_selectable; }
 };
 
-class Consumer : public Executor {
+class ConsumerBase : public Executor {
+public:
+    ConsumerBase(swss::Selectable *selectable, Orch *orch, const std::string &name)
+        : Executor(selectable, orch, name)
+    {
+    }
+
+    virtual swss::TableBase *getConsumerTable() const = 0;
+
+    std::string getTableName() const
+    {
+        return getConsumerTable()->getTableName();
+    }
+
+    std::string dumpTuple(const swss::KeyOpFieldsValuesTuple &tuple);
+    void dumpPendingTasks(std::vector<std::string> &ts);
+
+    /* Store the latest 'golden' status */
+    // TODO: hide?
+    SyncMap m_toSync;
+
+    void addToSync(const swss::KeyOpFieldsValuesTuple &entry);
+
+    // Returns: the number of entries added to m_toSync
+    size_t addToSync(const std::deque<swss::KeyOpFieldsValuesTuple> &entries);
+};
+
+class Consumer : public ConsumerBase {
 public:
     Consumer(swss::ConsumerTableBase *select, Orch *orch, const std::string &name)
-        : Executor(select, orch, name)
+        : ConsumerBase(select, orch, name)
     {
     }
 
-    swss::ConsumerTableBase *getConsumerTable() const
+    swss::TableBase *getConsumerTable() const override
     {
+        // ConsumerTableBase is a subclass of TableBase
         return static_cast<swss::ConsumerTableBase *>(getSelectable());
     }
 
-    std::string getTableName() const
+    const swss::DBConnector* getDbConnector() const
     {
-        return getConsumerTable()->getTableName();
+        auto table = static_cast<swss::ConsumerTableBase *>(getSelectable());
+        return table->getDbConnector();
     }
 
     int getDbId() const
     {
-        return getConsumerTable()->getDbConnector()->getDbId();
+        return getDbConnector()->getDbId();
     }
 
     std::string getDbName() const
     {
-        return getConsumerTable()->getDbConnector()->getDbName();
+        return getDbConnector()->getDbName();
     }
 
-    std::string dumpTuple(const swss::KeyOpFieldsValuesTuple &tuple);
-    void dumpPendingTasks(std::vector<std::string> &ts);
-
     size_t refillToSync();
     size_t refillToSync(swss::Table* table);
-    void execute();
-    void drain();
-
-    /* Store the latest 'golden' status */
-    // TODO: hide?
-    SyncMap m_toSync;
-
-    void addToSync(const swss::KeyOpFieldsValuesTuple &entry);
-
-    // Returns: the number of entries added to m_toSync
-    size_t addToSync(const std::deque<swss::KeyOpFieldsValuesTuple> &entries);
+    void execute() override;
+    void drain() override;
 };
 
 typedef std::map<std::string, std::shared_ptr<Executor>> ConsumerMap;
@@ -215,12 +232,12 @@ class Orch
     virtual void doTask();
 
     /* Run doTask against a specific executor */
-    virtual void doTask(Consumer &consumer) = 0;
+    virtual void doTask(Consumer &consumer) { };
     virtual void doTask(swss::NotificationConsumer &consumer) { }
     virtual void doTask(swss::SelectableTimer &timer) { }
 
     /* TODO: refactor recording */
-    static void recordTuple(Consumer &consumer, const swss::KeyOpFieldsValuesTuple &tuple);
+    static void recordTuple(ConsumerBase &consumer, const swss::KeyOpFieldsValuesTuple &tuple);
 
     void dumpPendingTasks(std::vector<std::string> &ts);