Merge pull request #68 from CESNET/unirec_output_fix

fixed unirec interface flow duplication when multiple plugins enabled

Author: cejkato2

output/unirec.cpp

@@ -94,8 +94,10 @@ UR_FIELDS (
 /**
  * \brief Constructor.
  */
-UnirecExporter::UnirecExporter() : m_basic_idx(-1), m_ext_cnt(0), m_ifc_map(nullptr),
-   m_tmplts(nullptr), m_records(nullptr), m_ifc_cnt(0), m_eof(false), m_odid(false), m_link_bit_field(0), m_dir_bit_field(0)
+UnirecExporter::UnirecExporter() : m_basic_idx(-1), m_ext_cnt(0),
+   m_ifc_map(nullptr), m_tmplts(nullptr), m_records(nullptr), m_ifc_cnt(0),
+   m_ext_id_flgs(nullptr), m_eof(false), m_odid(false), m_link_bit_field(0),
+   m_dir_bit_field(0)
 {
 }
 
@@ -189,6 +191,7 @@ void UnirecExporter::init(const char *params)
       m_tmplts = new ur_template_t*[m_ifc_cnt];
       m_records = new void*[m_ifc_cnt];
       m_ifc_map = new int[m_ext_cnt];
+      m_ext_id_flgs = new int[m_ext_cnt];
    } catch (std::bad_alloc &e) {
       throw PluginError("not enough memory");
    }
@@ -311,6 +314,7 @@ void UnirecExporter::close()
 
    m_basic_idx = -1;
    m_ifc_cnt = 0;
+   delete [] m_ext_id_flgs;
 }
 
 /**
@@ -359,10 +363,13 @@ int UnirecExporter::export_flow(const Flow &flow)
 
    m_flows_seen++;
    uint64_t tmplt_dbits = 0; // templates dirty bits
+   memset(m_ext_id_flgs, 0, sizeof(int) * m_ext_cnt); // in case one flow has multiple extension of same type
+   int ext_processed_cnd = 0;
    while (ext != nullptr) {
       if (ext->m_ext_id >= static_cast<int>(m_ext_cnt)) {
          throw PluginError("encountered invalid extension id");
       }
+      ext_processed_cnd++;
       int ifc_num = m_ifc_map[ext->m_ext_id];
       if (ifc_num >= 0) {
          tmplt_ptr = m_tmplts[ifc_num];
@@ -374,14 +381,24 @@ int UnirecExporter::export_flow(const Flow &flow)
             tmplt_dbits |= (1 << ifc_num);
          }
 
+         if (m_ext_id_flgs[ext->m_ext_id] == 1) {
+            // send the previously filled unirec record
+            trap_send(ifc_num, record_ptr, ur_rec_size(tmplt_ptr, record_ptr));
+         } else {
+            m_ext_id_flgs[ext->m_ext_id] = 1;
+         }
+
          fill_basic_flow(flow, tmplt_ptr, record_ptr);
          ext->fill_unirec(tmplt_ptr, record_ptr); /* Add each extension header into unirec record. */
-
-         trap_send(ifc_num, record_ptr, ur_rec_fixlen_size(tmplt_ptr) + ur_rec_varlen_size(tmplt_ptr, record_ptr));
       }
       ext = ext->m_next;
    }
-
+   //send the last record with all plugin data
+   for (size_t ifc_num = 0; ifc_num < m_ifc_cnt && !(m_basic_idx >= 0) && ext_processed_cnd > 0; ifc_num++) {
+      tmplt_ptr = m_tmplts[ifc_num];
+      record_ptr = m_records[ifc_num];
+      trap_send(ifc_num, record_ptr, ur_rec_size(tmplt_ptr, record_ptr));
+   }
    return 0;
 }
 

output/unirec.hpp

@@ -202,6 +202,7 @@ class UnirecExporter : public OutputPlugin
    ur_template_t **m_tmplts;    /**< Pointer to unirec templates. */
    void          **m_records;   /**< Pointer to unirec records. */
    size_t m_ifc_cnt;            /**< Number of output interfaces. */
+   int *m_ext_id_flgs;          /** flags of used extension during export*/
 
    bool m_eof;                  /**< Send eof when module exits. */
    bool m_odid;            /**< Export ODID field instead of LINK_BIT_FIELD. */