[Link Event Damping] Port state change handler class.

- Class to handle the port state change event callback from SAI and
  sending the events to syncd main thread for processing by link event
  damper.

Depends on: sonic-net#1297

HLD: sonic-net/SONiC#1071

syncd/Makefile.am

@@ -34,6 +34,7 @@ libSyncd_a_SOURCES = \
 				NotificationQueue.cpp \
 				PortMap.cpp \
 				PortMapParser.cpp \
+				PortStateChangeHandler.cpp \
 				RedisClient.cpp \
 				RedisNotificationProducer.cpp \
 				RequestShutdownCommandLineOptions.cpp \

syncd/PortStateChangeHandler.cpp

@@ -0,0 +1,46 @@
+#include "PortStateChangeHandler.h"
+
+#include <string>
+
+using namespace syncd;
+
+PortStateChangeHandler::PortStateChangeHandler(
+        _In_ std::shared_ptr<swss::SelectableEvent> portStateChangeEvent)
+    : m_portStateChangeEvent(portStateChangeEvent)
+{
+    SWSS_LOG_ENTER();
+
+    m_portStateChangeQueue = std::make_shared<portOperStatusNotificationQueue>(
+            m_portStateChangeQueueSize);
+}
+
+PortStateChangeHandler::~PortStateChangeHandler()
+{
+    SWSS_LOG_ENTER();
+}
+
+void PortStateChangeHandler::handlePortStateChangeNotification(
+        _In_ uint32_t count,
+        _In_ const sai_port_oper_status_notification_t *data)
+{
+    SWSS_LOG_ENTER();
+
+    if (m_portStateChangeEvent == nullptr)
+    {
+        SWSS_LOG_THROW("Unexpected error: port state change event is null.");
+    }
+
+    for (uint32_t idx = 0; idx < count; ++idx)
+    {
+        if (m_portStateChangeQueue->enqueue(data[idx]) == false)
+        {
+            SWSS_LOG_ERROR(
+                    "Unexpected error: failed to enqueue the port state change "
+                    "notification.");
+
+            return;
+        }
+    }
+
+    m_portStateChangeEvent->notify();
+}

syncd/PortStateChangeHandler.h

@@ -0,0 +1,59 @@
+#pragma once
+
+extern "C" {
+#include "saimetadata.h"
+}
+
+#include <memory>
+
+#include "ConcurrentQueue.h"
+#include "swss/logger.h"
+#include "swss/selectableevent.h"
+
+namespace syncd
+{
+
+    // Class to handle the port state change callback from SAI. This consists a
+    // selectable event that will be used to send notification from producer thread
+    // to consumer thread, and a mutex protected concurrent queue to share the port
+    // state change notification data between producer and consumer threads.
+    class PortStateChangeHandler
+    {
+        public:
+
+            using portOperStatusNotificationQueue =
+                    ConcurrentQueue<sai_port_oper_status_notification_t>;
+
+            PortStateChangeHandler(
+                    _In_ std::shared_ptr<swss::SelectableEvent> portStateChangeEvent);
+
+            virtual ~PortStateChangeHandler();
+
+            // Adds the port operational status notification data to a queue and generates a
+            // notification event.
+            void handlePortStateChangeNotification(
+                    _In_ uint32_t count,
+                    _In_ const sai_port_oper_status_notification_t *data);
+
+            // Returns the shared pointer of the queue.
+            std::shared_ptr<portOperStatusNotificationQueue> getQueue() const
+            {
+                SWSS_LOG_ENTER();
+
+                return m_portStateChangeQueue;
+            }
+
+        private:
+
+            // Choosing 4k max event queue size based on if we had 256 ports, it can
+            // accommodate on average 16 port events per ports in worst case.
+            static constexpr size_t m_portStateChangeQueueSize = 4096;
+
+            // SelectableEvent for producer to generate the event and for consumer to
+            // listen on.
+            std::shared_ptr<swss::SelectableEvent> m_portStateChangeEvent;
+
+            // Mutex protected queue to share the data between producer and consumer.
+            std::shared_ptr<portOperStatusNotificationQueue> m_portStateChangeQueue;
+    };
+}  // namespace syncd

unittest/syncd/Makefile.am

@@ -16,6 +16,7 @@ tests_SOURCES = main.cpp \
 				TestNotificationProcessor.cpp \
 				TestNotificationHandler.cpp \
 				TestMdioIpcServer.cpp \
+				TestPortStateChangeHandler.cpp \
 				TestVendorSai.cpp
 
 tests_CXXFLAGS = $(DBGFLAGS) $(AM_CXXFLAGS) $(CXXFLAGS_COMMON)

unittest/syncd/TestPortStateChangeHandler.cpp

@@ -0,0 +1,60 @@
+#include "PortStateChangeHandler.h"
+
+#include <gtest/gtest.h>
+
+using namespace syncd;
+
+constexpr size_t portStateChangeQueueSize = 4096;
+
+class PortStateChangeHandlerTest : public ::testing::Test
+{
+    protected:
+        PortStateChangeHandlerTest()
+            : m_portStateChangeHandler(std::make_shared<swss::SelectableEvent>())
+        {
+            SWSS_LOG_ENTER();
+        }
+
+        ~PortStateChangeHandlerTest() override
+        {
+            SWSS_LOG_ENTER();
+        }
+
+        PortStateChangeHandler m_portStateChangeHandler;
+};
+
+TEST_F(PortStateChangeHandlerTest, VerifyGetQueue)
+{
+    auto queue = m_portStateChangeHandler.getQueue();
+    EXPECT_EQ(queue->size(), 0);
+}
+
+TEST_F(PortStateChangeHandlerTest,
+       HandlePortStateChangeNotificationFailsOnEnqueuingData)
+{
+    auto queue = m_portStateChangeHandler.getQueue();
+    EXPECT_EQ(queue->size(), 0);
+
+    // Insert enough data in the queue so it reaches its capacity.
+    sai_port_oper_status_notification_t operStatus[portStateChangeQueueSize];
+    m_portStateChangeHandler.handlePortStateChangeNotification(
+              portStateChangeQueueSize, &operStatus[0]);
+    EXPECT_EQ(queue->size(), portStateChangeQueueSize);
+
+    // Since queue is at its maximum capacity, adding a new element should cause
+    // insert failure and new element should not get added.
+    m_portStateChangeHandler.handlePortStateChangeNotification(/*count=*/1,
+                                                               &operStatus[0]);
+    EXPECT_EQ(queue->size(), portStateChangeQueueSize);
+}
+
+TEST_F(PortStateChangeHandlerTest, HandlePortStateChangeNotificationSucceeds)
+{
+    auto queue = m_portStateChangeHandler.getQueue();
+    EXPECT_EQ(queue->size(), 0);
+
+    sai_port_oper_status_notification_t operStatus;
+    m_portStateChangeHandler.handlePortStateChangeNotification(/*count=*/1,
+                                                               &operStatus);
+    EXPECT_EQ(queue->size(), 1);
+}