TCPEventStreamer.h

Go to the documentation of this file.

#ifndef TCP_EVENT_STREAMER_H
#define TCP_EVENT_STREAMER_H

#include "TCPEventStreamDefs.h"

class CRPTProxy;
class EventStore;

class TCPEventBuffer
{
public:
        virtual int numBytes() const = 0;
        virtual int numBytesRemaining() const = 0;
        virtual const unsigned char* currentBytes() const = 0;
        virtual void advance(int offset) = 0; 
        virtual bool done() const = 0; 
        virtual TCPEventBuffer& operator=(const TCPEventBuffer& eb) = 0;
        virtual ~TCPEventBuffer() { }
};

template <typename T>
class TCPEventBufferT : public TCPEventBuffer
{
private:
        uint32_t m_current_pos;   
        uint32_t m_num_bytes;    
        Poco::SharedPtr<T> m_ptr;   
        unsigned char* m_bytes;   
        TCPEventBufferT()  { throw std::runtime_error("TCPEventBufferT"); }

public:
        TCPEventBufferT(const TCPEventBufferT& eb) : m_current_pos(eb.m_current_pos), m_num_bytes(eb.m_num_bytes), m_ptr(eb.m_ptr), m_bytes(eb.m_bytes) { }
        TCPEventBufferT(T* b, uint32_t n) : m_current_pos(0), m_num_bytes(n*sizeof(T)), m_bytes(b), m_ptr(b) { }
        virtual int numBytes() const { return m_num_bytes; }
        virtual int numBytesRemaining() const { return m_num_bytes - m_current_pos; }
        virtual const unsigned char* currentBytes() const { return (m_current_pos < m_num_bytes ? m_bytes + m_current_pos : NULL); }
        virtual void advance(int offset) { m_current_pos += offset; }
        virtual bool done() const { return m_current_pos >= m_num_bytes; }
        virtual TCPEventBufferT& operator=(const TCPEventBufferT& eb)
        {
                if (this != &eb)
                {
                        m_current_pos = eb.m_current_pos;
                        m_num_bytes = eb.m_num_bytes;
                        m_bytes = eb.m_bytes;
                        m_ptr = eb.m_ptr;
                }
        }
};

class TCPEventStreamConnection : public Poco::Net::TCPServerConnection, ISIS::Base
{
private:
        EventStore& m_es;
        std::list<TCPStreamEventDataNeutron*> m_buffers;
        LONGLONG m_discarded_events;
        bool m_discarding_events;
        Poco::RWLock m_lock;
        uint32_t m_bytes_backlog;
        const CRPTProxy& m_crpt;
        static uint32_t MAX_BYTES_BACKLOG;

public:
        TCPEventStreamConnection(const Poco::Net::StreamSocket& s, EventStore& es, const CRPTProxy& crpt);
        ~TCPEventStreamConnection();
        void run();
        void allEventCallback(const DAEEventHeader* head, const DetectorEvent32* det_ev, int n, int event_source_id, const int* mapping);
};

class TCPEventStreamConnectionFactory: public Poco::Net::TCPServerConnectionFactory
{
        EventStore& m_es;
        const CRPTProxy& m_crpt;
public:
        TCPEventStreamConnectionFactory(EventStore& es, const CRPTProxy& crpt) : Poco::Net::TCPServerConnectionFactory(), m_es(es), m_crpt(crpt)
        {
        }
        
        Poco::Net::TCPServerConnection* createConnection(const Poco::Net::StreamSocket& socket)
        {
                return new TCPEventStreamConnection(socket, m_es, m_crpt);
        }
};

class TCPEventStreamer : public ISIS::Base
{
        Poco::Net::ServerSocket* m_svs;
        Poco::Net::TCPServer* m_srv;
//      std::list<int> m_events;
        EventStore& m_es;
        const CRPTProxy& m_crpt;
public:
        TCPEventStreamer(EventStore& es, const CRPTProxy& crpt);        
        ~TCPEventStreamer();
};


#endif /* TCP_EVENT_STREAMER_H */