pipe_thread.h

#pragma once
 
#include <deque>
#include <mutex>
#include "thread.h"
#include "lib_begin.h"
 
namespace nes {
    // forward declaration of pipe output stream to avoid inclusion of pipe header here
    template<typename CharT, typename Traits> class basic_pipe_ostream;
    template <typename CharT, typename Traits> class basic_pipe_istream;
}
 
namespace cgv {
    namespace os {
        class CGV_API queued_output_thread : public cgv::os::thread
        {
            bool all_data_sent = false;
        protected:
            bool is_binary;
            bool connected = false;
            mutable cgv::os::mutex m;
            std::deque<std::pair<char*, size_t> > blocks;
            unsigned ms_to_wait;
            virtual bool connect_to_child_process() = 0;
            virtual void write_block_to_pipe(const char* data, size_t count) = 0;
            virtual void close() = 0;
        public:
            queued_output_thread(bool is_binary = true, unsigned _ms_to_wait = 20);
            void run();
            bool has_connection() const;
            bool send_block(const char* data, size_t count);
            size_t get_nr_blocks() const;
            size_t get_nr_bytes() const;
            void done();
        };
        class CGV_API named_pipe_output_thread : public cgv::os::queued_output_thread
        {
        protected:
            std::string pipe_name;
            nes::basic_pipe_ostream<char, std::char_traits<char>>* pipe_ptr = 0;
            bool connect_to_child_process();
            void write_block_to_pipe(const char* data, size_t count);
            void close();
        public:
            named_pipe_output_thread(const std::string& _pipe_name, bool is_binary = true, unsigned _ms_to_wait = 20);
            std::string get_pipe_path() const;
        };
 
        class CGV_API pipe_output_thread : public cgv::os::queued_output_thread
        {
        protected:
            std::string cmd;
            FILE* fp = 0;
            int result = -1;
            bool connect_to_child_process();
            void write_block_to_pipe(const char* data, size_t count);
            void close();
        public:
            pipe_output_thread(const std::string& _cmd, bool is_binary = true, unsigned _ms_to_wait = 20);
            int get_result() const;
        };
 
 
        class CGV_API queued_input_thread : public cgv::os::thread
        {
          protected:
            bool all_data_received = false;
            bool is_binary;
            size_t package_size;
            size_t package_index;
            size_t packages_per_block;
            mutable std::mutex mutex_packages;
            mutable std::mutex mutex_data_blocks;
            char* packages; 
            std::deque<std::vector<char>> data_blocks;
 
            unsigned ms_to_wait;
            virtual bool connect_to_source() = 0;
            virtual size_t read_package_from_pipe(char* buffer, size_t package_size) = 0;
            virtual void move_packages_to_data_blocks();
            virtual void close() = 0;
 
          public:
            queued_input_thread(bool _is_binary, size_t _package_size,
                                size_t _packeges_per_block, unsigned _ms_to_wait);
 
            void run();
            bool pop_data_block(char* buffer);
            size_t get_nr_blocks() const;
            void done();
        };
 
        class CGV_API named_pipe_input_thread : public cgv::os::queued_input_thread
        {
            protected:
            std::string pipe_name;
            nes::basic_pipe_istream<char, std::char_traits<char>>* pipe_ptr = 0;
            bool connect_to_source() override;
            size_t read_package_from_pipe(char* buffer,size_t package_size) override;
            void close() override;
 
          public:
            named_pipe_input_thread(const std::string& _pipe_name, bool is_binary, size_t _package_size,
                                    size_t _packeges_per_block, unsigned _ms_to_wait);
            std::string get_pipe_path() const;
        };
 
        class CGV_API pipe_input_thread : public cgv::os::queued_input_thread
        {
          protected:
            std::string cmd;
            FILE* fp = 0;
 
            int result = -1;
 
            bool connect_to_source() override;
            size_t read_package_from_pipe(char* buffer, size_t package_size) override;
            void close() override;
 
          public:
            pipe_input_thread(const std::string& _cmd, bool is_binary, size_t _package_size, size_t _packages_per_block,
                              unsigned _ms_to_wait);
            int get_result() const;
        };
    }
}
 
#include <cgv/config/lib_end.h>