PurpleI2P
/
i2pd
mirror of https://github.com/PurpleI2P/i2pd.git
1
0
Fork 0
Code Issues Releases Activity
Browse Source

Asyc receive from stream

pull/48/head
orignal 11 years ago
parent
5eb67a08c9
commit
86e233d77a
2 changed files with 48 additions and 22 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 60
      HTTPServer.cpp
  2. 10
      HTTPServer.h

60
HTTPServer.cpp
Unescape View File

@ -7,7 +7,6 @@
#include "TransitTunnel.h" #include "TransitTunnel.h"
#include "Transports.h" #include "Transports.h"
#include "NetDb.h" #include "NetDb.h"
#include "Streaming.h"
#include "HTTPServer.h" #include "HTTPServer.h"
namespace i2p namespace i2p
@ -44,6 +43,11 @@ namespace util
void HTTPConnection::Terminate () void HTTPConnection::Terminate ()
{ {
if (m_Stream)
{
m_Stream->Close ();
DeleteStream (m_Stream);
}
m_Socket->close (); m_Socket->close ();
delete this; delete this;
} }
@ -94,10 +98,17 @@ namespace util
return ""; return "";
} }
void HTTPConnection::HandleWrite (const boost::system::error_code& ecode, bool terminate) void HTTPConnection::HandleWriteReply (const boost::system::error_code& ecode)
{
Terminate ();
}
void HTTPConnection::HandleWrite (const boost::system::error_code& ecode)
{ {
if (terminate) if (ecode || (m_Stream && !m_Stream->IsOpen ()))
Terminate (); Terminate ();
else // data keeps coming
AsyncStreamReceive ();
} }
void HTTPConnection::HandleRequest () void HTTPConnection::HandleRequest ()
@ -231,47 +242,58 @@ namespace util
SendReply (leaseSet ? "<html>Leases expired</html>" : "<html>LeaseSet not found</html>"); SendReply (leaseSet ? "<html>Leases expired</html>" : "<html>LeaseSet not found</html>");
return; return;
} }
} }
auto s = i2p::stream::CreateStream (*leaseSet); if (!m_Stream)
if (s) m_Stream = i2p::stream::CreateStream (*leaseSet);
if (m_Stream)
{ {
std::string request = "GET " + uri + " HTTP/1.1\n Host:" + fullAddress + "\n"; std::string request = "GET " + uri + " HTTP/1.1\n Host:" + fullAddress + "\n";
s->Send ((uint8_t *)request.c_str (), request.length (), 10); m_Stream->Send ((uint8_t *)request.c_str (), request.length (), 10);
std::stringstream ss; std::stringstream ss;
uint8_t buf[8192]; uint8_t buf[8192];
size_t r = s->Receive (buf, 8192, 30); // 30 seconds size_t r = m_Stream->Receive (buf, 8192, 30); // 30 seconds
if (!r && s->IsEstablished ()) // nothing received but connection is established if (!r && m_Stream->IsEstablished ()) // nothing received but connection is established
r = s->Receive (buf, 8192, 30); // wait for another 30 secondd r = m_Stream->Receive (buf, 8192, 30); // wait for another 30 secondd
if (r) // we recieved data if (r) // we recieved data
{ {
ss << std::string ((char *)buf, r); ss << std::string ((char *)buf, r);
while (s->IsOpen () && (r = s->Receive (buf, 8192, 30)) > 0) while (m_Stream->IsOpen () && (r = m_Stream->Receive (buf, 8192, 30)) > 0)
ss << std::string ((char *)buf,r); ss << std::string ((char *)buf,r);
m_Reply.content = ss.str (); // send "as is" m_Reply.content = ss.str (); // send "as is"
m_Reply.headers.resize(0); // no headers m_Reply.headers.resize(0); // no headers
boost::asio::async_write (*m_Socket, m_Reply.to_buffers(), boost::asio::async_write (*m_Socket, m_Reply.to_buffers(),
boost::bind (&HTTPConnection::HandleWrite, this, boost::bind (&HTTPConnection::HandleWriteReply, this, boost::asio::placeholders::error));
boost::asio::placeholders::error, true));
return; return;
} }
else // nothing received else // nothing received
ss << "<html>Not responding</html>"; ss << "<html>Not responding</html>";
s->Close ();
DeleteStream (s);
SendReply (ss.str ()); SendReply (ss.str ());
} }
} }
void HTTPConnection::AsyncStreamReceive ()
{
if (m_Stream)
m_Stream->AsyncReceive (boost::asio::buffer (m_StreamBuffer, 8192),
boost::protect (boost::bind (&HTTPConnection::HandleStreamReceive, this,
boost::asio::placeholders::error, boost::asio::placeholders::bytes_transferred)),
30); // 30 seconds timeout
}
void HTTPConnection::HandleStreamReceive (const boost::system::error_code& ecode, std::size_t bytes_transferred) void HTTPConnection::HandleStreamReceive (const boost::system::error_code& ecode, std::size_t bytes_transferred)
{ {
if (bytes_transferred) if (bytes_transferred)
{ {
boost::asio::async_write (*m_Socket, boost::asio::buffer (m_StreamBuffer, bytes_transferred),
boost::bind (&HTTPConnection::HandleWrite, this, boost::asio::placeholders::error));
} }
else else
{ {
SendReply ("Not responding"); if (m_Stream && m_Stream->IsOpen ())
SendReply ("Not responding");
else
Terminate ();
} }
} }
@ -285,8 +307,8 @@ namespace util
m_Reply.headers[1].value = "text/html"; m_Reply.headers[1].value = "text/html";
boost::asio::async_write (*m_Socket, m_Reply.to_buffers(), boost::asio::async_write (*m_Socket, m_Reply.to_buffers(),
boost::bind (&HTTPConnection::HandleWrite, this, boost::bind (&HTTPConnection::HandleWriteReply, this,
boost::asio::placeholders::error, true)); boost::asio::placeholders::error));
} }
HTTPServer::HTTPServer (int port): HTTPServer::HTTPServer (int port):

10
HTTPServer.h
Unescape View File

@ -5,6 +5,7 @@
#include <thread> #include <thread>
#include <boost/asio.hpp> #include <boost/asio.hpp>
#include <boost/array.hpp> #include <boost/array.hpp>
#include "Streaming.h"
namespace i2p namespace i2p
{ {
@ -37,7 +38,7 @@ namespace util
public: public:
HTTPConnection (boost::asio::ip::tcp::socket * socket): m_Socket (socket) { Receive (); }; HTTPConnection (boost::asio::ip::tcp::socket * socket): m_Socket (socket), m_Stream (nullptr) { Receive (); };
~HTTPConnection () { delete m_Socket; } ~HTTPConnection () { delete m_Socket; }
private: private:
@ -45,8 +46,10 @@ namespace util
void Terminate (); void Terminate ();
void Receive (); void Receive ();
void HandleReceive (const boost::system::error_code& ecode, std::size_t bytes_transferred); void HandleReceive (const boost::system::error_code& ecode, std::size_t bytes_transferred);
void AsyncStreamReceive ();
void HandleStreamReceive (const boost::system::error_code& ecode, std::size_t bytes_transferred); void HandleStreamReceive (const boost::system::error_code& ecode, std::size_t bytes_transferred);
void HandleWrite(const boost::system::error_code& ecode, bool terminate); void HandleWriteReply(const boost::system::error_code& ecode);
void HandleWrite (const boost::system::error_code& ecode);
void SendReply (const std::string& content); void SendReply (const std::string& content);
void HandleRequest (); void HandleRequest ();
@ -57,7 +60,8 @@ namespace util
private: private:
boost::asio::ip::tcp::socket * m_Socket; boost::asio::ip::tcp::socket * m_Socket;
char m_Buffer[8192]; i2p::stream::Stream * m_Stream;
char m_Buffer[8192], m_StreamBuffer[8192];
request m_Request; request m_Request;
reply m_Reply; reply m_Reply;
}; };

Write Preview
Loading…
Cancel
Save