Add SSE style subscription support to eventservice

This commit adds the SSE style eventservice subscription style event
Using this, end user can subscribe for Redfish event logs using GET
on SSE uris from browser.

Tested:
 - From Browser did GET on above SSE URI and
   generated some Redfish event logs(power cycle)
   and saw redfish event logs streaming on browser.
 - After SSE registration, Check Subscription collections
   and GET on individual subscription and saw desired
   response.
 - Ran RedfishValidation and its passed.

Change-Id: I7f4b7a34974080739c4ba968ed570489af0474de
Signed-off-by: AppaRao Puli <[email protected]>
Signed-off-by: P Dheeraj Srujan Kumar <[email protected]>
Signed-off-by: Ed Tanous <[email protected]>

Author: apuli1

http/http_client.hpp

@@ -710,7 +710,7 @@ class ConnectionPool : public std::enable_shared_from_this<ConnectionPool>
         }
     }
 
-    void sendData(std::string& data, const std::string& destUri,
+    void sendData(std::string&& data, const std::string& destUri,
                   const boost::beast::http::fields& httpHeader,
                   const boost::beast::http::verb verb,
                   const std::function<void(Response&)>& resHandler)
@@ -866,19 +866,19 @@ class HttpClient
 
     // Send a request to destIP:destPort where additional processing of the
     // result is not required
-    void sendData(std::string& data, const std::string& destIP,
+    void sendData(std::string&& data, const std::string& destIP,
                   uint16_t destPort, const std::string& destUri, bool useSSL,
                   const boost::beast::http::fields& httpHeader,
                   const boost::beast::http::verb verb)
     {
         const std::function<void(Response&)> cb = genericResHandler;
-        sendDataWithCallback(data, destIP, destPort, destUri, useSSL,
+        sendDataWithCallback(std::move(data), destIP, destPort, destUri, useSSL,
                              httpHeader, verb, cb);
     }
 
     // Send request to destIP:destPort and use the provided callback to
     // handle the response
-    void sendDataWithCallback(std::string& data, const std::string& destIP,
+    void sendDataWithCallback(std::string&& data, const std::string& destIP,
                               uint16_t destPort, const std::string& destUri,
                               bool useSSL,
                               const boost::beast::http::fields& httpHeader,
@@ -897,7 +897,7 @@ class HttpClient
         }
         // Send the data using either the existing connection pool or the newly
         // created connection pool
-        pool.first->second->sendData(data, destUri, httpHeader, verb,
+        pool.first->second->sendData(std::move(data), destUri, httpHeader, verb,
                                      resHandler);
     }
 };

redfish-core/include/event_service_manager.hpp

@@ -24,6 +24,7 @@
 #include "registries.hpp"
 #include "registries/base_message_registry.hpp"
 #include "registries/openbmc_message_registry.hpp"
+#include "registries/privilege_registry.hpp"
 #include "registries/task_event_message_registry.hpp"
 #include "server_sent_events.hpp"
 #include "str_utility.hpp"
@@ -38,6 +39,7 @@
 #include <boost/url/format.hpp>
 #include <sdbusplus/bus/match.hpp>
 
+#include <algorithm>
 #include <cstdlib>
 #include <ctime>
 #include <fstream>
@@ -53,9 +55,13 @@ using ReadingsObjType =
 static constexpr const char* eventFormatType = "Event";
 static constexpr const char* metricReportFormatType = "MetricReport";
 
+static constexpr const char* subscriptionTypeSSE = "SSE";
 static constexpr const char* eventServiceFile =
     "/var/lib/bmcweb/eventservice_config.json";
 
+static constexpr const uint8_t maxNoOfSubscriptions = 20;
+static constexpr const uint8_t maxNoOfSSESubscriptions = 10;
+
 namespace registries
 {
 inline std::span<const MessageEntry>
@@ -382,15 +388,20 @@ class Subscription : public persistent_data::UserSubscription
                  boost::asio::io_context& ioc) :
         host(inHost),
         port(inPort), policy(std::make_shared<crow::ConnectionPolicy>()),
-        client(ioc, policy), path(inPath), uriProto(inUriProto)
+        path(inPath), uriProto(inUriProto)
     {
+        client.emplace(ioc, policy);
         // Subscription constructor
         policy->invalidResp = retryRespHandler;
     }
 
+    explicit Subscription(crow::sse_socket::Connection& connIn) :
+        sseConn(&connIn)
+    {}
+
     ~Subscription() = default;
 
-    bool sendEvent(std::string& msg)
+    bool sendEvent(std::string&& msg)
     {
         persistent_data::EventServiceConfig eventServiceConfig =
             persistent_data::EventServiceStore::getInstance()
@@ -402,13 +413,17 @@ class Subscription : public persistent_data::UserSubscription
 
         bool useSSL = (uriProto == "https");
         // A connection pool will be created if one does not already exist
-        client.sendData(msg, host, port, path, useSSL, httpHeaders,
-                        boost::beast::http::verb::post);
-        eventSeqNum++;
+        if (client)
+        {
+            client->sendData(std::move(msg), host, port, path, useSSL,
+                             httpHeaders, boost::beast::http::verb::post);
+            return true;
+        }
 
         if (sseConn != nullptr)
         {
-            sseConn->sendData(eventSeqNum, msg);
+            eventSeqNum++;
+            sseConn->sendEvent(std::to_string(eventSeqNum), msg);
         }
         return true;
     }
@@ -437,7 +452,7 @@ class Subscription : public persistent_data::UserSubscription
 
         std::string strMsg = msg.dump(2, ' ', true,
                                       nlohmann::json::error_handler_t::replace);
-        return this->sendEvent(strMsg);
+        return sendEvent(std::move(strMsg));
     }
 
 #ifndef BMCWEB_ENABLE_REDFISH_DBUS_LOG_ENTRIES
@@ -503,10 +518,10 @@ class Subscription : public persistent_data::UserSubscription
         msg["Id"] = std::to_string(eventSeqNum);
         msg["Name"] = "Event Log";
         msg["Events"] = logEntryArray;
-
         std::string strMsg = msg.dump(2, ' ', true,
                                       nlohmann::json::error_handler_t::replace);
-        this->sendEvent(strMsg);
+        sendEvent(std::move(strMsg));
+        eventSeqNum++;
     }
 #endif
 
@@ -546,7 +561,7 @@ class Subscription : public persistent_data::UserSubscription
 
         std::string strMsg = msg.dump(2, ' ', true,
                                       nlohmann::json::error_handler_t::replace);
-        this->sendEvent(strMsg);
+        sendEvent(std::move(strMsg));
     }
 
     void updateRetryConfig(uint32_t retryAttempts,
@@ -561,15 +576,32 @@ class Subscription : public persistent_data::UserSubscription
         return eventSeqNum;
     }
 
+    void setSubscriptionId(const std::string& id2)
+    {
+        BMCWEB_LOG_DEBUG << "Subscription ID: " << id2;
+        subId = id2;
+    }
+
+    std::string getSubscriptionId()
+    {
+        return subId;
+    }
+
+    bool matchSseId(const crow::sse_socket::Connection& thisConn)
+    {
+        return &thisConn == sseConn;
+    }
+
   private:
+    std::string subId;
     uint64_t eventSeqNum = 1;
     std::string host;
     uint16_t port = 0;
     std::shared_ptr<crow::ConnectionPolicy> policy;
-    crow::HttpClient client;
+    crow::sse_socket::Connection* sseConn = nullptr;
+    std::optional<crow::HttpClient> client;
     std::string path;
     std::string uriProto;
-    std::shared_ptr<crow::ServerSentEvents> sseConn = nullptr;
 
     // Check used to indicate what response codes are valid as part of our retry
     // policy.  2XX is considered acceptable
@@ -828,8 +860,8 @@ class EventServiceManager
             for (const auto& it :
                  EventServiceManager::getInstance().subscriptionsMap)
             {
-                std::shared_ptr<Subscription> entry = it.second;
-                entry->updateRetryConfig(retryAttempts, retryTimeoutInterval);
+                Subscription& entry = *it.second;
+                entry.updateRetryConfig(retryAttempts, retryTimeoutInterval);
             }
         }
     }
@@ -942,6 +974,8 @@ class EventServiceManager
         // Update retry configuration.
         subValue->updateRetryConfig(retryAttempts, retryTimeoutInterval);
 
+        // Set Subscription ID for back trace
+        subValue->setSubscriptionId(id);
         return id;
     }
 
@@ -966,11 +1000,38 @@ class EventServiceManager
         }
     }
 
-    size_t getNumberOfSubscriptions()
+    void deleteSseSubscription(const crow::sse_socket::Connection& thisConn)
+    {
+        for (const auto& it : subscriptionsMap)
+        {
+            std::shared_ptr<Subscription> entry = it.second;
+            bool entryIsThisConn = entry->matchSseId(thisConn);
+            if (entryIsThisConn)
+            {
+                persistent_data::EventServiceStore::getInstance()
+                    .subscriptionsConfigMap.erase(
+                        it.second->getSubscriptionId());
+                return;
+            }
+        }
+    }
+
+    size_t getNumberOfSubscriptions() const
     {
         return subscriptionsMap.size();
     }
 
+    size_t getNumberOfSSESubscriptions() const
+    {
+        auto size = std::count_if(
+            subscriptionsMap.begin(), subscriptionsMap.end(),
+            [](const std::pair<std::string, std::shared_ptr<Subscription>>&
+                   entry) {
+            return (entry.second->subscriptionType == subscriptionTypeSSE);
+            });
+        return static_cast<size_t>(size);
+    }
+
     std::vector<std::string> getAllIDs()
     {
         std::vector<std::string> idList;
@@ -981,7 +1042,7 @@ class EventServiceManager
         return idList;
     }
 
-    bool isDestinationExist(const std::string& destUrl)
+    bool isDestinationExist(const std::string& destUrl) const
     {
         for (const auto& it : subscriptionsMap)
         {
@@ -997,7 +1058,7 @@ class EventServiceManager
 
     bool sendTestEventLog()
     {
-        for (const auto& it : this->subscriptionsMap)
+        for (const auto& it : subscriptionsMap)
         {
             std::shared_ptr<Subscription> entry = it.second;
             if (!entry->sendTestEventLog())
@@ -1027,7 +1088,7 @@ class EventServiceManager
 
         eventRecord.emplace_back(std::move(eventMessage));
 
-        for (const auto& it : this->subscriptionsMap)
+        for (const auto& it : subscriptionsMap)
         {
             std::shared_ptr<Subscription> entry = it.second;
             bool isSubscribed = false;
@@ -1062,7 +1123,7 @@ class EventServiceManager
 
                 std::string strMsg = msgJson.dump(
                     2, ' ', true, nlohmann::json::error_handler_t::replace);
-                entry->sendEvent(strMsg);
+                entry->sendEvent(std::move(strMsg));
                 eventId++; // increament the eventId
             }
             else
@@ -1073,7 +1134,7 @@ class EventServiceManager
     }
     void sendBroadcastMsg(const std::string& broadcastMsg)
     {
-        for (const auto& it : this->subscriptionsMap)
+        for (const auto& it : subscriptionsMap)
         {
             std::shared_ptr<Subscription> entry = it.second;
             nlohmann::json msgJson;
@@ -1085,7 +1146,7 @@ class EventServiceManager
 
             std::string strMsg = msgJson.dump(
                 2, ' ', true, nlohmann::json::error_handler_t::replace);
-            entry->sendEvent(strMsg);
+            entry->sendEvent(std::move(strMsg));
         }
     }
 
@@ -1188,7 +1249,7 @@ class EventServiceManager
             return;
         }
 
-        for (const auto& it : this->subscriptionsMap)
+        for (const auto& it : subscriptionsMap)
         {
             std::shared_ptr<Subscription> entry = it.second;
             if (entry->eventFormatType == "Event")

redfish-core/include/redfish.hpp

@@ -24,6 +24,7 @@
 #include "environment_metrics.hpp"
 #include "ethernet.hpp"
 #include "event_service.hpp"
+#include "eventservice_sse.hpp"
 #include "fabric_adapters.hpp"
 #include "hypervisor_system.hpp"
 #include "log_services.hpp"
@@ -222,6 +223,7 @@ class RedfishService
         requestRoutesTaskCollection(app);
         requestRoutesTask(app);
         requestRoutesEventService(app);
+        requestRoutesEventServiceSse(app);
         requestRoutesEventDestinationCollection(app);
         requestRoutesEventDestination(app);
         requestRoutesFabricAdapters(app);

redfish-core/include/redfish_aggregator.hpp

@@ -702,10 +702,10 @@ class RedfishAggregator
             std::bind_front(processResponse, prefix, asyncResp);
 
         std::string data = thisReq.req.body();
-        client.sendDataWithCallback(data, std::string(sat->second.host()),
-                                    sat->second.port_number(), targetURI,
-                                    false /*useSSL*/, thisReq.fields(),
-                                    thisReq.method(), cb);
+        client.sendDataWithCallback(
+            std::move(data), std::string(sat->second.host()),
+            sat->second.port_number(), targetURI, false /*useSSL*/,
+            thisReq.fields(), thisReq.method(), cb);
     }
 
     // Forward a request for a collection URI to each known satellite BMC
@@ -721,10 +721,10 @@ class RedfishAggregator
 
             std::string targetURI(thisReq.target());
             std::string data = thisReq.req.body();
-            client.sendDataWithCallback(data, std::string(sat.second.host()),
-                                        sat.second.port_number(), targetURI,
-                                        false /*useSSL*/, thisReq.fields(),
-                                        thisReq.method(), cb);
+            client.sendDataWithCallback(
+                std::move(data), std::string(sat.second.host()),
+                sat.second.port_number(), targetURI, false /*useSSL*/,
+                thisReq.fields(), thisReq.method(), cb);
         }
     }
 

redfish-core/lib/event_service.hpp

@@ -43,8 +43,6 @@ static constexpr const std::array<const char*, 1> supportedResourceTypes = {
     "Task"};
 #endif
 
-static constexpr const uint8_t maxNoOfSubscriptions = 20;
-
 inline void requestRoutesEventService(App& app)
 {
     BMCWEB_ROUTE(app, "/redfish/v1/EventService/")
@@ -62,6 +60,9 @@ inline void requestRoutesEventService(App& app)
             "#EventService.v1_5_0.EventService";
         asyncResp->res.jsonValue["Id"] = "EventService";
         asyncResp->res.jsonValue["Name"] = "Event Service";
+        asyncResp->res.jsonValue["ServerSentEventUri"] =
+            "/redfish/v1/EventService/SSE";
+
         asyncResp->res.jsonValue["Subscriptions"]["@odata.id"] =
             "/redfish/v1/EventService/Subscriptions";
         asyncResp->res