DATAGRAM RECEIVED message

Datagram.cpp

@@ -13,7 +13,7 @@ namespace i2p
 namespace datagram
 {
 	DatagramDestination::DatagramDestination (i2p::client::ClientDestination& owner): 
-		m_Owner (owner) 
+		m_Owner (owner), m_Receiver (nullptr)
 	{
 	}
 
@@ -41,7 +41,7 @@ namespace datagram
 			service->post (boost::bind (&DatagramDestination::SendMsg, this, 
 				CreateDataMessage (buf, len + headerLen), remote));
 		else
-			LogPrint ("Failed to send datagram. Destination is not running");
+			LogPrint (eLogWarning, "Failed to send datagram. Destination is not running");
 	}
 
 	void DatagramDestination::SendMsg (I2NPMessage * msg, const i2p::data::LeaseSet& remote)
@@ -62,7 +62,7 @@ namespace datagram
 		}
 		else
 		{
-			LogPrint ("Failed to send datagram. All leases expired");
+			LogPrint (eLogWarning, "Failed to send datagram. All leases expired");
 			DeleteI2NPMessage (msg);	
 		}	
 	}
@@ -86,10 +86,13 @@ namespace datagram
 				
 		if (verified)
 		{
-			// TODO: invoke datagram handler
+			if (m_Receiver != nullptr)
+				m_Receiver (identity, buf + headerLen, len -headerLen);
+			else
+				LogPrint (eLogWarning, "Receiver for datagram is not set");	
 		}
 		else
-			LogPrint ("Datagram signature verification failed");	
+			LogPrint (eLogWarning, "Datagram signature verification failed");	
 	}
 
 	void DatagramDestination::HandleDataMessagePayload (const uint8_t * buf, size_t len)

Datagram.h

@@ -2,6 +2,8 @@
 #define DATAGRAM_H__
 
 #include <inttypes.h>
+#include <functional>
+#include "Identity.h"
 #include "LeaseSet.h"
 #include "I2NPProtocol.h"
 
@@ -16,6 +18,8 @@ namespace datagram
 	const size_t MAX_DATAGRAM_SIZE = 32768;
 	class DatagramDestination
 	{
+		typedef std::function<void (const i2p::data::IdentityEx& ident, const uint8_t *, size_t)> Receiver;
+
 		public:
 
 			DatagramDestination (i2p::client::ClientDestination& owner);
@@ -24,6 +28,9 @@ namespace datagram
 			void SendDatagramTo (const uint8_t * payload, size_t len, const i2p::data::LeaseSet& remote);
 			void HandleDataMessagePayload (const uint8_t * buf, size_t len);
 
+			void SetReceiver (const Receiver& receiver) { m_Receiver = receiver; };
+			void ResetReceiver () { m_Receiver = nullptr; };
+
 		private:
 
 			I2NPMessage * CreateDataMessage (const uint8_t * payload, size_t len);
@@ -33,6 +40,7 @@ namespace datagram
 		private:
 
 			i2p::client::ClientDestination& m_Owner;
+			Receiver m_Receiver;
 	};		
 }
 }

Destination.cpp

@@ -107,6 +107,12 @@ namespace client
 	void ClientDestination::Stop ()
 	{	
 		m_StreamingDestination->Stop ();	
+		if (m_DatagramDestination)
+		{
+			auto d = m_DatagramDestination;
+			m_DatagramDestination = nullptr;
+			delete d;
+		}	
 		if (m_Pool)
 			i2p::tunnel::tunnels.StopTunnelPool (m_Pool);
 		m_IsRunning = false;
@@ -294,10 +300,11 @@ namespace client
 		return false;
 	}	
 
-	void ClientDestination::CreateDatagramDestination ()
+	i2p::datagram::DatagramDestination * ClientDestination::CreateDatagramDestination ()
 	{
 		if (!m_DatagramDestination)
 			m_DatagramDestination = new i2p::datagram::DatagramDestination (*this);
+		return m_DatagramDestination;	
 	}
 }
 }

Destination.h

@@ -48,7 +48,7 @@ namespace client
 
 			// datagram
 			i2p::datagram::DatagramDestination * GetDatagramDestination () const { return m_DatagramDestination; };
-			void CreateDatagramDestination ();
+			i2p::datagram::DatagramDestination * CreateDatagramDestination ();
 
 			// implements LocalDestination
 			const i2p::data::PrivateKeys& GetPrivateKeys () const { return m_Keys; };

SAM.cpp

@@ -226,7 +226,11 @@ namespace client
 			if (m_Session->localDestination->IsReady ())
 			{
 				if (style == SAM_VALUE_DATAGRAM)
-					m_Session->localDestination->CreateDatagramDestination ();
+				{
+					auto dest = m_Session->localDestination->CreateDatagramDestination ();
+					dest->SetReceiver (std::bind (&SAMSocket::HandleI2PDatagramReceive, this, 
+						std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
+				}
 				SendSessionCreateReplyOk ();
 			}
 			else
@@ -548,6 +552,27 @@ namespace client
 		}
 	}	
 
+	void SAMSocket::HandleI2PDatagramReceive (const i2p::data::IdentityEx& ident, const uint8_t * buf, size_t len)
+	{
+		uint8_t identBuf[1024];
+		size_t l = ident.ToBuffer (identBuf, 1024);
+		size_t l1 = i2p::data::ByteStreamToBase64 (identBuf, l, m_Buffer, SAM_SOCKET_BUFFER_SIZE);
+		m_Buffer[l1] = 0;
+#ifdef _MSC_VER
+		size_t l2 = sprintf_s ((char *)m_StreamBuffer, SAM_SOCKET_BUFFER_SIZE, SAM_DATAGRAM_RECEIVED, m_Buffer, len); 	
+#else			
+		size_t l2 = snprintf ((char *)m_StreamBuffer, SAM_SOCKET_BUFFER_SIZE, SAM_DATAGRAM_RECEIVED, m_Buffer, len); 	
+#endif
+		if (len < SAM_SOCKET_BUFFER_SIZE - l2)	
+		{	
+			memcpy (m_StreamBuffer + l2, buf, len);
+			boost::asio::async_write (m_Socket, boost::asio::buffer (m_StreamBuffer, len + l2),
+        		boost::bind (&SAMSocket::HandleWriteI2PData, this, boost::asio::placeholders::error));
+		}
+		else
+			LogPrint (eLogWarning, "Datagram size ", len," exceeds buffer");
+	}
+
 	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)),

SAM.h

@@ -39,6 +39,7 @@ namespace client
 	const char SAM_DEST_REPLY_I2P_ERROR[] = "DEST REPLY RESULT=I2P_ERROR\n";
 	const char SAM_NAMING_LOOKUP[] = "NAMING LOOKUP";
 	const char SAM_NAMING_REPLY[] = "NAMING REPLY RESULT=OK NAME=ME VALUE=%s\n";
+	const char SAM_DATAGRAM_RECEIVED[] = "DATAGRAM_RECEIVED DESTINATION=%s SIZE=%i\n";	
 	const char SAM_NAMING_REPLY_INVALID_KEY[] = "NAMING REPLY RESULT=INVALID_KEY NAME=%s\n";
 	const char SAM_NAMING_REPLY_KEY_NOT_FOUND[] = "NAMING REPLY RESULT=INVALID_KEY_NOT_FOUND NAME=%s\n";		
 	const char SAM_PARAM_STYLE[] = "STYLE";		
@@ -88,6 +89,7 @@ namespace client
 			void HandleI2PReceive (const boost::system::error_code& ecode, std::size_t bytes_transferred);
 			void HandleI2PAccept (i2p::stream::Stream * stream);
 			void HandleWriteI2PData (const boost::system::error_code& ecode);
+			void HandleI2PDatagramReceive (const i2p::data::IdentityEx& ident, const uint8_t * buf, size_t len);
 
 			void ProcessSessionCreate (char * buf, size_t len);
 			void ProcessStreamConnect (char * buf, size_t len);