Mirrors/i2pd
2
0
Fork 0
mirror of https://github.com/PurpleI2P/i2pd.git synced 2025-02-08 22:13:48 +01:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

Merge pull request #21 from orignal/master

Browse source 
Merge pull request from orignal/master
This commit is contained in:
chertov 2014-03-29 17:34:52 +03:00
commit acb119eea4
4 changed files with 81 additions and 51 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

108
HTTPServer.cpp
View file

@ -1,4 +1,5 @@
#include <boost/bind.hpp> #include <boost/bind.hpp>
#include <boost/bind/protect.hpp>
#include <boost/lexical_cast.hpp> #include <boost/lexical_cast.hpp>
#include "base64.h" #include "base64.h"
#include "Log.h" #include "Log.h"
@ -6,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
@ -43,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;
} }
@ -76,10 +81,6 @@ namespace util
} }
else else
HandleRequest (); HandleRequest ();
boost::asio::async_write (*m_Socket, m_Reply.to_buffers(),
boost::bind (&HTTPConnection::HandleWrite, this,
boost::asio::placeholders::error));
//Receive ();
} }
else if (ecode != boost::asio::error::operation_aborted) else if (ecode != boost::asio::error::operation_aborted)
Terminate (); Terminate ();
@ -97,23 +98,26 @@ namespace util
return ""; return "";
} }
void HTTPConnection::HandleWrite (const boost::system::error_code& ecode) void HTTPConnection::HandleWriteReply (const boost::system::error_code& ecode)
{ {
Terminate (); Terminate ();
} }
void HTTPConnection::HandleWrite (const boost::system::error_code& ecode)
{
if (ecode || (m_Stream && !m_Stream->IsOpen ()))
Terminate ();
else // data keeps coming
AsyncStreamReceive ();
}
void HTTPConnection::HandleRequest () void HTTPConnection::HandleRequest ()
{ {
std::stringstream s; std::stringstream s;
s << "<html>"; s << "<html>";
FillContent (s); FillContent (s);
s << "</html>"; s << "</html>";
m_Reply.content = s.str (); SendReply (s.str ());
m_Reply.headers.resize(2);
m_Reply.headers[0].name = "Content-Length";
m_Reply.headers[0].value = boost::lexical_cast<std::string>(m_Reply.content.size());
m_Reply.headers[1].name = "Content-Type";
m_Reply.headers[1].value = "text/html";
} }
void HTTPConnection::FillContent (std::stringstream& s) void HTTPConnection::FillContent (std::stringstream& s)
@ -211,6 +215,7 @@ namespace util
if (!addr) if (!addr)
{ {
LogPrint ("Unknown address ", address); LogPrint ("Unknown address ", address);
SendReply ("Unknown address");
return; return;
} }
destination = *addr; destination = *addr;
@ -221,6 +226,7 @@ namespace util
if (i2p::data::Base32ToByteStream (address.c_str (), address.length (), (uint8_t *)destination, 32) != 32) if (i2p::data::Base32ToByteStream (address.c_str (), address.length (), (uint8_t *)destination, 32) != 32)
{ {
LogPrint ("Invalid Base32 address ", address); LogPrint ("Invalid Base32 address ", address);
SendReply ("Invalid Base32 address");
return; return;
} }
fullAddress = address + ".b32.i2p"; fullAddress = address + ".b32.i2p";
@ -234,51 +240,57 @@ namespace util
leaseSet = i2p::data::netdb.FindLeaseSet (destination); leaseSet = i2p::data::netdb.FindLeaseSet (destination);
if (!leaseSet || !leaseSet->HasNonExpiredLeases ()) // still no LeaseSet if (!leaseSet || !leaseSet->HasNonExpiredLeases ()) // still no LeaseSet
{ {
m_Reply.content = leaseSet ? "<html>Leases expired</html>" : "<html>LeaseSet not found</html>"; SendReply (leaseSet ? "<html>Leases expired</html>" : "<html>LeaseSet not found</html>");
m_Reply.headers.resize(2);
m_Reply.headers[0].name = "Content-Length";
m_Reply.headers[0].value = boost::lexical_cast<std::string>(m_Reply.content.size());
m_Reply.headers[1].name = "Content-Type";
m_Reply.headers[1].value = "text/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; AsyncStreamReceive ();
uint8_t buf[8192];
size_t r = s->Receive (buf, 8192, 30); // 30 seconds
if (!r && s->IsEstablished ()) // nothing received but connection is established
r = s->Receive (buf, 8192, 30); // wait for another 30 secondd
if (r) // we recieved data
{
ss << std::string ((char *)buf, r);
while (s->IsOpen () && (r = s->Receive (buf, 8192, 30)) > 0)
ss << std::string ((char *)buf,r);
m_Reply.content = ss.str (); // send "as is"
m_Reply.headers.resize(0); // no headers
return;
}
else // nothing received
ss << "<html>Not responding</html>";
s->Close ();
DeleteStream (s);
m_Reply.content = ss.str ();
m_Reply.headers.resize(2);
m_Reply.headers[0].name = "Content-Length";
m_Reply.headers[0].value = boost::lexical_cast<std::string>(m_Reply.content.size());
m_Reply.headers[1].name = "Content-Type";
m_Reply.headers[1].value = "text/html";
} }
} }
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)),
45); // 45 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)
{
boost::asio::async_write (*m_Socket, boost::asio::buffer (m_StreamBuffer, bytes_transferred),
boost::bind (&HTTPConnection::HandleWrite, this, boost::asio::placeholders::error));
}
else
{
if (m_Stream && m_Stream->IsOpen ())
SendReply ("Not responding");
else
Terminate ();
}
}
void HTTPConnection::SendReply (const std::string& content)
{
m_Reply.content = content;
m_Reply.headers.resize(2);
m_Reply.headers[0].name = "Content-Length";
m_Reply.headers[0].value = boost::lexical_cast<std::string>(m_Reply.content.size());
m_Reply.headers[1].name = "Content-Type";
m_Reply.headers[1].value = "text/html";
boost::asio::async_write (*m_Socket, m_Reply.to_buffers(),
boost::bind (&HTTPConnection::HandleWriteReply, this,
boost::asio::placeholders::error));
} }
HTTPServer::HTTPServer (int port): HTTPServer::HTTPServer (int port):

11
HTTPServer.h
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,11 @@ 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); void HandleWriteReply(const boost::system::error_code& ecode);
void HandleWrite (const boost::system::error_code& ecode);
void SendReply (const std::string& content);
void HandleRequest (); void HandleRequest ();
void HandleDestinationRequest (const std::string& address, const std::string& uri); void HandleDestinationRequest (const std::string& address, const std::string& uri);
@ -56,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;
}; };

3
Streaming.cpp
View file

@ -115,7 +115,10 @@ namespace stream
packet->offset = packet->GetPayload () - packet->buf; packet->offset = packet->GetPayload () - packet->buf;
if (packet->GetLength () > 0) if (packet->GetLength () > 0)
{
m_ReceiveQueue.Put (packet); m_ReceiveQueue.Put (packet);
m_ReceiveTimer.cancel ();
}
else else
delete packet; delete packet;

10
Streaming.h
View file

@ -200,6 +200,16 @@ namespace stream
template<typename Buffer, typename ReceiveHandler> template<typename Buffer, typename ReceiveHandler>
void Stream::AsyncReceive (const Buffer& buffer, ReceiveHandler handler, int timeout) void Stream::AsyncReceive (const Buffer& buffer, ReceiveHandler handler, int timeout)
{ {
if (!m_ReceiveQueue.IsEmpty ())
{
size_t received = ConcatenatePackets (boost::asio::buffer_cast<uint8_t *>(buffer), boost::asio::buffer_size(buffer));
if (received)
{
// TODO: post to stream's thread
handler (boost::system::error_code (), received);
return;
}
}
m_ReceiveTimer.expires_from_now (boost::posix_time::seconds(timeout)); m_ReceiveTimer.expires_from_now (boost::posix_time::seconds(timeout));
m_ReceiveTimer.async_wait (boost::bind (&Stream::HandleReceiveTimer<Buffer, ReceiveHandler>, m_ReceiveTimer.async_wait (boost::bind (&Stream::HandleReceiveTimer<Buffer, ReceiveHandler>,
this, boost::asio::placeholders::error, buffer, handler)); this, boost::asio::placeholders::error, buffer, handler));