Added support for message topic in sink and filter plugins. Plugin pr…

…otocol version 4

src/common.hpp

@@ -1,6 +1,8 @@
 #ifndef COMMON_HPP
 #define COMMON_HPP
 
+#define PLUGIN_PROTOCOL_VERSION 4
+
 /*!
 * @file common.hpp
 * @brief Common definitions for the pugg library

src/filter.hpp

@@ -64,9 +64,10 @@ class Filter {
    * data was loaded successfully, and false otherwise.
    *
    * @param data The input data
+   * @param topic The topic of the data
    * @return True if the data was loaded successfully, and false otherwise
    */
-  virtual return_type load_data(Tin const &data) = 0;
+  virtual return_type load_data(Tin const &data, std::string topic = "") = 0;
 
   /*!
    * Processes the input data
@@ -124,7 +125,7 @@ class Filter {
   /*!
    * Returns the plugin protocol version.
    */
-  static const int version = 3;
+  static const int version = PLUGIN_PROTOCOL_VERSION;
 
   /*!
    * Returns the plugin server name.

src/plugin/echoj.cpp

@@ -25,7 +25,7 @@ using json = nlohmann::json;
 class Echo : public Filter<json, json> {
 public:
   string kind() override { return PLUGIN_NAME; }
-  return_type load_data(json const &d) override {
+  return_type load_data(json const &d, string topic = "") override {
     _data = d;
     return return_type::success;
   }

src/plugin/running_avg.cpp

@@ -29,7 +29,7 @@ class RunningAverage : public Filter<json, json> {
   // We expect to have a dictionary of values in the input, and we feed them
   // into a map of double-ended queues (deques) to keep track of the last N
   // values for each key.
-  return_type load_data(json const &input) override {
+  return_type load_data(json const &input, string topic = "") override {
     if (input[_params["field"]].is_object() == false) {
       return return_type::error;
     }

src/plugin/to_console.cpp

@@ -25,9 +25,9 @@ using json = nlohmann::json;
 class ToConsole : public Sink<json> {
 public:
   string kind() override { return PLUGIN_NAME; }
-  return_type load_data(json const &d) override {
+  return_type load_data(json const &d, string topic = "") override {
     _data = d;
-    cout << "Data: " << _data << endl;
+    cout << "[" << topic << "] Data: " << _data << endl;
     return return_type::success;
   }
 

src/sink.hpp

@@ -63,9 +63,10 @@ class Sink {
    * data was loaded successfully, and false otherwise.
    *
    * @param data The input data
+   * @param topic The topic of the data
    * @return True if the data was loaded successfully, and false otherwise
    */
-  virtual return_type load_data(Tin const &data) = 0;
+  virtual return_type load_data(Tin const &data, std::string topic = "") = 0;
 
   /*!
    * Sets the parameters
@@ -112,7 +113,7 @@ class Sink {
   /*!
    * Returns the plugin protocol version.
    */
-  static const int version = 3;
+  static const int version = PLUGIN_PROTOCOL_VERSION;
 
   /*!
    * Returns the plugin server name.

src/source.hpp

@@ -118,7 +118,7 @@ class Source {
   bool dummy;
 
 
-  static const int version = 3;
+  static const int version = PLUGIN_PROTOCOL_VERSION;
 
   /*!
    * Returns the plugin server name.