LogSession.h

/*
 * Copyright (c) 2015 Pavel Vainerman.
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU Lesser General Public License as
 * published by the Free Software Foundation, version 2.1.
 *
 * This program 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
 * Lesser General Lesser Public License for more details.
 *
 * You should have received a copy of the GNU Lesser General Public License
 * along with this program. If not, see <http://www.gnu.org/licenses/>.
 */
// -------------------------------------------------------------------------
#ifndef LogSession_H_
#define LogSession_H_
// -------------------------------------------------------------------------
#include <string>
#include <memory>
#include <queue>
#include <ev++.h>
#include "Poco/Net/StreamSocket.h"
#include "Mutex.h"
#include "DebugStream.h"
#include "UTCPCore.h"
#include "UTCPStream.h"
#include "LogAgregator.h"
#ifndef DISABLE_REST_API
#include <Poco/JSON/Object.h>
#endif
// -------------------------------------------------------------------------
namespace uniset
{
 
    class LogSession
    {
        public:
 
            LogSession( const Poco::Net::StreamSocket& s, std::shared_ptr<DebugStream>& log, timeout_t cmdTimeout = 2000, timeout_t checkConnectionTime = 10000 );
            ~LogSession();
 
            typedef sigc::slot<void, LogSession*> FinalSlot;
            void connectFinalSession( FinalSlot sl ) noexcept;
 
            // сигнал о приходе команды: std::string func( LogSession*, command, logname );
            // \return какую-то информацию, которая будет послана client-у. Если return.empty(), то ничего послано не будет.
            typedef sigc::signal<std::string, LogSession*, LogServerTypes::Command, const std::string& > LogSessionCommand_Signal;
            LogSessionCommand_Signal signal_logsession_command();
 
            // прервать работу
            void cancel() noexcept;
 
            std::string getClientAddress() const noexcept;
 
            void setSessionLogLevel( Debug::type t ) noexcept;
            void addSessionLogLevel( Debug::type t ) noexcept;
            void delSessionLogLevel( Debug::type t ) noexcept;
 
            void setMaxBufSize( size_t num );
            size_t getMaxBufSize() const noexcept;
 
            // запуск обработки входящих запросов
            void run( const ev::loop_ref& loop ) noexcept;
            void terminate();
 
            bool isAcive() const noexcept;
 
            std::string name() const noexcept;
            std::string getShortInfo() noexcept;
 
#ifndef DISABLE_REST_API
            Poco::JSON::Object::Ptr httpGetShortInfo();
#endif
 
        protected:
            //      LogSession( ost::TCPSocket& server );
 
            void event( ev::async& watcher, int revents ) noexcept;
            void callback( ev::io& watcher, int revents ) noexcept;
            void readEvent( ev::io& watcher ) noexcept;
            void writeEvent( ev::io& watcher );
            size_t readData( unsigned char* buf, int len );
            void cmdProcessing( const std::string& cmdLogName, const LogServerTypes::lsMessage& msg );
            void onCmdTimeout( ev::timer& watcher, int revents ) noexcept;
            void onCheckConnectionTimer( ev::timer& watcher, int revents ) noexcept;
            void final() noexcept;
 
            void logOnEvent( const std::string& s ) noexcept;
 
            timeout_t cmdTimeout = { 2000 };
            double checkConnectionTime = { 10. }; // время на проверку живости соединения..(сек)
 
            // Т.к. сообщений может быть ОЧЕНЬ МНОГО.. сеть медленная
            // очередь будет не успевать рассасываться,
            // то потенциально может "скушаться" вся память.
            // Поэтому приходиться ограничить доступное количество записей.
            // Рассчитываем, что средний размер одного сообщения 150 символов (байт)
            // тогда выделяем буфер на 200 сообщений (~ 30кB)
            // На самом деле сообщения могут быть совершенно разные..
            size_t maxRecordsNum = { 30000 }; // максимальное количество сообщение в очереди
 
        private:
            std::queue<UTCPCore::Buffer*> logbuf;
            std::mutex logbuf_mutex;
            bool lostMsg = { false };
 
            // статистика по использованию буфера
            size_t maxCount = { 0 }; // максимальное количество побывавшее в очереди
            size_t minSizeMsg = { 0 }; // минимальная встретившаяся длинна сообщения
            size_t maxSizeMsg = { 0 }; // максимальная встретившаяся длинна сообщения
            size_t numLostMsg = { 0 }; // количество потерянных сообщений
 
            std::string peername = { "" };
            std::string caddr = { "" };
            std::shared_ptr<DebugStream> log;
            std::shared_ptr<LogAgregator> alog;
            sigc::connection conn;
 
            std::shared_ptr<UTCPStream> sock;
 
            ev::io  io;
            ev::timer  cmdTimer;
            ev::async  asyncEvent;
            ev::timer  checkConnectionTimer;
 
            FinalSlot slFin;
            std::atomic_bool cancelled = { false };
 
            LogSessionCommand_Signal m_command_sig;
 
            DebugStream mylog;
    };
    // -------------------------------------------------------------------------
} // end of uniset namespace
// -------------------------------------------------------------------------
#endif // LogSession_H_
// -------------------------------------------------------------------------