TCPEventStreamer.cpp

Go to the documentation of this file.

#include "stdafx.h"
#include "CRPTProxy.h"
#include "event_store.h"
#include "NeXusEventCallback.h"
#include "TCPEventStreamer.h"

        TCPEventStreamConnection::TCPEventStreamConnection(const Poco::Net::StreamSocket& s, EventStore& es, const CRPTProxy& crpt) : TCPServerConnection(s), m_es(es), m_crpt(crpt), m_bytes_backlog(0), m_discarded_events(0), m_discarding_events(false)
        {
                setLoggerName("TCPEventStreamConnection");
        }
        
        TCPEventStreamConnection::~TCPEventStreamConnection()
        {
                if (m_bytes_backlog > 0)
                {
                        LOGSTR_INFORMATION("Bytes backlog unsent " << m_bytes_backlog);
                }
                if (m_buffers.size() > 0)
                {
                        LOGSTR_INFORMATION("Buffers backlog unsent " << m_buffers.size());
                }
                BOOST_FOREACH(TCPStreamEventDataNeutron* d, m_buffers)
                {
                        delete d;
                }
                m_buffers.clear();
        }

        void TCPEventStreamConnection::run()
        {
                const Poco::Timespan timeout(5,0);
                Poco::Net::StreamSocket& sock = this->socket();
                LOGSTR_INFORMATION("Received request from " + sock.peerAddress().toString());
                TCPStreamEventDataSetup setup_packet;
                setup_packet.head_setup.start_time = m_crpt->start_time;
                setup_packet.head_setup.run_number = m_crpt->run_number;
                int sent = sock.sendBytes(&setup_packet, sizeof(setup_packet));
                std::list<boost::signals2::connection> connect_list = m_es.addGoodEventCallback(boost::bind(&TCPEventStreamConnection::allEventCallback, this, _1, _2, _3, _4, _5));
                try
                {
                        while(true)
                        {
                                if ( (m_buffers.size() > 0) && sock.poll(timeout, Poco::Net::Socket::SELECT_WRITE) )
                                {
                                        std::list<TCPStreamEventDataNeutron*>::iterator it;
                                        {
                                                Poco::RWLock::ScopedReadLock _lock(m_lock);
                                                it = m_buffers.begin();
                                        }
                                        sent = sock.sendBytes(&((*it)->head), sizeof((*it)->head));
                                        sent = sock.sendBytes(&((*it)->head_n), sizeof((*it)->head_n));
                                        sent = sock.sendBytes(&((*it)->data[0]), (*it)->data.size() * sizeof((*it)->data[0]));
                                        {
                                                Poco::RWLock::ScopedWriteLock _lock(m_lock);
//                                              m_bytes_backlog -= it->numBytes(); 
                                                delete *it;
                                                m_buffers.erase(it);
                                        }
                                }
                                else
                                {
                                        Poco::Thread::sleep(2000);
                                }
                        }
                }
                catch(const std::exception& ex)
                {
                        LOGSTR_WARNING("Terminating request from " + sock.peerAddress().toString() + ": " + ex.what());
                }
                if (m_buffers.size() > 0)
                {
                        LOGSTR_INFORMATION("Buffers backlog unsent " << m_buffers.size());
                }
                LOGSTR_INFORMATION("Events discarded during session = " << m_discarded_events);
                m_es.clearCallbacks(connect_list);
                Poco::RWLock::ScopedWriteLock _lock(m_lock);
                BOOST_FOREACH(TCPStreamEventDataNeutron* d, m_buffers)
                {
                        delete d;
                }
                m_buffers.clear();
        }

#define PPP_TO_UAMPH    1.738E-6

        void TCPEventStreamConnection::allEventCallback(const DAEEventHeader* head, const DetectorEvent32* det_ev, int n, int event_source_id, const int* mapping)
        {
                if (n == 0)
                {
                        return;
                }
                if ( m_buffers.size() > 50 )
                {
#ifdef _WIN64
                        InterlockedExchangeAdd64(&m_discarded_events, n);
#endif /* _WIN64 */
                        if ( !m_discarding_events )
                        {
                                m_discarding_events = true;
                                LOGSTR_WARNING("Starting to discard new events for live listener");
                        }
                        return;
                }
                if ( m_discarding_events )
                {
                        m_discarding_events = false;
                        LOGSTR_NOTICE("Resumed sending events for live listener, total discarded so far = " << m_discarded_events);
                }
                float time_offset;
                DAEEventList::DAETimeToOffset(head->time, m_crpt->start_time, time_offset);
                TCPStreamEventHeader header(TCPStreamEventHeader::Neutron);
                TCPStreamEventHeaderNeutron header_n;
                header_n.frame_number = head->frame_number; 
                header_n.protons = head->protons * PPP_TO_UAMPH;
                header_n.frame_time_zero = time_offset;
                header_n.nevents = n;
                TCPStreamEventDataNeutron* t = new TCPStreamEventDataNeutron;
                t->data.resize(n);
                int card_index = m_crpt->cardIndexFromPos(event_source_id);
                int spec, tr;
                for(int i=0; i<n; ++i)
                {
                        spec = m_crpt->dae1SpecForCardIndex(card_index, det_ev[i].spectrum);
                        t->data[i].spectrum = spec;
                        tr = m_crpt->spectrumDAETR(spec);
                        t->data[i].time_of_flight = NeXusEventCallback::getTimeOffset(m_crpt.CRPT(), det_ev, event_source_id, tr, i); 
                }
                t->head = header;
                t->head_n = header_n;
                if ( t->isValid() )
                {
                        Poco::RWLock::ScopedWriteLock _lock(m_lock);
                        m_buffers.push_back(t);
                }
                else
                {
                        LOGSTR_WARNING("Invalid event packet");
                }
        }


uint32_t TCPEventStreamConnection::MAX_BYTES_BACKLOG = 10000;


        TCPEventStreamer::TCPEventStreamer(EventStore& es, const CRPTProxy& crpt) : m_es(es), m_crpt(crpt)
        {
                setLoggerName("TCPEventStreamer");
                LOGSTR_INFORMATION("Creating TCPEventStreamer");
                unsigned short port = 10000;

                // set-up a server socket
                m_svs = new Poco::Net::ServerSocket(port);
                // set-up a TCPServer instance
                m_srv = new Poco::Net::TCPServer(new TCPEventStreamConnectionFactory(m_es, m_crpt), *m_svs);
                // start the TCPServer
                m_srv->start();
        }

        TCPEventStreamer::~TCPEventStreamer()
        {
                LOGSTR_INFORMATION("Terminating TCPEventStreamer");
                m_srv->stop();
                delete m_srv;
                delete m_svs; 
        }