PurpleI2P/i2pd
1
0
Fork 0
mirror of https://github.com/PurpleI2P/i2pd.git synced 2025-01-22 04:04:16 +00:00
Code Issues Releases Activity

AsyncReceive added

Browse Source
This commit is contained in:
orignal 2014-03-25 14:26:39 -04:00
parent 88214a0c58
commit 1bbc12b36a
2 changed files with 34 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
Streaming.cpp
View File

@ -1,6 +1,5 @@
#include <fstream> #include <fstream>
#include <algorithm> #include <algorithm>
#include <boost/bind.hpp>
#include <cryptopp/gzip.h> #include <cryptopp/gzip.h>
#include "Log.h" #include "Log.h"
#include "RouterInfo.h" #include "RouterInfo.h"
@ -19,7 +18,7 @@ namespace stream
const i2p::data::LeaseSet& remote): m_Service (service), m_SendStreamID (0), const i2p::data::LeaseSet& remote): m_Service (service), m_SendStreamID (0),
m_SequenceNumber (0), m_LastReceivedSequenceNumber (0), m_IsOpen (false), m_SequenceNumber (0), m_LastReceivedSequenceNumber (0), m_IsOpen (false),
m_LeaseSetUpdated (true), m_LocalDestination (local), m_RemoteLeaseSet (remote), m_LeaseSetUpdated (true), m_LocalDestination (local), m_RemoteLeaseSet (remote),
m_OutboundTunnel (nullptr) m_OutboundTunnel (nullptr), m_ReceiveTimer (m_Service)
{ {
m_RecvStreamID = i2p::context.GetRandomNumberGenerator ().GenerateWord32 (); m_RecvStreamID = i2p::context.GetRandomNumberGenerator ().GenerateWord32 ();
UpdateCurrentRemoteLease (); UpdateCurrentRemoteLease ();
@ -27,6 +26,7 @@ namespace stream
Stream::~Stream () Stream::~Stream ()
{ {
m_ReceiveTimer.cancel ();
while (auto packet = m_ReceiveQueue.Get ()) while (auto packet = m_ReceiveQueue.Get ())
delete packet; delete packet;
for (auto it: m_SavedPackets) for (auto it: m_SavedPackets)
@ -124,6 +124,7 @@ namespace stream
LogPrint ("Closed"); LogPrint ("Closed");
m_IsOpen = false; m_IsOpen = false;
m_ReceiveQueue.WakeUp (); m_ReceiveQueue.WakeUp ();
m_ReceiveTimer.cancel ();
} }
} }

32
Streaming.h
View File

@ -7,6 +7,7 @@
#include <set> #include <set>
#include <thread> #include <thread>
#include <boost/asio.hpp> #include <boost/asio.hpp>
#include <boost/bind.hpp>
#include <cryptopp/dsa.h> #include <cryptopp/dsa.h>
#include "I2PEndian.h" #include "I2PEndian.h"
#include "Queue.h" #include "Queue.h"
@ -80,10 +81,13 @@ namespace stream
void HandleNextPacket (Packet * packet); void HandleNextPacket (Packet * packet);
size_t Send (uint8_t * buf, size_t len, int timeout); // timeout in seconds size_t Send (uint8_t * buf, size_t len, int timeout); // timeout in seconds
size_t Receive (uint8_t * buf, size_t len, int timeout = 0); // returns 0 if timeout expired size_t Receive (uint8_t * buf, size_t len, int timeout = 0); // returns 0 if timeout expired
template<typename Buffer, typename ReceiveHandler>
void AsyncReceive (const Buffer& buffer, ReceiveHandler handler, int timeout = 0);
void Close (); void Close ();
void SetLeaseSetUpdated () { m_LeaseSetUpdated = true; }; void SetLeaseSetUpdated () { m_LeaseSetUpdated = true; };
private: private:
void ConnectAndSend (uint8_t * buf, size_t len); void ConnectAndSend (uint8_t * buf, size_t len);
@ -96,6 +100,9 @@ namespace stream
void UpdateCurrentRemoteLease (); void UpdateCurrentRemoteLease ();
template<typename Buffer, typename ReceiveHandler>
void HandleReceiveTimer (const boost::system::error_code& ecode, const Buffer& buffer, ReceiveHandler handler);
private: private:
boost::asio::io_service& m_Service; boost::asio::io_service& m_Service;
@ -107,6 +114,7 @@ namespace stream
i2p::util::Queue<Packet> m_ReceiveQueue; i2p::util::Queue<Packet> m_ReceiveQueue;
std::set<Packet *, PacketCmp> m_SavedPackets; std::set<Packet *, PacketCmp> m_SavedPackets;
i2p::tunnel::OutboundTunnel * m_OutboundTunnel; i2p::tunnel::OutboundTunnel * m_OutboundTunnel;
boost::asio::deadline_timer m_ReceiveTimer;
}; };
class StreamingDestination: public i2p::data::LocalDestination class StreamingDestination: public i2p::data::LocalDestination
@ -185,6 +193,28 @@ namespace stream
// assuming data is I2CP message // assuming data is I2CP message
void HandleDataMessage (i2p::data::IdentHash destination, const uint8_t * buf, size_t len); void HandleDataMessage (i2p::data::IdentHash destination, const uint8_t * buf, size_t len);
I2NPMessage * CreateDataMessage (Stream * s, const uint8_t * payload, size_t len); I2NPMessage * CreateDataMessage (Stream * s, const uint8_t * payload, size_t len);
//-------------------------------------------------
template<typename Buffer, typename ReceiveHandler>
void Stream::AsyncReceive (const Buffer& buffer, ReceiveHandler handler, int timeout)
{
m_ReceiveTimer.expires_from_now (boost::posix_time::seconds(timeout));
m_ReceiveTimer.async_wait (boost::bind (&Stream::HandleReceiveTimer,
this, boost::asio::placeholders::error, buffer, handler));
}
template<typename Buffer, typename ReceiveHandler>
void Stream::HandleReceiveTimer (const boost::system::error_code& ecode, const Buffer& buffer, ReceiveHandler handler)
{
// TODO:
if (ecode == boost::asio::error::operation_aborted)
// timeout not expired
handler (boost::system::error_code (), 0);
else
// timeout expired
handler (boost::asio::error::make_error_code (boost::asio::error::timed_out), 0);
}
} }
} }