BOB command parser

BOB.cpp

@@ -1,3 +1,4 @@
+#include <string.h>
 #include "Log.h"
 #include "BOB.h"
 
@@ -5,15 +6,74 @@ namespace i2p
 {
 namespace client
 {
-	BOBDataStream::BOBDataStream (std::shared_ptr<boost::asio::ip::tcp::socket> socket,
-		std::shared_ptr<i2p::stream::Stream> 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<boost::asio::ip::tcp::socket> (m_Service);
-		m_Acceptor.async_accept (*newSocket, std::bind (&BOBCommandChannel::HandleAccept, this,
-			std::placeholders::_1, newSocket));
+		auto newSession = std::make_shared<BOBCommandSession> (*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<boost::asio::ip::tcp::socket> socket)
+	void BOBCommandChannel::HandleAccept(const boost::system::error_code& ecode, std::shared_ptr<BOBCommandSession> 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 ());

BOB.h

@@ -1,28 +1,45 @@
 #ifndef BOB_H__
 #define BOB_H__
 
+#include <inttypes.h>
 #include <thread>
 #include <memory>
-#include <list>
+#include <map>
+#include <string>
 #include <boost/asio.hpp>
-#include "Streaming.h"
+#include "I2PTunnel.h"
 
 namespace i2p
 {
 namespace client
 {
-	class BOBDataStream: public std::enable_shared_from_this<BOBDataStream>
+	const size_t BOB_COMMAND_BUFFER_SIZE = 1024;
+	const char BOB_COMMAND_ZAP[] = "zap";
+
+	class BOBCommandChannel;
+	class BOBCommandSession: public std::enable_shared_from_this<BOBCommandSession>
 	{
 		public:
 
-			BOBDataStream (std::shared_ptr<boost::asio::ip::tcp::socket> socket,
-				std::shared_ptr<i2p::stream::Stream> 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<boost::asio::ip::tcp::socket> m_Socket;
-			std::shared_ptr<i2p::stream::Stream> 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<std::string, BOBCommandHandler>& GetCommandHandlers () { return m_CommandHandlers; };
+
 		private:
 
 			void Run ();
 			void Accept ();
-			void HandleAccept(const boost::system::error_code& ecode, std::shared_ptr<boost::asio::ip::tcp::socket> socket);
+			void HandleAccept(const boost::system::error_code& ecode, std::shared_ptr<BOBCommandSession> 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<std::shared_ptr<BOBDataStream> > m_DataStreams;
+			std::map<std::string, I2PTunnel *> m_Tunnels;
+			std::map<std::string, BOBCommandHandler> m_CommandHandlers;
 	};	
 }
 }

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 ();