add execute-mode support to the raw reader - allows to directly call commands and read their output.

Note that fdstream.h is from boost and has a separate license:
 * (C) Copyright Nicolai M. Josuttis 2001.
 * Permission to copy, use, modify, sell and distribute this software
 * is granted provided this copyright notice appears in all copies.
 * This software is provided "as is" without express or implied
 * warranty, and with no claim as to its suitability for any purpose.

src/input/fdstream.h

@@ -0,0 +1,184 @@
+/* The following code declares classes to read from and write to
+ * file descriptore or file handles.
+ *
+ * See
+ *      http://www.josuttis.com/cppcode
+ * for details and the latest version.
+ *
+ * - open:
+ *      - integrating BUFSIZ on some systems?
+ *      - optimized reading of multiple characters
+ *      - stream for reading AND writing
+ *      - i18n
+ *
+ * (C) Copyright Nicolai M. Josuttis 2001.
+ * Permission to copy, use, modify, sell and distribute this software
+ * is granted provided this copyright notice appears in all copies.
+ * This software is provided "as is" without express or implied
+ * warranty, and with no claim as to its suitability for any purpose.
+ *
+ * Version: Jul 28, 2002
+ * History:
+ *  Jul 28, 2002: bugfix memcpy() => memmove()
+ *                fdinbuf::underflow(): cast for return statements
+ *  Aug 05, 2001: first public version
+ */
+#ifndef BOOST_FDSTREAM_HPP
+#define BOOST_FDSTREAM_HPP
+
+#include <istream>
+#include <ostream>
+#include <streambuf>
+// for EOF:
+#include <cstdio>
+// for memmove():
+#include <cstring>
+
+
+// low-level read and write functions
+#ifdef _MSC_VER
+# include <io.h>
+#else
+# include <unistd.h>
+//extern "C" {
+//    int write (int fd, const char* buf, int num);
+//    int read (int fd, char* buf, int num);
+//}
+#endif
+
+
+// BEGIN namespace BOOST
+namespace boost {
+
+
+/************************************************************
+ * fdostream
+ * - a stream that writes on a file descriptor
+ ************************************************************/
+
+
+class fdoutbuf : public std::streambuf {
+  protected:
+    int fd;    // file descriptor
+  public:
+    // constructor
+    fdoutbuf (int _fd) : fd(_fd) {
+    }
+  protected:
+    // write one character
+    virtual int_type overflow (int_type c) {
+        if (c != EOF) {
+            char z = c;
+            if (write (fd, &z, 1) != 1) {
+                return EOF;
+            }
+        }
+        return c;
+    }
+    // write multiple characters
+    virtual
+    std::streamsize xsputn (const char* s,
+                            std::streamsize num) {
+        return write(fd,s,num);
+    }
+};
+
+class fdostream : public std::ostream {
+  protected:
+    fdoutbuf buf;
+  public:
+    fdostream (int fd) : std::ostream(0), buf(fd) {
+        rdbuf(&buf);
+    }
+};
+
+
+/************************************************************
+ * fdistream
+ * - a stream that reads on a file descriptor
+ ************************************************************/
+
+class fdinbuf : public std::streambuf {
+  protected:
+    int fd;    // file descriptor
+  protected:
+    /* data buffer:
+     * - at most, pbSize characters in putback area plus
+     * - at most, bufSize characters in ordinary read buffer
+     */
+    static const int pbSize = 4;        // size of putback area
+    static const int bufSize = 1024;    // size of the data buffer
+    char buffer[bufSize+pbSize];        // data buffer
+
+  public:
+    /* constructor
+     * - initialize file descriptor
+     * - initialize empty data buffer
+     * - no putback area
+     * => force underflow()
+     */
+    fdinbuf (int _fd) : fd(_fd) {
+        setg (buffer+pbSize,     // beginning of putback area
+              buffer+pbSize,     // read position
+              buffer+pbSize);    // end position
+    }
+
+  protected:
+    // insert new characters into the buffer
+    virtual int_type underflow () {
+#ifndef _MSC_VER
+        using std::memmove;
+#endif
+
+        // is read position before end of buffer?
+        if (gptr() < egptr()) {
+            return traits_type::to_int_type(*gptr());
+        }
+
+        /* process size of putback area
+         * - use number of characters read
+         * - but at most size of putback area
+         */
+        int numPutback;
+        numPutback = gptr() - eback();
+        if (numPutback > pbSize) {
+            numPutback = pbSize;
+        }
+
+        /* copy up to pbSize characters previously read into
+         * the putback area
+         */
+        memmove (buffer+(pbSize-numPutback), gptr()-numPutback,
+                numPutback);
+
+        // read at most bufSize new characters
+        int num;
+        num = read (fd, buffer+pbSize, bufSize);
+        if (num <= 0) {
+            // ERROR or EOF
+            return EOF;
+        }
+
+        // reset buffer pointers
+        setg (buffer+(pbSize-numPutback),   // beginning of putback area
+              buffer+pbSize,                // read position
+              buffer+pbSize+num);           // end of buffer
+
+        // return next character
+        return traits_type::to_int_type(*gptr());
+    }
+};
+
+class fdistream : public std::istream {
+  protected:
+    fdinbuf buf;
+  public:
+    fdistream (int fd) : std::istream(0), buf(fd) {
+        rdbuf(&buf);
+    }
+};
+
+
+} // END namespace boost
+
+#endif /*BOOST_FDSTREAM_HPP*/

src/input/readers/Raw.cc

@@ -7,10 +7,12 @@
 #include <sstream>
 
 #include "../../threading/SerialTypes.h"
+#include "../fdstream.h"
 
 #define MANUAL 0
 #define REREAD 1
 #define STREAM 2
+#define EXECUTE 3
 
 #include <sys/types.h>
 #include <sys/stat.h>
@@ -23,6 +25,7 @@ using threading::Field;
 Raw::Raw(ReaderFrontend *frontend) : ReaderBackend(frontend)
 {
 	file = 0;
+	in = 0;
 
 	//keyMap = new map<string, string>();
 	
@@ -41,9 +44,15 @@ Raw::~Raw()
 void Raw::DoFinish()
 {
 	if ( file != 0 ) {
-		file->close();
-		delete(file);
+		if ( mode != EXECUTE ) {
+			file->close();
+			delete(file);
+		} else { // mode == EXECUTE
+			delete(in);
+			pclose(pfile);
+		}
 		file = 0;
+		in = 0;
 	}
 }
 
@@ -53,15 +62,29 @@ bool Raw::DoInit(string path, int arg_mode, int arg_num_fields, const Field* con
 	mode = arg_mode;
 	mtime = 0;
 	
-	if ( ( mode != MANUAL ) && (mode != REREAD) && ( mode != STREAM ) ) {
+	if ( ( mode != MANUAL ) && (mode != REREAD) && ( mode != STREAM ) && ( mode != EXECUTE ) ) {
 		Error(Fmt("Unsupported read mode %d for source %s", mode, path.c_str()));
 		return false;
 	} 	
 
-	file = new ifstream(path.c_str());
-	if ( !file->is_open() ) {
-		Error(Fmt("Init: cannot open %s", fname.c_str()));
-		return false;
+	if ( mode != EXECUTE ) {
+
+		file = new ifstream(path.c_str());
+		if ( !file->is_open() ) {
+			Error(Fmt("Init: cannot open %s", fname.c_str()));
+			return false;
+		}
+		in = file;
+
+	} else { // mode == EXECUTE
+
+		pfile = popen(path.c_str(), "r");
+		if ( pfile == NULL ) {
+			Error(Fmt("Could not execute command %s", path.c_str()));
+			return false;
+		}
+
+		in = new boost::fdistream(fileno(pfile));
 	}
 	
 	num_fields = arg_num_fields;
@@ -81,23 +104,14 @@ bool Raw::DoInit(string path, int arg_mode, int arg_num_fields, const Field* con
 	Debug(DBG_INPUT, "Raw reader created, will perform first update");
 #endif
 
-	switch ( mode ) {
-		case MANUAL:
-		case REREAD:
-		case STREAM:
-			DoUpdate();
-			break;
-		default:
-			assert(false);
-	}
-
+	DoUpdate();
 
 	return true;
 }
 
 
 bool Raw::GetLine(string& str) {
-	while ( getline(*file, str, separator[0]) ) {
+	while ( getline(*in, str, separator[0]) ) {
 		return true;
 	}
 
@@ -141,6 +155,18 @@ bool Raw::DoUpdate() {
 				return false;
 			}
 
+			break;
+		case EXECUTE:
+			// re-execute it...
+			pclose(pfile);
+
+			pfile = popen(fname.c_str(), "r");
+			if ( pfile == NULL ) {
+				Error(Fmt("Could not execute command %s", fname.c_str()));
+				return false;
+			}
+
+			in = new boost::fdistream(fileno(pfile));
 			break;
 		default:
 			assert(false);
@@ -171,6 +197,7 @@ bool Raw::DoHeartbeat(double network_time, double current_time)
 
 	switch ( mode ) {
 		case MANUAL:
+		case EXECUTE:
 			// yay, we do nothing :)
 			break;
 		case REREAD:

src/input/readers/Raw.h

@@ -31,7 +31,11 @@ private:
 
 	bool GetLine(string& str);
 	
+	istream* in;
 	ifstream* file;
+
+	FILE* pfile;
+	
 	string fname;
 
 	// Options set from the script-level.