Hide boost::iostreams implementation behind smart pointers (#187)

This allows eventually rewriting the file IO ABI while ensuring STL iostream compatibility.

Author: arahlin

core/include/core/G3Reader.h

@@ -3,7 +3,7 @@
 
 #include <string>
 #include <vector>
-#include <dataio.h>
+#include <iostream>
 
 #include <G3Module.h>
 
@@ -25,7 +25,7 @@ class G3Reader : public G3Module {
 	bool prefix_file_;
 	std::string cur_file_;
 	std::deque<std::string> filename_;
-	g3_istream stream_;
+	std::shared_ptr<std::istream> stream_;
 	int n_frames_to_read_;
 	int n_frames_read_;
 	int n_frames_cur_;

core/include/core/G3Writer.h

@@ -2,7 +2,7 @@
 #define _G3_ARCWRITER_H
 
 #include <string>
-#include <dataio.h>
+#include <iostream>
 
 #include <G3Module.h>
 
@@ -13,12 +13,11 @@ class G3Writer : public G3Module {
 	// Writes to file <filename> all frames with types in <streams>.
 	// If <streams> is empty (default), writes all frames.
 
-	virtual ~G3Writer();
 	void Process(G3FramePtr frame, std::deque<G3FramePtr> &out);
 	void Flush();
 private:
 	std::string filename_;
-	g3_ostream stream_;
+	std::shared_ptr<std::ostream> stream_;
 	std::vector<G3Frame::FrameType> streams_;
 
 	SET_LOGGER("G3Writer");

core/include/core/dataio.h

@@ -2,19 +2,13 @@
 #define _G3_DATAIO_H
 
 #include <string>
-#include <vector>
-#include <boost/iostreams/filtering_stream.hpp>
-
-typedef boost::iostreams::filtering_istream g3_istream;
-typedef boost::iostreams::filtering_ostream g3_ostream;
+#include <memory>
+#include <iostream>
 
 /**
  * Configure a filtering stream for G3Frame decompression from a local or remote
  * file source.
  *
- * @param  stream   A reference to the filtering istream that will be configured
- *                  by this function.  Must be instantiated prior to this
- *                  function.
  * @param  path     A valid filename on disk, or a TCP socket address.  If a
  *                  filename, the compression scheme is determined from the file
  *                  extension.  Supported compression schemes are gzip or bzip2.
@@ -25,36 +19,52 @@ typedef boost::iostreams::filtering_ostream g3_ostream;
  *                  read until EOF.
  * @param  timeout  Timeout in seconds for socket connections.
  * @param  buffersize Advisory buffer size in bytes for aggregating reads
- * @return File descriptor for socket connections, or -1 for file input.
+ * @return stream   The input stream configured by this function.
+ */
+std::shared_ptr<std::istream>
+g3_istream_from_path(const std::string &path, float timeout=-1.0,
+    size_t buffersize=1024*1024);
+
+/**
+ * Return the file descriptor handle for socket connections.
+ *
+ * @param  stream   A reference to the input stream, as configured by
+ *                  g3_istream_from_path.
+ * @return fd       The socket file descriptor.
  */
-int g3_istream_from_path(g3_istream &stream, const std::string &path,
-    float timeout=-1.0, size_t buffersize=1024*1024);
+int g3_istream_handle(std::shared_ptr<std::istream> &stream);
 
 /**
  * Configure a filtering stream for G3Frame compression to a local file.
  *
- * @param  stream   A reference to the filtering ostream that will be configured
- *                  by this function.  Must be instantiated prior to this
- *                  function.
  * @param  path     A valid filename on disk.  If a filename, the compression
  *                  scheme is determined from the file extension.  Supported
  *                  compression schemes are gzip or bzip2.
  * @param  append   If true, append to an existing file on disk.  Otherwise,
  *                  Create a new file or overwrite an existing file.
  * @param  counter  If true, add a counter filter to the stream configuration,
  *                  for use by the g3_ostream_count function.
+ * @return stream   The output stream configured by this function.
  */
-void g3_ostream_to_path(g3_ostream &stream, const std::string &path,
-    bool append=false, bool counter=false);
+std::shared_ptr<std::ostream>
+g3_ostream_to_path(const std::string &path, bool append=false, bool counter=false);
 
 /**
  * Count the number of bytes written to the output file stream.
  *
- * @param  stream   A reference to the filtering ostream, as configured by
+ * @param  stream   A reference to the output stream, as configured by
  *                  g3_ostream_to_path with the counter argument set to true.
  * @return Number of bytes written to disk.
  */
-size_t g3_ostream_count(g3_ostream &stream);
+size_t g3_ostream_count(std::shared_ptr<std::ostream> &stream);
+
+/**
+ * Flush the output file stream.
+ *
+ * @param  stream   A reference to the output stream, as configured by
+ *                  g3_ostream_to_path.
+ */
+void g3_ostream_flush(std::shared_ptr<std::ostream> &stream);
 
 /**
  * Check that the input filename is a valid filename on disk.

core/src/G3Frame.cxx

@@ -3,7 +3,6 @@
 #include <G3Quat.h>
 #include <serialization.h>
 #include <pybindings.h>
-#include <dataio.h>
 
 #include <stdlib.h>
 #include <cxxabi.h>
@@ -382,12 +381,12 @@ void G3Frame::blob_encode(struct blob_container &blob)
 	item_os.flush();
 }
 
-template void G3Frame::loads(g3_istream &);
+template <> void G3Frame::loads(std::shared_ptr<std::istream> &is) { loads(*is); }
 template void G3Frame::loads(G3BufferInputStream &);
 template void G3Frame::loads(std::istream &);
 template void G3Frame::loads(std::istringstream &);
 
-template void G3Frame::saves(g3_ostream &) const;
+template <> void G3Frame::saves(std::shared_ptr<std::ostream> &os) const { saves(*os); }
 template void G3Frame::saves(G3BufferOutputStream &) const;
 template void G3Frame::saves(std::ostream &) const;
 template void G3Frame::saves(std::ostringstream &) const;

core/src/G3MultiFileWriter.cxx

@@ -9,7 +9,6 @@ class G3MultiFileWriter : public G3Module {
 	G3MultiFileWriter(boost::python::object filename,
 	    size_t size_limit,
 	    boost::python::object divide_on = boost::python::object());
-	virtual ~G3MultiFileWriter();
 	void Process(G3FramePtr frame, std::deque<G3FramePtr> &out);
 	std::string CurrentFile() { return current_filename_; }
 private:
@@ -23,7 +22,7 @@ class G3MultiFileWriter : public G3Module {
 	std::vector<G3Frame::FrameType> always_break_on_;
 	boost::python::object newfile_callback_;
 
-	g3_ostream stream_;
+	std::shared_ptr<std::ostream> stream_;
 	std::vector<G3FramePtr> metadata_cache_;
 	int seqno;
 
@@ -78,7 +77,7 @@ G3MultiFileWriter::CheckNewFile(G3FramePtr frame)
 {
 	// If we are already saving data, check file size. Otherwise, open
 	// a new file unconditionally.
-	if (!stream_.empty()) {
+	if (stream_ != nullptr) {
 		bool start_new_ = false;
 
 		if (g3_ostream_count(stream_) > size_limit_)
@@ -117,19 +116,14 @@ G3MultiFileWriter::CheckNewFile(G3FramePtr frame)
 	}
 
 	current_filename_ = filename;
-	g3_ostream_to_path(stream_, filename, false, true);
+	stream_ = g3_ostream_to_path(filename, false, true);
 
 	for (auto i = metadata_cache_.begin(); i != metadata_cache_.end(); i++)
 		(*i)->saves(stream_);
 
 	return true;
 }
 
-G3MultiFileWriter::~G3MultiFileWriter()
-{
-	stream_.reset();
-}
-
 void G3MultiFileWriter::Process(G3FramePtr frame, std::deque<G3FramePtr> &out)
 {
 	bool new_file(false), meta_cached(false);

core/src/G3Reader.cxx

@@ -34,7 +34,7 @@ void G3Reader::StartFile(std::string path)
 	log_info("Starting file %s\n", path.c_str());
 	cur_file_ = path;
 	n_frames_cur_ = 0;
-	(void) g3_istream_from_path(stream_, path, timeout_, buffersize_);
+	stream_ = g3_istream_from_path(path, timeout_, buffersize_);
 }
 
 void G3Reader::Process(G3FramePtr frame, std::deque<G3FramePtr> &out)
@@ -71,7 +71,7 @@ void G3Reader::Process(G3FramePtr frame, std::deque<G3FramePtr> &out)
 	// function reacquire the lock, if it was released.
 	G3PythonContext ctx("G3Reader", false);
 
-	while (stream_.peek() == EOF) {
+	while (stream_->peek() == EOF) {
 		if (n_frames_cur_ == 0)
 			log_error("Empty file %s", cur_file_.c_str());
 		if (filename_.size() > 0) {
@@ -100,14 +100,14 @@ void G3Reader::Process(G3FramePtr frame, std::deque<G3FramePtr> &out)
 }
 
 off_t G3Reader::Seek(off_t offset) {
-	if (stream_.peek() == EOF && offset != Tell())
+	if (stream_->peek() == EOF && offset != Tell())
 		log_fatal("Cannot seek %s; stream closed at EOF.", cur_file_.c_str());
-	stream_.seekg(offset, std::ios_base::beg);
+	stream_->seekg(offset, std::ios_base::beg);
 	return offset;
 }
 
 off_t G3Reader::Tell() {
-	return stream_.tellg();
+	return stream_->tellg();
 }
 
 PYBINDINGS("core") {

core/src/G3Writer.cxx

@@ -8,12 +8,7 @@ G3Writer::G3Writer(std::string filename,
     filename_(filename), streams_(streams)
 {
 	g3_check_output_path(filename);
-	g3_ostream_to_path(stream_, filename, append);
-}
-
-G3Writer::~G3Writer()
-{
-	stream_.reset();
+	stream_ = g3_ostream_to_path(filename, append);
 }
 
 void G3Writer::Process(G3FramePtr frame, std::deque<G3FramePtr> &out)
@@ -38,9 +33,10 @@ void G3Writer::Process(G3FramePtr frame, std::deque<G3FramePtr> &out)
 
 void G3Writer::Flush()
 {
-    if (!stream_.strict_sync()){
-        printf("There was a problem flushing the stream...\n");
-    }
+	try {
+		g3_ostream_flush(stream_);
+	} catch (...) {
+	}
 }
 
 PYBINDINGS("core") {