FOGL-6891: Handle storage service restart

C/services/common/include/notification_api.h

@@ -65,6 +65,7 @@ class NotificationApi
 		void		start();
 		void		startServer();
 		void		wait();
+		bool		serverDown();
 		void		stop();
 		void		stopServer();
 		unsigned short	getListenerPort();

C/services/common/include/notification_manager.h

@@ -50,6 +50,11 @@ class EvaluationType
 
 		EVAL_TYPE		getType() const { return m_type; };
 		time_t			getInterval() const { return m_interval; };
+		inline bool operator==(const EvaluationType& x) const
+		{
+			return ( x.m_type==m_type &&
+				x.m_interval==m_interval );
+		}
 
 	private:
 		EVAL_TYPE	m_type;
@@ -75,6 +80,13 @@ class NotificationDetail
 					getType() const { return m_value.getType(); };
 		const time_t		getInterval() const { return m_value.getInterval(); };
 
+		inline bool operator==(const NotificationDetail& x) const
+		{
+			return ( x.m_asset==m_asset &&
+				x.m_rule==m_rule &&
+				x.m_value==m_value );
+		}
+
 	private:
 		std::string		m_asset;
 		std::string		m_rule;
@@ -130,7 +142,18 @@ class NotificationRule : public NotificationElement
 		// Add an asset name
 		void			addAsset(NotificationDetail& info)
 		{
+			for (const auto & a : m_assets)
+			{
+				if (a==info)
+				{
+					Logger::getLogger()->info("NotificationRule::addAsset(): Asset %s already present in rule %s, not adding again, m_assets.size()=%d", 
+										info.getAssetName().c_str(), info.getRuleName().c_str(), m_assets.size());
+					return;
+				}
+			}
 			m_assets.push_back(info);
+			Logger::getLogger()->info("NotificationRule::addAsset(): Added asset %s to rule %s, m_assets.size()=%d", 
+										info.getAssetName().c_str(), info.getRuleName().c_str(), m_assets.size());
 		};
 		std::string		toJSON();
 		bool			isTimeBased() { return m_timeBased != 0; };

C/services/common/include/notification_service.h

@@ -35,7 +35,8 @@ class NotificationService : public ServiceAuthHandler
 		const std::string&	getName() { return m_name; };
 		bool 			start(std::string& coreAddress,
 					      unsigned short corePort);
-		void 			stop();
+		bool 			connectToStorage(std::map<std::thread::id, std::atomic<int>>* m);
+		void 			stop(bool unregisterSubscriptions=true);
 		void			shutdown();
 		void			restart();
 		bool			isRunning() { return !m_shutdown; };
@@ -58,6 +59,9 @@ class NotificationService : public ServiceAuthHandler
 		void			setDryRun() { m_dryRun = true; };
 		bool			sendToDispatcher(const string& path,
 							const string& payload);
+		bool			getStorageServiceRestartPendingFlag() { return m_storageServiceRestartPending; }
+		bool			setStorageServiceRestartPendingFlag() { m_storageServiceRestartPending = true; }
+		bool			resetStorageServiceRestartPendingFlag() { m_storageServiceRestartPending = false; }
 
 	private:
 		Logger*			m_logger;
@@ -74,5 +78,6 @@ class NotificationService : public ServiceAuthHandler
 		const std::string	m_token;
 		bool			m_dryRun;
 		bool			m_restartRequest;
+		bool			m_storageServiceRestartPending;
 };
 #endif

C/services/common/include/notification_subscription.h

@@ -84,7 +84,8 @@ class NotificationSubscription
 		void			lockSubscriptions() { m_subscriptionMutex.lock(); };
 		void			unlockSubscriptions() { m_subscriptionMutex.unlock(); };
 		void			removeSubscription(const string& assetName,
-							   const string& ruleName);
+							   const string& ruleName, bool storageServiceRestartPending=false);
+		void 			removeAllSubscriptions(bool storageServiceRestartPending);
 
 	private:
 		EvaluationType		getEvalType(const Value& value);

C/services/common/notification_api.cpp

@@ -244,13 +244,15 @@ void NotificationApi::start() {
  * Start method for HTTP server
  */
 void NotificationApi::startServer() {
+	Logger::getLogger()->info("NotificationApi::startServer()");
 	m_server->start();
 }
 
 /**
  * Stop method for HTTP server
  */
 void NotificationApi::stopServer() {
+	Logger::getLogger()->info("NotificationApi::stopServer()");
 	m_server->stop();
 }
 
@@ -269,6 +271,14 @@ void NotificationApi::wait() {
 	m_thread->join();
 }
 
+/**
+ * Check whether HTTP server has shutdown
+ */
+bool NotificationApi::serverDown() {
+	return m_thread->joinable();
+}
+
+
 /**
  * Initialise the API entry points for the common data resource and
  * the readings resource.
@@ -359,6 +369,7 @@ void NotificationApi::respond(shared_ptr<HttpServer::Response> response,
 void NotificationApi::processCallback(shared_ptr<HttpServer::Response> response,
 				      shared_ptr<HttpServer::Request> request)
 {
+	PRINT_FUNC;
 	try
 	{
 		// URL decode assetName

C/services/common/notification_manager.cpp

@@ -322,6 +322,7 @@ NotificationManager::NotificationManager(const std::string& serviceName,
  */
 NotificationManager::~NotificationManager()
 {
+	PRINT_FUNC;
 	lock_guard<mutex> guard(m_instancesMutex);
 	// Mark is instance as zombie
         for (auto it = m_instances.begin();