diff --git a/SAM.cpp b/SAM.cpp index 156df4cf..1e4238c9 100644 --- a/SAM.cpp +++ b/SAM.cpp @@ -1,3 +1,4 @@ +#include #include #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_Acceptor.async_accept (*m_NewSocket, boost::bind (&SAMBridge::HandleAccept, this, + m_NewSocket = new SAMSocket (*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 { diff --git a/SAM.h b/SAM.h index e547adfe..9a22a0b7 100644 --- a/SAM.h +++ b/SAM.h @@ -1,13 +1,53 @@ #ifndef SAM_H__ #define SAM_H__ +#include +#include #include #include +#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: @@ -17,6 +57,8 @@ namespace stream void Start (); void Stop (); + + boost::asio::io_service& GetService () { return m_Service; }; private: @@ -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; }; } }