diff --git a/BOB.cpp b/BOB.cpp index 00957a0b..0a908ad2 100644 --- a/BOB.cpp +++ b/BOB.cpp @@ -1,3 +1,4 @@ +#include #include "Log.h" #include "BOB.h" @@ -5,15 +6,74 @@ namespace i2p { namespace client { - BOBDataStream::BOBDataStream (std::shared_ptr socket, - std::shared_ptr stream): m_Socket (socket), m_Stream (stream) + BOBCommandSession::BOBCommandSession (BOBCommandChannel& owner): + m_Owner (owner), m_Socket (m_Owner.GetService ()), m_ReceiveBufferOffset (0), + m_IsOpen (true) { } + BOBCommandSession::~BOBCommandSession () + { + } + + void BOBCommandSession::Receive () + { + m_Socket.async_read_some (boost::asio::buffer(m_ReceiveBuffer + m_ReceiveBufferOffset, BOB_COMMAND_BUFFER_SIZE - m_ReceiveBufferOffset), + std::bind(&BOBCommandSession::HandleReceived, shared_from_this (), + std::placeholders::_1, std::placeholders::_2)); + } + + void BOBCommandSession::HandleReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred) + { + if (ecode) + LogPrint ("BOB command channel read error: ", ecode.message ()); + else + { + size_t size = m_ReceiveBufferOffset + bytes_transferred; + m_ReceiveBuffer[size] = 0; + char * eol = strchr (m_ReceiveBuffer, '\n'); + if (eol) + { + *eol = 0; + char * operand = strchr (m_ReceiveBuffer, ' '); + if (operand) *operand = 0; + // process command + auto handlers = m_Owner.GetCommandHandlers (); + auto it = handlers.find (m_ReceiveBuffer); + if (it != handlers.end ()) + (this->*(it->second))(operand, operand ? eol - operand : 0); + else + LogPrint (eLogError, "BOB unknown command", m_ReceiveBuffer); + + m_ReceiveBufferOffset = size - (eol - m_ReceiveBuffer) - 1; + memmove (m_ReceiveBuffer, eol + 1, m_ReceiveBufferOffset); + } + else + { + if (size < BOB_COMMAND_BUFFER_SIZE) + m_ReceiveBufferOffset = size; + else + { + LogPrint (eLogError, "Malformed input of the BOB command channel"); + return; + } + } + if (m_IsOpen) + Receive (); + } + } + + void BOBCommandSession::ZapCommandHandler (const char * operand, size_t len) + { + LogPrint (eLogDebug, "BOB: zap"); + m_IsOpen = false; + } + BOBCommandChannel::BOBCommandChannel (int port): m_IsRunning (false), m_Thread (nullptr), m_Acceptor (m_Service, boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port)) { + m_CommandHandlers[BOB_COMMAND_ZAP] = &BOBCommandSession::ZapCommandHandler; } BOBCommandChannel::~BOBCommandChannel () @@ -30,7 +90,9 @@ namespace client void BOBCommandChannel::Stop () { - m_DataStreams.clear (); + for (auto it: m_Tunnels) + delete it.second; + m_Tunnels.clear (); m_IsRunning = false; m_Service.stop (); if (m_Thread) @@ -58,20 +120,20 @@ namespace client void BOBCommandChannel::Accept () { - auto newSocket = std::make_shared (m_Service); - m_Acceptor.async_accept (*newSocket, std::bind (&BOBCommandChannel::HandleAccept, this, - std::placeholders::_1, newSocket)); + auto newSession = std::make_shared (*this); + m_Acceptor.async_accept (newSession->GetSocket (), std::bind (&BOBCommandChannel::HandleAccept, this, + std::placeholders::_1, newSession)); } - void BOBCommandChannel::HandleAccept(const boost::system::error_code& ecode, std::shared_ptr socket) + void BOBCommandChannel::HandleAccept(const boost::system::error_code& ecode, std::shared_ptr session) { if (ecode != boost::asio::error::operation_aborted) Accept (); if (!ecode) { - LogPrint (eLogInfo, "New BOB command connection from ", socket->remote_endpoint ()); - // TODO: + LogPrint (eLogInfo, "New BOB command connection from ", session->GetSocket ().remote_endpoint ()); + session->Receive (); } else LogPrint (eLogError, "BOB accept error: ", ecode.message ()); diff --git a/BOB.h b/BOB.h index d05df7cc..25407058 100644 --- a/BOB.h +++ b/BOB.h @@ -1,28 +1,45 @@ #ifndef BOB_H__ #define BOB_H__ +#include #include #include -#include +#include +#include #include -#include "Streaming.h" +#include "I2PTunnel.h" namespace i2p { namespace client { - class BOBDataStream: public std::enable_shared_from_this + const size_t BOB_COMMAND_BUFFER_SIZE = 1024; + const char BOB_COMMAND_ZAP[] = "zap"; + + class BOBCommandChannel; + class BOBCommandSession: public std::enable_shared_from_this { public: - BOBDataStream (std::shared_ptr socket, - std::shared_ptr stream); + BOBCommandSession (BOBCommandChannel& owner); + ~BOBCommandSession (); + + boost::asio::ip::tcp::socket& GetSocket () { return m_Socket; }; + void Receive (); + void HandleReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred); + + // command handlers + void ZapCommandHandler (const char * operand, size_t len); private: - std::shared_ptr m_Socket; - std::shared_ptr m_Stream; - }; + BOBCommandChannel& m_Owner; + boost::asio::ip::tcp::socket m_Socket; + char m_ReceiveBuffer[BOB_COMMAND_BUFFER_SIZE + 1]; + size_t m_ReceiveBufferOffset; + bool m_IsOpen; + }; + typedef void (BOBCommandSession::*BOBCommandHandler)(const char * operand, size_t len); class BOBCommandChannel { @@ -34,11 +51,14 @@ namespace client void Start (); void Stop (); + boost::asio::io_service& GetService () { return m_Service; }; + std::map& GetCommandHandlers () { return m_CommandHandlers; }; + private: void Run (); void Accept (); - void HandleAccept(const boost::system::error_code& ecode, std::shared_ptr socket); + void HandleAccept(const boost::system::error_code& ecode, std::shared_ptr session); private: @@ -46,7 +66,8 @@ namespace client std::thread * m_Thread; boost::asio::io_service m_Service; boost::asio::ip::tcp::acceptor m_Acceptor; - std::list > m_DataStreams; + std::map m_Tunnels; + std::map m_CommandHandlers; }; } } diff --git a/ClientContext.cpp b/ClientContext.cpp index bc2d37d4..eb824338 100644 --- a/ClientContext.cpp +++ b/ClientContext.cpp @@ -70,6 +70,13 @@ namespace client m_SamBridge->Start (); LogPrint("SAM bridge started"); } + int bobPort = i2p::util::config::GetArg("-bobport", 0); + if (bobPort) + { + m_BOBCommandChannel = new BOBCommandChannel (bobPort); + m_BOBCommandChannel->Start (); + LogPrint("BOB command channel started"); + } } void ClientContext::Stop () @@ -103,7 +110,14 @@ namespace client m_SamBridge = nullptr; LogPrint("SAM brdige stoped"); } - + if (m_BOBCommandChannel) + { + m_BOBCommandChannel->Stop (); + delete m_BOBCommandChannel; + m_BOBCommandChannel = nullptr; + LogPrint("BOB command channel stoped"); + } + for (auto it: m_Destinations) { it.second->Stop ();