monitoring_manager.h

Go to the documentation of this file.

#pragma once
 
#include <chrono>
#include <condition_variable>
#include <cstddef>
#include <cstdint>
#include <expected>
#include <functional>
#include <map>
#include <mutex>
#include <nlohmann/json_fwd.hpp>
#include <optional>
#include <string>
#include <thread>
#include <vector>
 
#include "node/device_manager.h"
#include "node/monitoring.h"
#include "node/parameter_refresher.h"
 
namespace mm::node {
 
class MonitoringManager {
 public:
  using PublishFn = std::function<void(std::string topic, std::string json)>;
 
  explicit MonitoringManager(DeviceManager& deviceManager);
 
  ~MonitoringManager();
 
  MonitoringManager(const MonitoringManager&) = delete;
  MonitoringManager& operator=(const MonitoringManager&) = delete;
 
  void setPublish(PublishFn publish);
 
  std::expected<Monitoring, std::string> create(Monitoring config);
 
  bool remove(const std::string& topic);
 
  std::optional<nlohmann::json> get(const std::string& topic) const;
 
  nlohmann::json list() const;
 
  void start();
 
  void stop();
 
  void keepFresh(uint16_t devicePosition, uint16_t index, uint8_t subindex,
                 std::chrono::milliseconds period);
 
  void stopKeepingFresh(uint16_t devicePosition, uint16_t index, uint8_t subindex);
 
  void sampleAll();
 
  std::size_t monitoringCount() const;
 
  std::size_t polledSdoCount() const;
 
 private:
  enum class Source { Pdo, Sdo };
 
  struct ParamPlan {
    uint16_t devicePosition = 0;
    uint16_t index = 0;
    uint8_t subindex = 0;
    Source source = Source::Pdo;
    std::optional<DeviceManager::PdoSampleSpec> pdoSpec;  // PDO only
  };
 
  struct Sample {
    int64_t timestampUs = 0;
    std::vector<std::optional<DeviceParameterValue>> values;
  };
 
  struct Entry {
    Monitoring config;
    std::vector<ParamPlan> plans;
    uint64_t epoch = 0;            // identifies this registration across a remove + re-create
    uint64_t cursor = 0;           // next recorder sequence number to deliver ([cursor, head))
    bool cursorPrimed = false;     // false until the first flush seeds cursor from recorderHead()
    uint64_t imageGeneration = 0;  // processImageGeneration the PDO specs were captured under
    std::chrono::steady_clock::time_point nextDue{};  // default (epoch) => due on the next wake
  };
 
  struct FlushState {
    std::string topic;
    uint64_t epoch = 0;
    std::chrono::milliseconds interval{};  // poll period handed to the refresher on an SDO switch
    std::vector<ParamPlan> plans;
    uint64_t cursor = 0;
    bool cursorPrimed = false;
    uint64_t imageGeneration = 0;
  };
 
  std::vector<FlushState> takeDue(std::chrono::steady_clock::time_point now, bool forceAll);
 
  void flushDetached(FlushState& state, const PublishFn& publish);
 
  void commitFlush(const FlushState& state);
 
  void flushDue(std::unique_lock<std::mutex>& lock, std::chrono::steady_clock::time_point now,
                bool forceAll);
 
  void recaptureIfRemapped(FlushState& state);
 
  static nlohmann::json resourceJson(const Entry& entry);  // assumes mutex_ held
 
  void run();
 
  DeviceManager& deviceManager_;
  ParameterRefresher refresher_;
  mutable std::mutex mutex_;    
  std::condition_variable cv_;  
  std::map<std::string, Entry> entries_;
  uint64_t nextEpoch_ = 1;  
  PublishFn publish_;
  bool running_ = false;  
  std::thread thread_;
};
 
}  // namespace mm::node