streams.hpp

Go to the documentation of this file.

/*
 * Copyright 2008,2009,2010,2011,2012 Stephan Ossowski, Korbinian Schneeberger,
 * Felix Ott, Joerg Hagmann, Alf Scotland, Sebastian Bender
 * 
 * This file is part of SHORE.
 * 
 * SHORE is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 * 
 * SHORE is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 * GNU General Public License for more details.
 * 
 * You should have received a copy of the GNU General Public License
 * along with SHORE.  If not, see <http://www.gnu.org/licenses/>.
 */


#ifndef STREAMSET_HPP__
#define STREAMSET_HPP__


#ifdef HAVE_CONFIG_H
  #include "config.h"
#endif

#include <iostream>
#include <stdexcept>
#include <string>
#include <vector>

#include <boost/iostreams/stream.hpp>
#include <boost/iostreams/device/file_descriptor.hpp>
#include <boost/iostreams/device/null.hpp>
#include <boost/iostreams/concepts.hpp>
#include <boost/iostreams/categories.hpp>
#include <boost/thread/recursive_mutex.hpp>

#include "shore/base/stringops.hpp"
#include "shore/base/memops.hpp"

namespace shore {

class streams_base
{
 public:

        enum Codec
        {
                CODEC_PLAIN,
                CODEC_GZIP,
                CODEC_XZ
        };

        enum RandomAccessGranularity
        {
                GRANULARITY_FAST,
                GRANULARITY_MEDIUM,
                GRANULARITY_SLOW
        };

        static const int BLOCKSIZE_PRESET_VALUES[3];

        static const std::string COMPRESSION_FILEEXTS[3];
};

S_MAKE_ENUM_TRAITS(streams_base::Codec)
S_MAKE_ENUM_TRAITS(streams_base::RandomAccessGranularity)

class istreams
:public streams_base
{
 private:

        typedef boost::recursive_mutex mutex_t;
        typedef boost::recursive_mutex::scoped_lock lock_t;
        static int m_filecount;
        static mutex_t m_filecountmutex;

        struct stream_info
        {
                std::string streamname;
                std::vector<std::istream *> stream_stack;
                std::vector<int64_t> streamsizes;
                bool stream_seekable;
                bool delete_ptr;

                stream_info(const std::string &name);
        };

        std::vector<stream_info> m_streams;

        bool m_use_mmap;

        static bool filecount_inc();
        static void filecount_dec();
        std::istream *rawstream_alloc(const std::string &fn);


        istreams(const istreams &);
        istreams &operator=(const istreams &);

 public:

        static bool has_magic(std::istream *const in,const int m1,const int m2);

        static bool has_gzip_magic(std::istream *const in);

        static bool has_xz_magic(std::istream *const in);

        static bool is_gzip(const std::string &fn);

        static bool is_xz(const std::string &fn);

        static Codec get_codec(const std::string &fn);

        static uint64_t uncompressed_file_size(const std::string &fn);

        static void read_error_check(std::istream &s,
                                     const std::string &stream_name,
                                     const std::string &label=std::string());

        static void read_error_eof_check(std::istream &s,
                                         const std::string &stream_name,
                                         const std::string &label=std::string());

        static void read_error_check_rdbuf(std::istream &s,
                                           const std::string &stream_name,
                                           const std::string &label=std::string());

        static void save_stream_to_string(std::string &res,std::istream &s,
                                          const std::string &streamname);

        static void save_stream_to_string(std::string &res,const std::string &s);

        istreams();

        explicit istreams(const std::string& fn);

        explicit istreams(const std::vector<std::string>& fn);

        template<class Iter>
        istreams(Iter f_beg,Iter f_end)
        :m_use_mmap(false)
        {
                open_files(f_beg,f_end);
        }

        ~istreams();

        std::istream &open_file(const std::string &fnam);

        std::istream &open_file(const std::string &fnam,const Codec c);

        template<class Iter>
        void open_files(Iter f_beg,Iter f_end)
        {
                for(Iter itr=f_beg;itr!=f_end;++itr)
                {
                        open_file(*itr);
                }
        }

        std::istream &open_normalfile(const std::string &fn,const Codec c=CODEC_PLAIN);

        std::istream &push_back(std::istream& stream,const std::string &name=std::string(),
                       const Codec c=CODEC_PLAIN);

        std::istream& open_pipe(const std::string& command,const bool tolerant=false,
                                const Codec c=CODEC_PLAIN);

        void add_codec(const size_t index,const Codec c);

        void read_error_check(const int index,const std::string &label=std::string());

        void read_error_check_rdbuf(const int index,const std::string &label=std::string());

        void close_all();

        void close_stream(const std::istream*const s);

        void close_stream(const size_t i);
        void close_stream(const int i);

        void use_mmap(const bool b);

        size_t size() const;

        std::istream& operator[](size_t n);

        std::istream &front();

        std::istream &back();

        uint64_t get_streamsize(const size_t i) const;

        const std::string &get_streamname(const size_t i) const;

        bool is_raw_seekable(const size_t i) const;

        bool is_stream_seekable(const size_t i) const;
};

class ostreams
:public std::ostream,
 public streams_base
{
 public:

        enum FileExistsOp
        {
                EXISTS_FAIL,
                EXISTS_REPLACE,
                EXISTS_APPEND
        };

        // \brief   Automatically throw a (pretty uninformative) exception, e.g.
        //          if disk is full.
        enum ErrorCheck
        {
                ERRCHECK_MANUAL,
                ERRCHECK_AUTO
        };

 private:

        static Codec DEFAULT_ENC;
        static Codec DEFAULT_FASTENC;

        static size_t DEFAULT_ENCBS;

        struct stream_info
        {
                std::string streamname;
                std::vector<std::ostream *> stream_stack;
                bool delete_ptr;
                bool tempfile;
                std::string rename_name;
        };

        struct output_flushable
        :boost::iostreams::output,
         boost::iostreams::flushable_tag
        {};

        class streamset_sink
        :public boost::iostreams::device<output_flushable>
        {
         private:

                std::vector<stream_info> *const m_streams;

         public:

                streamset_sink(std::vector<stream_info> *const streams);

                std::streamsize write(const char* s,std::streamsize n);

                bool flush();

                void close();

                void imbue(const std::locale& l);
        };

        boost::iostreams::stream_buffer<streamset_sink>* m_streambuf;
        
        boost::iostreams::stream<boost::iostreams::null_sink> m_null;

        std::vector<stream_info> m_streams;

        size_t m_enc_bs;

        void init();


        ostreams(const ostreams &);
        ostreams &operator=(const ostreams &);

 public:

        static void set_default_enc(const Codec c);

        static void set_default_fastenc(const Codec c);

        static void set_default_enc_bs(const size_t t);

        static void set_default_enc_bs(const RandomAccessGranularity g);

        static Codec default_enc();

        static size_t default_enc_bs();

        static Codec default_fastenc();

        static const std::string &default_ext();

        static void write_error_check(std::ostream &s,
                                      const std::string &stream_name,
                                      const std::string &label=std::string());


        ostreams();

        explicit ostreams(const std::string &fn,const FileExistsOp exo=EXISTS_FAIL);

        explicit ostreams(const std::vector<std::string> &fn,
                          const FileExistsOp exo=EXISTS_FAIL);

        template<class Iter>
        ostreams(Iter f_beg,Iter f_end,const FileExistsOp exo=EXISTS_FAIL)
        :std::ostream(0)
        {
                init();
                open_files(f_beg,f_end,exo);
        }

        ~ostreams();

        void set_enc_blocksize(const size_t bs);

        void set_enc_blocksize(const RandomAccessGranularity g);

        std::ostream &open_file(const std::string& fn,
                                const FileExistsOp exo=EXISTS_FAIL,
                                const ErrorCheck ec=ERRCHECK_AUTO);

        std::ostream &open_file(const std::string& fn,const Codec c,
                                const FileExistsOp exo=EXISTS_FAIL,
                                const ErrorCheck ec=ERRCHECK_AUTO);

        template<class Iter>
        void open_files(Iter f_beg,Iter f_end,const FileExistsOp exo=EXISTS_FAIL)
        {
                for(Iter itr=f_beg;itr!=f_end;++itr)
                {
                        open_file(*itr,exo);
                }
        }

        std::ostream &open_normalfile(const std::string &fn,
                                      const Codec c=CODEC_PLAIN,
                                      const FileExistsOp exists_op=EXISTS_FAIL,
                                      const ErrorCheck ec=ERRCHECK_AUTO);

        std::ostream &open_tempfile(const std::string &pathprefix,
                                    const std::string &rename_name,
                                    const Codec c=default_fastenc(),
                                    const FileExistsOp exists_op=EXISTS_FAIL,
                                    const ErrorCheck ec=ERRCHECK_AUTO);

        std::ostream &open_tempfile(const std::string &pathprefix,
                                    const Codec c,
                                    const ErrorCheck ec=ERRCHECK_AUTO);

        std::ostream &open_tempfile(const std::string &pathprefix,
                                    const ErrorCheck ec=ERRCHECK_AUTO);

        std::ostream &open_tmpdir_file(const std::string &fn_prefix,
                                       const Codec c,
                                       const ErrorCheck ec=ERRCHECK_AUTO);

        std::ostream &open_tmpdir_file(const std::string &fn_prefix,
                                       const ErrorCheck ec=ERRCHECK_AUTO);

        std::ostream &open_pipe(const std::string &command,
                                const Codec c=CODEC_PLAIN,
                                const ErrorCheck ec=ERRCHECK_AUTO);

        std::ostream &push_back(std::ostream &os,
                                const std::string &name=std::string(),
                                const Codec c=CODEC_PLAIN,
                                const ErrorCheck ec=ERRCHECK_AUTO);

        void add_codec(const size_t index,const Codec c,
                       const ErrorCheck ec=ERRCHECK_AUTO);

        void write_error_check(const int index,const std::string &label=std::string());

        void close_all();

        void close_stream(const std::ostream*const s);

        void close_stream(const size_t i);

        void close_stream(const int i);

        size_t size() const;

        std::ostream &operator[](int n);

        std::ostream &front();

        std::ostream &back();

        const std::string &get_streamname(const size_t i) const;
};

class ipipe
:public boost::iostreams::stream<boost::iostreams::file_descriptor_source>
{
 private:

        typedef boost::iostreams::file_descriptor_source source_t;
        typedef boost::iostreams::stream<source_t> base_t;

        FILE* m_pipe;

        const std::string m_command;

        const bool m_tolerate_errexit;

 public:

        explicit ipipe(const std::string& command,const bool tolerant=false);
        ~ipipe();
};

class opipe
:public boost::iostreams::stream<boost::iostreams::file_descriptor_sink>
{
 private:

        typedef boost::iostreams::file_descriptor_sink sink_t;
        typedef boost::iostreams::stream<sink_t> base_t;

        FILE* m_pipe;

        const std::string m_command;

 public:

        explicit opipe(const std::string& command);
        ~opipe();
};

class mmapping_source
:public boost::iostreams::device<boost::iostreams::input_seekable>
{
 private:

        static const std::streamsize m_mmapsize;
        shore::mmapping m_mmpp;

        off_t m_pos;

        off_t m_blockofs;

 public:

        mmapping_source(const std::string &fn);

        std::streamsize read(char* s,std::streamsize n);

        std::streampos seek(boost::iostreams::stream_offset off,
                            std::ios_base::seekdir way);
};

class mmapping_ifstream
:public boost::iostreams::stream<mmapping_source>
{
 public:

        mmapping_ifstream(const std::string &fn);
};
 
} // namespace shore

#endif // STREAMSET_HPP__

// vim: sw=4:softtabstop=4:expandtab