telemetry/Service.h

Go to the documentation of this file.

#pragma once

#include <swarmio/services/PeriodicService.h>
#include <swarmio/services/discovery/Discoverable.h>
#include <swarmio/services/telemetry/UpdateAwaiter.h>
#include <swarmio/services/telemetry/Tracker.h>
#include <swarmio/services/telemetry/Observer.h>
#include <swarmio/data/Variant.pb.h>
#include <list>
#include <string>
#include <shared_mutex>
#include <thread>

namespace swarmio::services::telemetry 
{
    class SWARMIO_API Service final : public PeriodicService, public discovery::Discoverable
    {
        private:

            std::map<std::string, data::Variant> _values;

            std::shared_timed_mutex _valuesMutex;

            data::discovery::Schema _schema;

            std::set<std::string> _statusKeys;

            std::shared_timed_mutex _schemaMutex;

            std::list<Tracker> _trackers;

            std::shared_timed_mutex _trackersMutex;

            std::set<Observer*> _observers;

            std::shared_timed_mutex _observersMutex;

            std::map<const Node*, data::telemetry::Status> _reports;

            std::shared_timed_mutex _reportsMutex;

        protected:

            virtual void Update() override final;

        public:

            static UpdateAwaiter Subscribe(Endpoint* endpoint, const Node* node, uint32_t interval = 1)
            {
                std::list<std::string> emptyList;
                return Subscribe(endpoint, node, interval, emptyList);
            }

            static UpdateAwaiter Subscribe(Endpoint* endpoint, const Node* node, uint32_t interval, const std::list<std::string>& keys);

            Service(Endpoint* endpoint, std::chrono::milliseconds period = std::chrono::milliseconds(10))
                : PeriodicService(endpoint, period)
            { 
                FinishConstruction();
            }

            void SetValue(const std::string& key, const data::Variant& value)
            {
                std::unique_lock<std::shared_timed_mutex> guard(_valuesMutex);
                _values[key] = value;
            }

            void RemoveValue(const std::string& key)
            {
                std::unique_lock<std::shared_timed_mutex> guard(_valuesMutex);
                _values.erase(key);
            }

            void SetFieldDefinitionForKey(const std::string& key, const data::discovery::Field& field, bool includeInStatus)
            {
                std::unique_lock<std::shared_timed_mutex> guard(_schemaMutex);
                (*_schema.mutable_fields())[key] = field;
                if (includeInStatus)
                {
                    _statusKeys.insert(key);
                }
                else
                {
                    _statusKeys.erase(key);
                }
            }

            void RemoveFieldDefinitionForKey(const std::string& key)
            {
                std::unique_lock<std::shared_timed_mutex> guard(_schemaMutex);
                _schema.mutable_fields()->erase(key);
                _statusKeys.erase(key);
            }

            void RegisterObserver(Observer* observer)
            {
                std::unique_lock<std::shared_timed_mutex> guard(_observersMutex);
                _observers.insert(observer);
            }

            void UnregisterObserver(Observer* observer)
            {
                std::unique_lock<std::shared_timed_mutex> guard(_observersMutex);
                _observers.erase(observer);
            }

            data::telemetry::Status GetCachedStatus(const Node* node)
            {
                std::shared_lock<std::shared_timed_mutex> guard(_reportsMutex);
                return _reports[node];
            }

            virtual bool ReceiveMessage(const Node* sender, const data::Message* message) override;

            virtual void DescribeService(data::discovery::Response& descriptor) override;
    };
}