handle through traffic

SAM.cpp

@@ -1,3 +1,4 @@
+#include <string.h>
 #include <boost/bind.hpp>
 #include "Log.h"
 #include "SAM.h"
@@ -6,6 +7,131 @@ namespace i2p
 {
 namespace stream
 {
+	SAMSocket::SAMSocket (SAMBridge& owner): 
+		m_Owner (owner), m_Socket (m_Owner.GetService ()), m_Stream (nullptr)
+	{
+	}
+
+	SAMSocket::~SAMSocket ()
+	{
+		delete m_Stream;
+	}	
+
+	void SAMSocket::Terminate ()
+	{
+		if (m_Stream)
+		{
+			m_Stream->Close ();
+			delete m_Stream;
+			m_Stream = nullptr;
+		}
+		delete this;
+	}
+
+	void SAMSocket::ReceiveHandshake ()
+	{
+		m_Socket.async_read_some (boost::asio::buffer(m_Buffer, SAM_SOCKET_BUFFER_SIZE),                
+			boost::bind(&SAMSocket::HandleHandshakeReceived, this, 
+			boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
+	}
+
+	void SAMSocket::HandleHandshakeReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred)
+	{
+		if (ecode)
+        {
+			LogPrint ("SAM handshake read error: ", ecode.message ());
+			if (ecode != boost::asio::error::operation_aborted)
+				Terminate ();
+		}
+		else
+		{	
+			m_Buffer[bytes_transferred] = 0;
+			LogPrint ("SAM handshake ", m_Buffer);
+			if (!memcmp (m_Buffer, SAM_HANDSHAKE, sizeof (SAM_HANDSHAKE)))
+			{
+				// TODO: check version
+				boost::asio::async_write (m_Socket, boost::asio::buffer (SAM_HANDSHAKE_REPLY, sizeof (SAM_HANDSHAKE_REPLY)), boost::asio::transfer_all (),
+        			boost::bind(&SAMSocket::HandleHandshakeReplySent, this, 
+					boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
+			}
+			else
+			{
+				LogPrint ("SAM hannshake mismatch");
+				Terminate ();
+			}
+		}
+	}
+
+	void SAMSocket::HandleHandshakeReplySent (const boost::system::error_code& ecode, std::size_t bytes_transferred)
+	{
+		if (ecode)
+        {
+			LogPrint ("SAM handshake reply send error: ", ecode.message ());
+			if (ecode != boost::asio::error::operation_aborted)
+				Terminate ();
+		}
+		else
+			Receive ();
+	}
+
+	void SAMSocket::Receive ()
+	{
+		m_Socket.async_read_some (boost::asio::buffer(m_Buffer, SAM_SOCKET_BUFFER_SIZE),                
+			boost::bind(&SAMSocket::HandleReceived, this, 
+			boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred));
+	}
+
+	void SAMSocket::HandleReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred)
+	{
+		if (ecode)
+        {
+			LogPrint ("SAM read error: ", ecode.message ());
+			if (ecode != boost::asio::error::operation_aborted)
+				Terminate ();
+		}
+		else
+		{
+			if (m_Stream)
+				m_Stream->Send ((uint8_t *)m_Buffer, bytes_transferred, 0);
+			Receive ();
+		}
+	}
+
+	void SAMSocket::StreamReceive ()
+	{
+		if (m_Stream)
+			m_Stream->AsyncReceive (boost::asio::buffer (m_StreamBuffer, SAM_SOCKET_BUFFER_SIZE),
+				boost::bind (&SAMSocket::HandleStreamReceive, this,
+					boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred),
+				SAM_SOCKET_CONNECTION_MAX_IDLE);
+	}	
+
+	void SAMSocket::HandleStreamReceive (const boost::system::error_code& ecode, std::size_t bytes_transferred)
+	{
+		if (ecode)
+		{
+			LogPrint ("SAM stream read error: ", ecode.message ());
+			Terminate ();
+		}
+		else
+		{
+			boost::asio::async_write (m_Socket, boost::asio::buffer (m_StreamBuffer, bytes_transferred),
+        		boost::bind (&SAMSocket::HandleWriteStreamData, this, boost::asio::placeholders::error));
+		}
+	}
+
+	void SAMSocket::HandleWriteStreamData (const boost::system::error_code& ecode)
+	{
+		if (ecode)
+		{
+			LogPrint ("SAM socket write error: ", ecode.message ());
+			if (ecode != boost::asio::error::operation_aborted)
+				Terminate ();
+		}
+		else
+			StreamReceive ();
+	}
+
 	SAMBridge::SAMBridge (int port):
 		m_IsRunning (false), m_Thread (nullptr),
 		m_Acceptor (m_Service, boost::asio::ip::tcp::endpoint(boost::asio::ip::tcp::v4(), port)),
@@ -54,8 +180,8 @@ namespace stream
 
 	void SAMBridge::Accept ()
 	{
-		m_NewSocket = new boost::asio::ip::tcp::socket (m_Service);
+		m_NewSocket = new SAMSocket (*this);
-		m_Acceptor.async_accept (*m_NewSocket, boost::bind (&SAMBridge::HandleAccept, this,
+		m_Acceptor.async_accept (m_NewSocket->GetSocket (), boost::bind (&SAMBridge::HandleAccept, this,
 			boost::asio::placeholders::error));
 	}
 
@@ -63,7 +189,8 @@ namespace stream
 	{
 		if (!ecode)
 		{
-			//TODO:
+			LogPrint ("New SAM connection from ", m_NewSocket->GetSocket ().remote_endpoint ());
+			m_NewSocket->ReceiveHandshake ();		
 		}
 		else
 		{

SAM.h

@@ -1,13 +1,53 @@
 #ifndef SAM_H__
 #define SAM_H__
 
+#include <inttypes.h>
+#include <string>
 #include <thread>
 #include <boost/asio.hpp>
+#include "Streaming.h"
 
 namespace i2p
 {
 namespace stream
 {
+	const size_t SAM_SOCKET_BUFFER_SIZE = 4096;
+	const int SAM_SOCKET_CONNECTION_MAX_IDLE = 3600; // in seconds	
+	const char SAM_HANDSHAKE[] = "HELLO VERSION";
+	const char SAM_HANDSHAKE_REPLY[] = "HELLO REPLY RESULT=OK VERSION=3.1";
+
+	class SAMBridge;
+	class SAMSocket
+	{
+		public:
+
+			SAMSocket (SAMBridge& owner);
+			~SAMSocket ();			
+
+			boost::asio::ip::tcp::socket& GetSocket () { return m_Socket; };
+			void ReceiveHandshake ();
+
+		private:
+
+			void Terminate ();
+			void HandleHandshakeReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred);
+			void HandleHandshakeReplySent (const boost::system::error_code& ecode, std::size_t bytes_transferred);
+			void Receive ();
+			void HandleReceived (const boost::system::error_code& ecode, std::size_t bytes_transferred);
+
+			void StreamReceive ();	
+			void HandleStreamReceive (const boost::system::error_code& ecode, std::size_t bytes_transferred);
+			void HandleWriteStreamData (const boost::system::error_code& ecode);
+
+		private:
+
+			SAMBridge& m_Owner;
+			boost::asio::ip::tcp::socket m_Socket;
+			char m_Buffer[SAM_SOCKET_BUFFER_SIZE + 1];
+			uint8_t m_StreamBuffer[SAM_SOCKET_BUFFER_SIZE];
+			Stream * m_Stream;
+	};	
+
 	class SAMBridge
 	{
 		public:
@@ -18,6 +58,8 @@ namespace stream
 			void Start ();
 			void Stop ();
 			
+			boost::asio::io_service& GetService () { return m_Service; };
+
 		private:
 
 			void Run ();
@@ -31,7 +73,7 @@ namespace stream
 			std::thread * m_Thread;	
 			boost::asio::io_service m_Service;
 			boost::asio::ip::tcp::acceptor m_Acceptor;
-			boost::asio::ip::tcp::socket * m_NewSocket;
+			SAMSocket * m_NewSocket;
 	};		
 }
 }