topic_stats.hpp

Go to the documentation of this file.

/*
  _____           _         ____  _        _
 |_   _|__  _ __ (_) ___   / ___|| |_ __ _| |_ ___
   | |/ _ \| '_ \| |/ __|  \___ \| __/ _` | __/ __|
   | | (_) | |_) | | (__    ___) | || (_| | |_\__ \
   |_|\___/| .__/|_|\___|  |____/ \__\__,_|\__|___/
           |_|
 
Pure, sliding-window per-topic activity aggregator for `mads top`. No ZMQ, no
Agent dependency: callers feed it (topic, bytes[, sample]) tuples with an
explicit or implicit timestamp, and it reports messages/s and bytes/s
averaged over a trailing window, plus last-seen/last-sample bookkeeping that
never expires. This split keeps the counting logic directly unit-testable
with synthetic timestamps (no sockets, no real sleeps) while `top.cpp` stays
a thin CLI/subscribe/render loop.
 
Author(s): Paolo Bosetti
*/
#pragma once
 
#include <chrono>
#include <deque>
#include <map>
#include <mutex>
#include <string>
#include <vector>
 
namespace Mads {
 
struct TopicStat {
  std::string topic;
  size_t window_messages = 0;
  size_t window_bytes = 0;
  double messages_per_second = 0.0;
  double bytes_per_second = 0.0;
  size_t total_messages = 0;
  std::chrono::steady_clock::time_point last_seen{};
  size_t last_size = 0;
  std::string last_sample;
};
 
class TopicStats {
public:
  explicit TopicStats(
      std::chrono::milliseconds window = std::chrono::seconds(5));
 
  void record(const std::string &topic, size_t bytes,
              std::string sample = std::string(),
              std::chrono::steady_clock::time_point now =
                  std::chrono::steady_clock::now());
 
  std::vector<TopicStat> snapshot(std::chrono::steady_clock::time_point now =
                                       std::chrono::steady_clock::now()) const;
 
  void set_window(std::chrono::milliseconds window);
  std::chrono::milliseconds window() const;
 
  void clear();
 
private:
  struct Entry {
    // Ascending by timestamp (append-only in record(), pruned from the
    // front): (when, bytes) per observed message still within living memory
    // of the window.
    std::deque<std::pair<std::chrono::steady_clock::time_point, size_t>>
        events;
    size_t total_messages = 0;
    std::chrono::steady_clock::time_point last_seen{};
    size_t last_size = 0;
    std::string last_sample;
  };
 
  mutable std::mutex _mtx;
  std::chrono::milliseconds _window;
  // map (not unordered_map) so snapshot() iterates in topic-name order with
  // no extra sort step.
  std::map<std::string, Entry> _entries;
};
 
} // namespace Mads