jyapayne/Simple-Web-Server
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Cleanup of chunked transfer: now writes directly to request/response streambuf

Browse source
This commit is contained in:
eidheim 2020-01-14 12:44:26 +01:00
commit 5544612a19
2 changed files with 97 additions and 92 deletions
Download patch file Download diff file Expand all files Collapse all files

90
client_http.hpp
View file

@ -561,16 +561,17 @@ namespace SimpleWeb {
session->callback(ec); session->callback(ec);
} }
else if((header_it = session->response->header.find("Transfer-Encoding")) != session->response->header.end() && header_it->second == "chunked") { else if((header_it = session->response->header.find("Transfer-Encoding")) != session->response->header.end() && header_it->second == "chunked") {
auto chunks_streambuf = std::make_shared<asio::streambuf>(this->config.max_response_streambuf_size); // Expect hex number to not exceed 16 bytes (64-bit number), but take into account previous additional read bytes
auto chunk_size_streambuf = std::make_shared<asio::streambuf>(std::max<std::size_t>(16 + 2, session->response->streambuf.size()));
// Copy leftover bytes // Copy leftover bytes
std::ostream ostream(chunks_streambuf.get()); std::ostream ostream(chunk_size_streambuf.get());
auto size = session->response->streambuf.size(); auto size = session->response->streambuf.size();
std::unique_ptr<char[]> buffer(new char[size]); std::unique_ptr<char[]> buffer(new char[size]);
session->response->content.read(buffer.get(), static_cast<std::streamsize>(size)); session->response->content.read(buffer.get(), static_cast<std::streamsize>(size));
ostream.write(buffer.get(), static_cast<std::streamsize>(size)); ostream.write(buffer.get(), static_cast<std::streamsize>(size));
this->read_chunked_transfer_encoded(session, chunks_streambuf); this->read_chunked_transfer_encoded(session, chunk_size_streambuf);
} }
else if(session->response->http_version < "1.1" || ((header_it = session->response->header.find("Session")) != session->response->header.end() && header_it->second == "close")) { else if(session->response->http_version < "1.1" || ((header_it = session->response->header.find("Session")) != session->response->header.end() && header_it->second == "close")) {
session->connection->set_timeout(); session->connection->set_timeout();
@ -639,83 +640,86 @@ namespace SimpleWeb {
}); });
} }
void read_chunked_transfer_encoded(const std::shared_ptr<Session> &session, const std::shared_ptr<asio::streambuf> &chunks_streambuf) { void read_chunked_transfer_encoded(const std::shared_ptr<Session> &session, const std::shared_ptr<asio::streambuf> &chunk_size_streambuf) {
session->connection->set_timeout(); session->connection->set_timeout();
asio::async_read_until(*session->connection->socket, *chunks_streambuf, "\r\n", [this, session, chunks_streambuf](const error_code &ec, size_t bytes_transferred) { asio::async_read_until(*session->connection->socket, *chunk_size_streambuf, "\r\n", [this, session, chunk_size_streambuf](const error_code &ec, size_t bytes_transferred) {
session->connection->cancel_timeout(); session->connection->cancel_timeout();
auto lock = session->connection->handler_runner->continue_lock(); auto lock = session->connection->handler_runner->continue_lock();
if(!lock) if(!lock)
return; return;
if(chunks_streambuf->size() == chunks_streambuf->max_size()) {
session->callback(make_error_code::make_error_code(errc::message_size));
return;
}
if(!ec) { if(!ec) {
std::istream istream(chunks_streambuf.get()); std::istream istream(chunk_size_streambuf.get());
std::string line; std::string line;
getline(istream, line); getline(istream, line);
bytes_transferred -= line.size() + 1; bytes_transferred -= line.size() + 1;
unsigned long length = 0; unsigned long chunk_size = 0;
try { try {
length = stoul(line, 0, 16); chunk_size = stoul(line, 0, 16);
} }
catch(...) { catch(...) {
session->callback(make_error_code::make_error_code(errc::protocol_error)); session->callback(make_error_code::make_error_code(errc::protocol_error));
return; return;
} }
auto num_additional_bytes = chunks_streambuf->size() - bytes_transferred; auto num_additional_bytes = chunk_size_streambuf->size() - bytes_transferred;
if((2 + length) > num_additional_bytes) { auto bytes_to_move = std::min<std::size_t>(chunk_size, num_additional_bytes);
if(bytes_to_move > 0) {
std::ostream ostream(&session->response->streambuf);
std::unique_ptr<char[]> buffer(new char[bytes_to_move]);
istream.read(buffer.get(), static_cast<std::streamsize>(bytes_to_move));
ostream.write(buffer.get(), static_cast<std::streamsize>(bytes_to_move));
if(session->response->streambuf.size() == session->response->streambuf.max_size()) {
session->callback(make_error_code::make_error_code(errc::message_size));
return;
}
}
if((2 + chunk_size) > num_additional_bytes) {
session->connection->set_timeout(); session->connection->set_timeout();
asio::async_read(*session->connection->socket, *chunks_streambuf, asio::transfer_exactly(2 + length - num_additional_bytes), [this, session, chunks_streambuf, length](const error_code &ec, size_t /*bytes_transferred*/) { asio::async_read(*session->connection->socket, session->response->streambuf, asio::transfer_exactly(2 + chunk_size - num_additional_bytes), [this, session, chunk_size_streambuf, chunk_size](const error_code &ec, size_t /*bytes_transferred*/) {
session->connection->cancel_timeout(); session->connection->cancel_timeout();
auto lock = session->connection->handler_runner->continue_lock(); auto lock = session->connection->handler_runner->continue_lock();
if(!lock) if(!lock)
return; return;
if(chunks_streambuf->size() == chunks_streambuf->max_size()) { if(session->response->streambuf.size() == session->response->streambuf.max_size()) {
session->callback(make_error_code::make_error_code(errc::message_size)); session->callback(make_error_code::make_error_code(errc::message_size));
return; return;
} }
if(!ec) if(!ec) {
this->read_chunked_transfer_encoded_chunk(session, chunks_streambuf, length); std::istream istream(&session->response->streambuf);
// Remove "\r\n"
istream.get();
istream.get();
if(chunk_size > 0)
read_chunked_transfer_encoded(session, chunk_size_streambuf);
else
session->callback(error_code());
}
else else
session->callback(ec); session->callback(ec);
}); });
} }
else else {
this->read_chunked_transfer_encoded_chunk(session, chunks_streambuf, length); // Remove "\r\n"
istream.get();
istream.get();
if(chunk_size > 0)
read_chunked_transfer_encoded(session, chunk_size_streambuf);
else
session->callback(error_code());
}
} }
else else
session->callback(ec); session->callback(ec);
}); });
} }
void read_chunked_transfer_encoded_chunk(const std::shared_ptr<Session> &session, const std::shared_ptr<asio::streambuf> &chunks_streambuf, unsigned long length) {
std::istream istream(chunks_streambuf.get());
if(length > 0) {
std::ostream ostream(&session->response->streambuf);
std::unique_ptr<char[]> buffer(new char[length]);
istream.read(buffer.get(), static_cast<std::streamsize>(length));
ostream.write(buffer.get(), static_cast<std::streamsize>(length));
if(session->response->streambuf.size() == session->response->streambuf.max_size()) {
session->callback(make_error_code::make_error_code(errc::message_size));
return;
}
}
// Remove "\r\n"
istream.get();
istream.get();
if(length > 0)
read_chunked_transfer_encoded(session, chunks_streambuf);
else
session->callback(error_code());
}
void read_server_sent_event(const std::shared_ptr<Session> &session, const std::shared_ptr<asio::streambuf> &events_streambuf) { void read_server_sent_event(const std::shared_ptr<Session> &session, const std::shared_ptr<asio::streambuf> &events_streambuf) {
session->connection->set_timeout(); session->connection->set_timeout();
asio::async_read_until(*session->connection->socket, *events_streambuf, HeaderEndMatch(), [this, session, events_streambuf](const error_code &ec, std::size_t /*bytes_transferred*/) { asio::async_read_until(*session->connection->socket, *events_streambuf, HeaderEndMatch(), [this, session, events_streambuf](const error_code &ec, std::size_t /*bytes_transferred*/) {

99
server_http.hpp
View file

@ -595,16 +595,17 @@ namespace SimpleWeb {
this->find_resource(session); this->find_resource(session);
} }
else if((header_it = session->request->header.find("Transfer-Encoding")) != session->request->header.end() && header_it->second == "chunked") { else if((header_it = session->request->header.find("Transfer-Encoding")) != session->request->header.end() && header_it->second == "chunked") {
auto chunks_streambuf = std::make_shared<asio::streambuf>(this->config.max_request_streambuf_size); // Expect hex number to not exceed 16 bytes (64-bit number), but take into account previous additional read bytes
auto chunk_size_streambuf = std::make_shared<asio::streambuf>(std::max<std::size_t>(16 + 2, session->request->streambuf.size()));
// Copy leftover bytes // Copy leftover bytes
std::ostream ostream(chunks_streambuf.get()); std::ostream ostream(chunk_size_streambuf.get());
auto size = session->request->streambuf.size(); auto size = session->request->streambuf.size();
std::unique_ptr<char[]> buffer(new char[size]); std::unique_ptr<char[]> buffer(new char[size]);
session->request->content.read(buffer.get(), static_cast<std::streamsize>(size)); session->request->content.read(buffer.get(), static_cast<std::streamsize>(size));
ostream.write(buffer.get(), static_cast<std::streamsize>(size)); ostream.write(buffer.get(), static_cast<std::streamsize>(size));
this->read_chunked_transfer_encoded(session, chunks_streambuf); this->read_chunked_transfer_encoded(session, chunk_size_streambuf);
} }
else else
this->find_resource(session); this->find_resource(session);
@ -614,29 +615,22 @@ namespace SimpleWeb {
}); });
} }
void read_chunked_transfer_encoded(const std::shared_ptr<Session> &session, const std::shared_ptr<asio::streambuf> &chunks_streambuf) { void read_chunked_transfer_encoded(const std::shared_ptr<Session> &session, const std::shared_ptr<asio::streambuf> &chunk_size_streambuf) {
session->connection->set_timeout(config.timeout_content); session->connection->set_timeout(config.timeout_content);
asio::async_read_until(*session->connection->socket, *chunks_streambuf, "\r\n", [this, session, chunks_streambuf](const error_code &ec, size_t bytes_transferred) { asio::async_read_until(*session->connection->socket, *chunk_size_streambuf, "\r\n", [this, session, chunk_size_streambuf](const error_code &ec, size_t bytes_transferred) {
session->connection->cancel_timeout(); session->connection->cancel_timeout();
auto lock = session->connection->handler_runner->continue_lock(); auto lock = session->connection->handler_runner->continue_lock();
if(!lock) if(!lock)
return; return;
if(chunks_streambuf->size() == chunks_streambuf->max_size()) {
auto response = std::shared_ptr<Response>(new Response(session, this->config.timeout_content));
response->write(StatusCode::client_error_payload_too_large);
if(this->on_error)
this->on_error(session->request, make_error_code::make_error_code(errc::message_size));
return;
}
if(!ec) { if(!ec) {
std::istream istream(chunks_streambuf.get()); std::istream istream(chunk_size_streambuf.get());
std::string line; std::string line;
getline(istream, line); getline(istream, line);
bytes_transferred -= line.size() + 1; bytes_transferred -= line.size() + 1;
unsigned long length = 0; unsigned long chunk_size = 0;
try { try {
length = stoul(line, 0, 16); chunk_size = stoul(line, 0, 16);
} }
catch(...) { catch(...) {
if(this->on_error) if(this->on_error)
@ -644,16 +638,31 @@ namespace SimpleWeb {
return; return;
} }
auto num_additional_bytes = chunks_streambuf->size() - bytes_transferred; auto num_additional_bytes = chunk_size_streambuf->size() - bytes_transferred;
if((2 + length) > num_additional_bytes) { auto bytes_to_move = std::min<std::size_t>(chunk_size, num_additional_bytes);
if(bytes_to_move > 0) {
std::ostream ostream(&session->request->streambuf);
std::unique_ptr<char[]> buffer(new char[bytes_to_move]);
istream.read(buffer.get(), static_cast<std::streamsize>(bytes_to_move));
ostream.write(buffer.get(), static_cast<std::streamsize>(bytes_to_move));
if(session->request->streambuf.size() == session->request->streambuf.max_size()) {
auto response = std::shared_ptr<Response>(new Response(session, this->config.timeout_content));
response->write(StatusCode::client_error_payload_too_large);
if(this->on_error)
this->on_error(session->request, make_error_code::make_error_code(errc::message_size));
return;
}
}
if((2 + chunk_size) > num_additional_bytes) {
session->connection->set_timeout(config.timeout_content); session->connection->set_timeout(config.timeout_content);
asio::async_read(*session->connection->socket, *chunks_streambuf, asio::transfer_exactly(2 + length - num_additional_bytes), [this, session, chunks_streambuf, length](const error_code &ec, size_t /*bytes_transferred*/) { asio::async_read(*session->connection->socket, session->request->streambuf, asio::transfer_exactly(2 + chunk_size - num_additional_bytes), [this, session, chunk_size_streambuf, chunk_size](const error_code &ec, size_t /*bytes_transferred*/) {
session->connection->cancel_timeout(); session->connection->cancel_timeout();
auto lock = session->connection->handler_runner->continue_lock(); auto lock = session->connection->handler_runner->continue_lock();
if(!lock) if(!lock)
return; return;
if(chunks_streambuf->size() == chunks_streambuf->max_size()) { if(chunk_size_streambuf->size() == chunk_size_streambuf->max_size()) {
auto response = std::shared_ptr<Response>(new Response(session, this->config.timeout_content)); auto response = std::shared_ptr<Response>(new Response(session, this->config.timeout_content));
response->write(StatusCode::client_error_payload_too_large); response->write(StatusCode::client_error_payload_too_large);
if(this->on_error) if(this->on_error)
@ -661,46 +670,38 @@ namespace SimpleWeb {
return; return;
} }
if(!ec) if(!ec) {
this->read_chunked_transfer_encoded_chunk(session, chunks_streambuf, length); std::istream istream(&session->request->streambuf);
// Remove "\r\n"
istream.get();
istream.get();
if(chunk_size > 0)
read_chunked_transfer_encoded(session, chunk_size_streambuf);
else
this->find_resource(session);
}
else if(this->on_error) else if(this->on_error)
this->on_error(session->request, ec); this->on_error(session->request, ec);
}); });
} }
else else {
this->read_chunked_transfer_encoded_chunk(session, chunks_streambuf, length); // Remove "\r\n"
istream.get();
istream.get();
if(chunk_size > 0)
read_chunked_transfer_encoded(session, chunk_size_streambuf);
else
this->find_resource(session);
}
} }
else if(this->on_error) else if(this->on_error)
this->on_error(session->request, ec); this->on_error(session->request, ec);
}); });
} }
void read_chunked_transfer_encoded_chunk(const std::shared_ptr<Session> &session, const std::shared_ptr<asio::streambuf> &chunks_streambuf, unsigned long length) {
std::istream istream(chunks_streambuf.get());
if(length > 0) {
std::ostream ostream(&session->request->streambuf);
std::unique_ptr<char[]> buffer(new char[length]);
istream.read(buffer.get(), static_cast<std::streamsize>(length));
ostream.write(buffer.get(), static_cast<std::streamsize>(length));
if(session->request->streambuf.size() == session->request->streambuf.max_size()) {
auto response = std::shared_ptr<Response>(new Response(session, this->config.timeout_content));
response->write(StatusCode::client_error_payload_too_large);
if(this->on_error)
this->on_error(session->request, make_error_code::make_error_code(errc::message_size));
return;
}
}
// Remove "\r\n"
istream.get();
istream.get();
if(length > 0)
read_chunked_transfer_encoded(session, chunks_streambuf);
else
this->find_resource(session);
}
void find_resource(const std::shared_ptr<Session> &session) { void find_resource(const std::shared_ptr<Session> &session) {
// Upgrade connection // Upgrade connection
if(on_upgrade) { if(on_upgrade) {