inputStream.h

Go to the documentation of this file.

 
#ifndef MAXOS_COMMON_INPUTSTREAM_H
#define MAXOS_COMMON_INPUTSTREAM_H
 
#include <cstdint>
#include <common/vector.h>
#include <common/string.h>
 
namespace MaxOS::common {
 
    // Forward declaration
    template<class Type> class GenericInputStream;
 
    template<class Type> class InputStreamEventHandler {
            friend class GenericInputStream<Type>;
 
        protected:
            common::Vector<GenericInputStream<Type>*> m_generic_input_streams;        
        public:
            InputStreamEventHandler();
            ~InputStreamEventHandler();
 
            virtual void on_stream_read(Type);
            virtual void on_end_of_stream(GenericInputStream<Type>*);
    };
 
    template<class Type, class ProcessorType> class InputStreamProcessor : public InputStreamEventHandler<Type>, public GenericInputStream<ProcessorType> {
        public:
            InputStreamProcessor();
            explicit InputStreamProcessor(InputStreamEventHandler<ProcessorType>* generic_stream_event_handler);
            ~InputStreamProcessor();
 
            void on_end_of_stream(GenericInputStream<Type>* stream) override;
    };
 
 
    template<class Type> class InputStreamSocket : public InputStreamProcessor<Type, Type> {
        public:
            InputStreamSocket();
            explicit InputStreamSocket(InputStreamEventHandler<Type>*);
            ~InputStreamSocket();
 
            void on_stream_read(Type) override;
    };
 
    template<class Type> class InputStreamBuffer : protected InputStreamProcessor<Type, Type*> {
        protected:
            Type m_buffer[10240];               
            int m_offset { 0 };                 
            Type m_event_fire_element;          
            Type m_termination_element;         
 
        public:
            InputStreamBuffer(Type event_fire_element, Type termination_element);
            ~InputStreamBuffer();
 
            void on_stream_read(Type) override;
            void on_end_of_stream(GenericInputStream<Type>*) override;
            void flush();
    };
 
    template<class Type> class GenericInputStream {
 
        protected:
            common::Vector<InputStreamEventHandler<Type>*> m_input_stream_event_handlers; 
 
        public:
            GenericInputStream();
            explicit GenericInputStream(InputStreamEventHandler<Type>*);
            ~GenericInputStream();
 
            void connect_input_stream_event_handler(InputStreamEventHandler<Type>*);
            void disconnect_input_stream_event_handler(InputStreamEventHandler<Type>*);
 
    };
 
    template<class Type> void operator >>(GenericInputStream<Type>& source, InputStreamEventHandler<Type>& input_stream_event_handler);
 
    template<class Type, class ProcessorType> GenericInputStream<ProcessorType>& operator >>(GenericInputStream<Type>& source, InputStreamProcessor<Type, ProcessorType>& processor);
 
    class InputStream : public GenericInputStream<string> {
        public:
            explicit InputStream(InputStreamEventHandler<string>*);
    };
 
 
 
    template<class Type> InputStreamEventHandler<Type>::InputStreamEventHandler() = default;
 
    template<class Type> InputStreamEventHandler<Type>::~InputStreamEventHandler() {
 
        // Disconnect the handler from all streams
        while(!m_generic_input_streams.empty())
            (*m_generic_input_streams.begin())->disconnect_input_stream_event_handler(this);
 
    }
 
    template<class Type> void InputStreamEventHandler<Type>::on_stream_read(Type) {
 
    }
 
    template<class Type> void InputStreamEventHandler<Type>::on_end_of_stream(GenericInputStream<Type>* stream) {
 
        // Remove the stream
        m_generic_input_streams.erase(stream);
    }
 
 
    template<class Type, class ProcessorType> InputStreamProcessor<Type, ProcessorType>::InputStreamProcessor()
            : InputStreamEventHandler<Type>(),
            GenericInputStream<ProcessorType>() {
 
    }
 
    template<class Type, class ProcessorType> InputStreamProcessor<Type, ProcessorType>::InputStreamProcessor(InputStreamEventHandler<ProcessorType>* generic_stream_event_handler)
            : InputStreamEventHandler<Type>(),
            GenericInputStream<ProcessorType>(generic_stream_event_handler) {
 
    }
 
    template<class Type, class ProcessorType> InputStreamProcessor<Type, ProcessorType>::~InputStreamProcessor() = default;
 
    template<class Type, class ProcessorType> void InputStreamProcessor<Type, ProcessorType>::on_end_of_stream(GenericInputStream<Type>* stream) {
 
        // Pass the end of stream event on to the handlers
        for(auto& event_handler : this->m_input_stream_event_handlers)
            event_handler->on_end_of_stream(this);
 
        // Remove the stream
        InputStreamEventHandler<Type>::on_end_of_stream(stream);
 
    }
 
    template<class Type> InputStreamSocket<Type>::InputStreamSocket() = default;
 
    template<class Type> InputStreamSocket<Type>::InputStreamSocket(InputStreamEventHandler<Type>* processor_handler)
            : InputStreamProcessor<Type, Type>(processor_handler) {
 
    }
 
    template<class Type> InputStreamSocket<Type>::~InputStreamSocket() = default;
 
    template<class Type> void InputStreamSocket<Type>::on_stream_read(Type read_element) {
 
        // Pass the read event on to the handlers
        for(auto& event_handler : this->m_input_stream_event_handlers)
            event_handler->on_stream_read(read_element);
 
    }
 
    template<class Type> InputStreamBuffer<Type>::InputStreamBuffer(Type event_fire_element, Type termination_element)
            : m_event_fire_element(event_fire_element),
            m_termination_element(termination_element) {
 
    }
 
    template<class Type> InputStreamBuffer<Type>::~InputStreamBuffer() = default;
 
    template<class Type> void InputStreamBuffer<Type>::on_stream_read(Type read_element) {
 
        // flush the buffer if the event fire element is read
        if(read_element == m_event_fire_element) {
            flush();
            return;
        }
 
        // Ensure the buffer is not full
        if(m_offset >= 10238) {
            flush();
        }
 
        // Add the element
        m_buffer[m_offset++] = read_element;
 
    }
 
    template<class Type> void InputStreamBuffer<Type>::on_end_of_stream(GenericInputStream<Type>* stream) {
 
        // flush the buffer if there is any data in it
        if(m_offset > 0)
            flush();
 
        // Pass the event on to the handlers and remove the stream
        InputStreamProcessor<Type, Type*>::on_end_of_stream(stream);
    }
 
    template<class Type> void InputStreamBuffer<Type>::flush() {
 
        // Ensure the buffer is not empty
        if(m_offset == 0)
            return;
 
        // Add the termination element to the buffer
        m_buffer[m_offset] = m_termination_element;
 
        // Fire the on read event
        InputStreamProcessor<Type, Type*>::on_stream_read(m_buffer);
 
        // Reset the offset
        m_offset = 0;
 
    }
 
    template<class Type> GenericInputStream<Type>::GenericInputStream() = default;
 
    template<class Type> GenericInputStream<Type>::GenericInputStream(InputStreamEventHandler<Type>* input_stream_event_handler) {
 
        // Connect the handler
        connect_input_stream_event_handler(input_stream_event_handler);
    }
 
    template<class Type> GenericInputStream<Type>::~GenericInputStream() {
 
        // Disconnect all handlers
        while(!m_input_stream_event_handlers.empty())
            disconnect_input_stream_event_handler(*(m_input_stream_event_handlers.begin()));
    }
 
    template<class Type> void GenericInputStream<Type>::connect_input_stream_event_handler(InputStreamEventHandler<Type>* input_stream_event_handler) {
 
        // Don't add the handler if it is already connected
        if(m_input_stream_event_handlers.find(input_stream_event_handler) != m_input_stream_event_handlers.end())
            return;
 
        // Add the handler
        m_input_stream_event_handlers.push_back(input_stream_event_handler);
 
    }
 
    template<class Type> void GenericInputStream<Type>::disconnect_input_stream_event_handler(InputStreamEventHandler<Type>* input_stream_event_handler) {
 
        // Don't remove the handler if it is not connected
        if(m_input_stream_event_handlers.find(input_stream_event_handler) == m_input_stream_event_handlers.end())
            return;
 
        // Remove the handler
        m_input_stream_event_handlers.erase(input_stream_event_handler);
 
        // Fire the end of stream event
        input_stream_event_handler->on_end_of_stream(this);
 
    }
}
 
#endif //MAXOS_COMMON_INPUTSTREAM_H