JSON output: flush outputs to allow for more timely appearance of records in file output.

Author: norrisjeremy

src/plugins/output/json/src/File.cpp

@@ -220,6 +220,14 @@ File::process(const char *str, size_t len)
     return IPX_OK;
 }
 
+void
+File::flush(void)
+{
+    if (_file) {
+        fflush(_file);
+    }
+}
+
 /**
  * \brief Get a directory path for a time window
  * \param[in]  tm    Time window

src/plugins/output/json/src/File.hpp

@@ -61,6 +61,8 @@ class File : public Output {
 
     // Store a record to the file
     int process(const char *str, size_t len);
+
+    void flush();
 private:
     /** Minimal window size */
     const unsigned int _WINDOW_MIN_SIZE = 60; // seconds

src/plugins/output/json/src/Storage.cpp

@@ -149,6 +149,8 @@ Storage::records_store(ipx_msg_ipfix_t *msg, const fds_iemgr_t *iemgr)
 {
     // Process all data records
     const uint32_t rec_cnt = ipx_msg_ipfix_get_drec_cnt(msg);
+    bool flush = false;
+    int ret = IPX_OK;
 
     // Message header
     auto hdr = (fds_ipfix_msg_hdr*) ipx_msg_ipfix_get_packet(msg);
@@ -162,13 +164,16 @@ Storage::records_store(ipx_msg_ipfix_t *msg, const fds_iemgr_t *iemgr)
             continue;
         }
 
+        flush = true;
+
         // Convert the record
         convert(ipfix_rec->rec, iemgr, hdr, false);
 
         // Store it
         for (Output *output : m_outputs) {
             if (output->process(m_record.buffer, m_record.size_used) != IPX_OK) {
-                return IPX_ERR_DENIED;
+                ret = IPX_ERR_DENIED;
+                goto endloop;
             }
         }
 
@@ -183,12 +188,20 @@ Storage::records_store(ipx_msg_ipfix_t *msg, const fds_iemgr_t *iemgr)
         // Store it
         for (Output *output : m_outputs) {
             if (output->process(m_record.buffer, m_record.size_used) != IPX_OK) {
-                return IPX_ERR_DENIED;
+                ret = IPX_ERR_DENIED;
+                goto endloop;
             }
         }
     }
 
-    return IPX_OK;
+endloop:
+    if (flush) {
+        for (Output *output : m_outputs) {
+            output->flush();
+        }
+    }
+
+    return ret;
 }
 
 /**

src/plugins/output/json/src/Storage.hpp

@@ -74,6 +74,9 @@ class Output {
      */
     virtual int
     process(const char *str, size_t len) = 0;
+
+    virtual void
+    flush() {};
 };
 
 /** JSON converter and output manager                                                            */