implemented AsyncSend

Streaming.cpp

@@ -61,6 +61,11 @@ namespace stream
 		m_AckSendTimer.cancel ();
 		m_ReceiveTimer.cancel ();
 		m_ResendTimer.cancel ();
+		if (m_SendHandler) 
+		{
+			m_SendHandler (boost::asio::error::make_error_code (boost::asio::error::operation_aborted));
+			m_SendHandler = nullptr;
+		}
 	}	
 		
 	void Stream::HandleNextPacket (Packet * packet)
@@ -299,6 +304,15 @@ namespace stream
 		return len;
 	}	
 
+	void Stream::AsyncSend (const uint8_t * buf, size_t len, SendHandler handler)
+	{
+		if (m_SendHandler) 
+			handler (boost::asio::error::make_error_code (boost::asio::error::in_progress));
+		else
+			m_SendHandler = handler;
+		Send (buf, len);
+	}
+
 	void Stream::SendBuffer ()
 	{	
 		int numMsgs = m_WindowSize - m_SentPackets.size ();
@@ -367,6 +381,11 @@ namespace stream
 				packets.push_back (p);
 				numMsgs--;
 			}
+			if (m_SendBuffer.eof () && m_SendHandler)
+			{
+				m_SendHandler (boost::system::error_code ());
+				m_SendHandler = nullptr;
+			}
 		}	
 		if (packets.size () > 0)
 		{

Streaming.h

@@ -97,6 +97,8 @@ namespace stream
 	{	
 		public:
 
+			typedef std::function<void (const boost::system::error_code& ecode)> SendHandler;
+
 			Stream (boost::asio::io_service& service, StreamingDestination& local, 
 				std::shared_ptr<const i2p::data::LeaseSet> remote, int port = 0); // outgoing
 			Stream (boost::asio::io_service& service, StreamingDestination& local); // incoming			
@@ -112,6 +114,7 @@ namespace stream
 			
 			void HandleNextPacket (Packet * packet);
 			size_t Send (const uint8_t * buf, size_t len);
+			void AsyncSend (const uint8_t * buf, size_t len, SendHandler handler);
 			
 			template<typename Buffer, typename ReceiveHandler>
 			void AsyncReceive (const Buffer& buffer, ReceiveHandler handler, int timeout = 0);
@@ -179,6 +182,7 @@ namespace stream
 			int m_WindowSize, m_RTT, m_RTO;
 			uint64_t m_LastWindowSizeIncreaseTime;
 			int m_NumResendAttempts;
+			SendHandler m_SendHandler;
 	};
 
 	class StreamingDestination