Browse Source

create io_service together with destination

pull/118/head
orignal 10 years ago
parent
commit
e3f077ee9a
  1. 6
      Datagram.cpp
  2. 60
      Destination.cpp
  3. 8
      Destination.h
  4. 4
      Streaming.cpp

6
Datagram.cpp

@ -36,12 +36,8 @@ namespace datagram
else else
m_Owner.Sign (buf1, len, signature); m_Owner.Sign (buf1, len, signature);
auto service = m_Owner.GetService (); m_Owner.GetService ().post (std::bind (&DatagramDestination::SendMsg, this,
if (service)
service->post (std::bind (&DatagramDestination::SendMsg, this,
CreateDataMessage (buf, len + headerLen), remote)); CreateDataMessage (buf, len + headerLen), remote));
else
LogPrint (eLogWarning, "Failed to send datagram. Destination is not running");
} }
void DatagramDestination::SendMsg (I2NPMessage * msg, const i2p::data::LeaseSet& remote) void DatagramDestination::SendMsg (I2NPMessage * msg, const i2p::data::LeaseSet& remote)

60
Destination.cpp

@ -12,9 +12,9 @@ namespace client
{ {
ClientDestination::ClientDestination (const i2p::data::PrivateKeys& keys, bool isPublic, ClientDestination::ClientDestination (const i2p::data::PrivateKeys& keys, bool isPublic,
const std::map<std::string, std::string> * params): const std::map<std::string, std::string> * params):
m_IsRunning (false), m_Thread (nullptr), m_Service (nullptr), m_Work (nullptr), m_IsRunning (false), m_Thread (nullptr), m_Work (m_Service),
m_Keys (keys), m_LeaseSet (nullptr), m_IsPublic (isPublic), m_PublishReplyToken (0), m_Keys (keys), m_LeaseSet (nullptr), m_IsPublic (isPublic), m_PublishReplyToken (0),
m_DatagramDestination (nullptr), m_PublishConfirmationTimer (nullptr) m_DatagramDestination (nullptr), m_PublishConfirmationTimer (m_Service)
{ {
CryptoPP::DH dh (i2p::crypto::elgp, i2p::crypto::elgg); CryptoPP::DH dh (i2p::crypto::elgp, i2p::crypto::elgg);
dh.GenerateKeyPair(i2p::context.GetRandomNumberGenerator (), m_EncryptionPrivateKey, m_EncryptionPublicKey); dh.GenerateKeyPair(i2p::context.GetRandomNumberGenerator (), m_EncryptionPrivateKey, m_EncryptionPublicKey);
@ -56,12 +56,6 @@ namespace client
delete it.second; delete it.second;
if (m_Pool) if (m_Pool)
i2p::tunnel::tunnels.DeleteTunnelPool (m_Pool); i2p::tunnel::tunnels.DeleteTunnelPool (m_Pool);
delete m_LeaseSet;
delete m_Work;
delete m_PublishConfirmationTimer;
delete m_Service;
delete m_StreamingDestination;
delete m_DatagramDestination;
} }
void ClientDestination::Run () void ClientDestination::Run ()
@ -70,8 +64,7 @@ namespace client
{ {
try try
{ {
if (m_Service) m_Service.run ();
m_Service->run ();
} }
catch (std::exception& ex) catch (std::exception& ex)
{ {
@ -82,9 +75,6 @@ namespace client
void ClientDestination::Start () void ClientDestination::Start ()
{ {
m_Service = new boost::asio::io_service;
m_PublishConfirmationTimer = new boost::asio::deadline_timer (*m_Service);
m_Work = new boost::asio::io_service::work (*m_Service);
m_Pool->SetLocalDestination (this); m_Pool->SetLocalDestination (this);
m_Pool->SetActive (true); m_Pool->SetActive (true);
m_IsRunning = true; m_IsRunning = true;
@ -107,17 +97,13 @@ namespace client
i2p::tunnel::tunnels.StopTunnelPool (m_Pool); i2p::tunnel::tunnels.StopTunnelPool (m_Pool);
} }
m_IsRunning = false; m_IsRunning = false;
if (m_Service) m_Service.stop ();
m_Service->stop ();
if (m_Thread) if (m_Thread)
{ {
m_Thread->join (); m_Thread->join ();
delete m_Thread; delete m_Thread;
m_Thread = 0; m_Thread = 0;
} }
delete m_PublishConfirmationTimer; m_PublishConfirmationTimer = nullptr;
delete m_Work; m_Work = nullptr;
delete m_Service; m_Service = nullptr;
} }
const i2p::data::LeaseSet * ClientDestination::FindLeaseSet (const i2p::data::IdentHash& ident) const i2p::data::LeaseSet * ClientDestination::FindLeaseSet (const i2p::data::IdentHash& ident)
@ -168,8 +154,6 @@ namespace client
} }
bool ClientDestination::SubmitSessionKey (const uint8_t * key, const uint8_t * tag) bool ClientDestination::SubmitSessionKey (const uint8_t * key, const uint8_t * tag)
{
if (m_Service)
{ {
struct struct
{ {
@ -177,39 +161,21 @@ namespace client
} data; } data;
memcpy (data.k, key, 32); memcpy (data.k, key, 32);
memcpy (data.t, tag, 32); memcpy (data.t, tag, 32);
m_Service->post ([this,data](void) m_Service.post ([this,data](void)
{ {
this->AddSessionKey (data.k, data.t); this->AddSessionKey (data.k, data.t);
}); });
return true; return true;
} }
else
{
LogPrint (eLogWarning, "Destination's thread is not running");
return false;
}
}
void ClientDestination::ProcessGarlicMessage (I2NPMessage * msg) void ClientDestination::ProcessGarlicMessage (I2NPMessage * msg)
{ {
if (m_Service) m_Service.post (std::bind (&ClientDestination::HandleGarlicMessage, this, msg));
m_Service->post (std::bind (&ClientDestination::HandleGarlicMessage, this, msg));
else
{
LogPrint (eLogWarning, "Destination's thread is not running");
i2p::DeleteI2NPMessage (msg);
}
} }
void ClientDestination::ProcessDeliveryStatusMessage (I2NPMessage * msg) void ClientDestination::ProcessDeliveryStatusMessage (I2NPMessage * msg)
{ {
if (m_Service) m_Service.post (std::bind (&ClientDestination::HandleDeliveryStatusMessage, this, msg));
m_Service->post (std::bind (&ClientDestination::HandleDeliveryStatusMessage, this, msg));
else
{
LogPrint (eLogWarning, "Destination's thread is not running");
i2p::DeleteI2NPMessage (msg);
}
} }
void ClientDestination::HandleI2NPMessage (const uint8_t * buf, size_t len, i2p::tunnel::InboundTunnel * from) void ClientDestination::HandleI2NPMessage (const uint8_t * buf, size_t len, i2p::tunnel::InboundTunnel * from)
@ -307,17 +273,9 @@ namespace client
LogPrint (eLogDebug, "Publish LeaseSet of ", GetIdentHash ().ToBase32 ()); LogPrint (eLogDebug, "Publish LeaseSet of ", GetIdentHash ().ToBase32 ());
m_PublishReplyToken = i2p::context.GetRandomNumberGenerator ().GenerateWord32 (); m_PublishReplyToken = i2p::context.GetRandomNumberGenerator ().GenerateWord32 ();
auto msg = WrapMessage (*floodfill, i2p::CreateDatabaseStoreMsg (m_LeaseSet, m_PublishReplyToken)); auto msg = WrapMessage (*floodfill, i2p::CreateDatabaseStoreMsg (m_LeaseSet, m_PublishReplyToken));
if (m_PublishConfirmationTimer) m_PublishConfirmationTimer.expires_from_now (boost::posix_time::seconds(PUBLISH_CONFIRMATION_TIMEOUT));
{ m_PublishConfirmationTimer.async_wait (std::bind (&ClientDestination::HandlePublishConfirmationTimer,
m_PublishConfirmationTimer->expires_from_now (boost::posix_time::seconds(PUBLISH_CONFIRMATION_TIMEOUT));
m_PublishConfirmationTimer->async_wait (std::bind (&ClientDestination::HandlePublishConfirmationTimer,
this, std::placeholders::_1)); this, std::placeholders::_1));
}
else
{
LogPrint (eLogWarning, "Destination's thread is not running");
m_PublishReplyToken = 0;
}
outbound->SendTunnelDataMsg (floodfill->GetIdentHash (), 0, msg); outbound->SendTunnelDataMsg (floodfill->GetIdentHash (), 0, msg);
} }

8
Destination.h

@ -39,7 +39,7 @@ namespace client
virtual void Start (); virtual void Start ();
virtual void Stop (); virtual void Stop ();
bool IsRunning () const { return m_IsRunning; }; bool IsRunning () const { return m_IsRunning; };
boost::asio::io_service * GetService () { return m_Service; }; boost::asio::io_service& GetService () { return m_Service; };
i2p::tunnel::TunnelPool * GetTunnelPool () { return m_Pool; }; i2p::tunnel::TunnelPool * GetTunnelPool () { return m_Pool; };
bool IsReady () const { return m_LeaseSet && m_LeaseSet->HasNonExpiredLeases (); }; bool IsReady () const { return m_LeaseSet && m_LeaseSet->HasNonExpiredLeases (); };
const i2p::data::LeaseSet * FindLeaseSet (const i2p::data::IdentHash& ident); const i2p::data::LeaseSet * FindLeaseSet (const i2p::data::IdentHash& ident);
@ -86,8 +86,8 @@ namespace client
bool m_IsRunning; bool m_IsRunning;
std::thread * m_Thread; std::thread * m_Thread;
boost::asio::io_service * m_Service; boost::asio::io_service m_Service;
boost::asio::io_service::work * m_Work; boost::asio::io_service::work m_Work;
i2p::data::PrivateKeys m_Keys; i2p::data::PrivateKeys m_Keys;
uint8_t m_EncryptionPublicKey[256], m_EncryptionPrivateKey[256]; uint8_t m_EncryptionPublicKey[256], m_EncryptionPrivateKey[256];
std::map<i2p::data::IdentHash, i2p::data::LeaseSet *> m_RemoteLeaseSets; std::map<i2p::data::IdentHash, i2p::data::LeaseSet *> m_RemoteLeaseSets;
@ -101,7 +101,7 @@ namespace client
i2p::stream::StreamingDestination * m_StreamingDestination; i2p::stream::StreamingDestination * m_StreamingDestination;
i2p::datagram::DatagramDestination * m_DatagramDestination; i2p::datagram::DatagramDestination * m_DatagramDestination;
boost::asio::deadline_timer * m_PublishConfirmationTimer; boost::asio::deadline_timer m_PublishConfirmationTimer;
public: public:

4
Streaming.cpp

@ -672,7 +672,7 @@ namespace stream
std::shared_ptr<Stream> StreamingDestination::CreateNewOutgoingStream (const i2p::data::LeaseSet& remote, int port) std::shared_ptr<Stream> StreamingDestination::CreateNewOutgoingStream (const i2p::data::LeaseSet& remote, int port)
{ {
auto s = std::make_shared<Stream> (*m_Owner.GetService (), *this, remote, port); auto s = std::make_shared<Stream> (m_Owner.GetService (), *this, remote, port);
std::unique_lock<std::mutex> l(m_StreamsMutex); std::unique_lock<std::mutex> l(m_StreamsMutex);
m_Streams[s->GetRecvStreamID ()] = s; m_Streams[s->GetRecvStreamID ()] = s;
return s; return s;
@ -680,7 +680,7 @@ namespace stream
std::shared_ptr<Stream> StreamingDestination::CreateNewIncomingStream () std::shared_ptr<Stream> StreamingDestination::CreateNewIncomingStream ()
{ {
auto s = std::make_shared<Stream> (*m_Owner.GetService (), *this); auto s = std::make_shared<Stream> (m_Owner.GetService (), *this);
std::unique_lock<std::mutex> l(m_StreamsMutex); std::unique_lock<std::mutex> l(m_StreamsMutex);
m_Streams[s->GetRecvStreamID ()] = s; m_Streams[s->GetRecvStreamID ()] = s;
return s; return s;

Loading…
Cancel
Save