Commit 5d82001a authored by octal's avatar octal

Started working on file serving

parent 1dac2f9a
...@@ -50,6 +50,13 @@ class MyHandler : public Net::Http::Handler { ...@@ -50,6 +50,13 @@ class MyHandler : public Net::Http::Handler {
else if (req.resource() == "/timeout") { else if (req.resource() == "/timeout") {
timeout.arm(std::chrono::seconds(5)); timeout.arm(std::chrono::seconds(5));
} }
else if (req.resource() == "/image") {
if (req.method() == Net::Http::Method::Get) {
response.sendFile(Net::Http::Code::Ok, "datacratic.png").then([](ssize_t bytes) {;
std::cout << "Sent " << bytes << " bytes" << std::endl;
}, Async::NoExcept);
}
}
} }
void onTimeout(const Net::Http::Request& req, Net::Http::Response response) { void onTimeout(const Net::Http::Request& req, Net::Http::Response response) {
...@@ -80,7 +87,7 @@ struct LoadMonitor { ...@@ -80,7 +87,7 @@ struct LoadMonitor {
~LoadMonitor() { ~LoadMonitor() {
shutdown_ = true; shutdown_ = true;
thread->join(); if (thread) thread->join();
} }
private: private:
......
...@@ -9,6 +9,11 @@ ...@@ -9,6 +9,11 @@
#include <stdexcept> #include <stdexcept>
#include <ctime> #include <ctime>
#include <iomanip> #include <iomanip>
#include <sys/types.h>
#include <sys/stat.h>
#include <unistd.h>
#include <sys/stat.h>
#include <fcntl.h>
#include "common.h" #include "common.h"
#include "http.h" #include "http.h"
#include "net.h" #include "net.h"
...@@ -21,7 +26,7 @@ namespace Net { ...@@ -21,7 +26,7 @@ namespace Net {
namespace Http { namespace Http {
template<typename H, typename Stream, typename... Args> template<typename H, typename Stream, typename... Args>
typename std::enable_if<Header::IsHeader<H>::value, void>::type typename std::enable_if<Header::IsHeader<H>::value, Stream&>::type
writeHeader(Stream& stream, Args&& ...args) { writeHeader(Stream& stream, Args&& ...args) {
H header(std::forward<Args>(args)...); H header(std::forward<Args>(args)...);
...@@ -29,10 +34,55 @@ writeHeader(Stream& stream, Args&& ...args) { ...@@ -29,10 +34,55 @@ writeHeader(Stream& stream, Args&& ...args) {
header.write(stream); header.write(stream);
stream << crlf; stream << crlf;
return stream;
} }
static constexpr const char* ParserData = "__Parser"; static constexpr const char* ParserData = "__Parser";
namespace {
bool writeStatusLine(Code code, DynamicStreamBuf& buf) {
#define OUT(...) \
do { \
__VA_ARGS__; \
if (!os) return false; \
} while (0)
std::ostream os(&buf);
OUT(os << "HTTP/1.1 ");
OUT(os << static_cast<int>(code));
OUT(os << ' ');
OUT(os << code);
OUT(os << crlf);
return true;
#undef OUT
}
bool writeHeaders(const Header::Collection& headers, DynamicStreamBuf& buf) {
#define OUT(...) \
do { \
__VA_ARGS__; \
if (!os) return false; \
} while (0)
std::ostream os(&buf);
for (const auto& header: headers.list()) {
OUT(os << header->name() << ": ");
OUT(header->write(os));
OUT(os << crlf);
}
return true;
#undef OUT
}
}
namespace Private { namespace Private {
void void
...@@ -352,50 +402,28 @@ Request::peer() const { ...@@ -352,50 +402,28 @@ Request::peer() const {
} }
#endif #endif
void ResponseStream::ResponseStream(
ResponseStream::writeStatusLine() { Message&& other,
std::ostream os(&buf_); std::weak_ptr<Tcp::Peer> peer,
Tcp::IoWorker* io,
#define OUT(...) \ size_t streamSize)
do { \ : Message(std::move(other))
__VA_ARGS__; \ , peer_(std::move(peer))
if (!os) { \ , buf_(streamSize)
throw Error("Response exceeded buffer size"); \ , io_(io)
} \ {
} while (0); if (!writeStatusLine(code_, buf_))
throw Error("Response exceeded buffer size");
OUT(os << "HTTP/1.1 ");
OUT(os << static_cast<int>(code_));
OUT(os << ' ');
OUT(os << code_);
OUT(os << crlf);
#undef OUT
}
void
ResponseStream::writeHeaders() {
std::ostream os(&buf_);
#define OUT(...) \ if (writeHeaders(headers_, buf_)) {
do { \ std::ostream os(&buf_);
__VA_ARGS__; \ if (!writeHeader<Header::TransferEncoding>(os, Header::Encoding::Chunked))
if (!os) { \ throw Error("Response exceeded buffer size");
throw Error("Response exceeded buffer size"); \ os << crlf;
} \ }
} while (0); else {
throw Error("Response exceeded buffer size");
for (const auto& header: headers_.list()) {
OUT(os << header->name() << ": ");
OUT(header->write(os));
OUT(os << crlf);
} }
OUT(writeHeader<Header::TransferEncoding>(os, Header::Encoding::Chunked));
OUT(os << crlf);
#undef OUT
} }
void void
...@@ -420,6 +448,20 @@ ResponseStream::ends() { ...@@ -420,6 +448,20 @@ ResponseStream::ends() {
flush(); flush();
} }
Async::Promise<ssize_t>
Response::sendFile(Code code, const std::string& file) {
int fd = open(file.c_str(), O_RDONLY);
if (fd == -1) {
/* @Improvement: maybe could we check for errno here and emit a different error
message
*/
throw HttpError(Net::Http::Code::Not_Found, "");
}
code_ = code;
return putOnWire(fd);
}
Async::Promise<ssize_t> Async::Promise<ssize_t>
Response::putOnWire(const char* data, size_t len) Response::putOnWire(const char* data, size_t len)
{ {
...@@ -434,17 +476,8 @@ Response::putOnWire(const char* data, size_t len) ...@@ -434,17 +476,8 @@ Response::putOnWire(const char* data, size_t len)
} \ } \
} while (0); } while (0);
OUT(os << "HTTP/1.1 "); OUT(writeStatusLine(code_, buf_));
OUT(os << static_cast<int>(code_)); OUT(writeHeaders(headers_, buf_));
OUT(os << ' ');
OUT(os << code_);
OUT(os << crlf);
for (const auto& header: headers_.list()) {
OUT(os << header->name() << ": ");
OUT(header->write(os));
OUT(os << crlf);
}
OUT(writeHeader<Header::ContentLength>(os, len)); OUT(writeHeader<Header::ContentLength>(os, len));
...@@ -467,6 +500,45 @@ Response::putOnWire(const char* data, size_t len) ...@@ -467,6 +500,45 @@ Response::putOnWire(const char* data, size_t len)
} }
} }
Async::Promise<ssize_t>
Response::putOnWire(Fd fd)
{
struct stat sb;
int res = ::fstat(fd, &sb);
if (res == -1) {
throw HttpError(Code::Internal_Server_Error, "");
}
std::ostream os(&buf_);
#define OUT(...) \
do { \
__VA_ARGS__; \
if (!os) { \
return Async::Promise<ssize_t>::rejected(Error("Response exceeded buffer size")); \
} \
} while (0);
OUT(writeStatusLine(code_, buf_));
OUT(writeHeaders(headers_, buf_));
const size_t len = sb.st_size;
OUT(writeHeader<Header::ContentLength>(os, len));
OUT(os << crlf);
auto buffer = buf_.buffer();
return peer()->send(buffer.data, buffer.len, MSG_MORE).then([=](ssize_t bytes) {
return peer()->sendFile(fd, len);
}, Async::Throw);
#undef OUT
}
void void
Handler::onInput(const char* buffer, size_t len, const std::shared_ptr<Tcp::Peer>& peer) { Handler::onInput(const char* buffer, size_t len, const std::shared_ptr<Tcp::Peer>& peer) {
try { try {
......
...@@ -219,18 +219,7 @@ private: ...@@ -219,18 +219,7 @@ private:
Message&& other, Message&& other,
std::weak_ptr<Tcp::Peer> peer, std::weak_ptr<Tcp::Peer> peer,
Tcp::IoWorker* io, Tcp::IoWorker* io,
size_t streamSize) size_t streamSize);
: Message(std::move(other))
, peer_(std::move(peer))
, buf_(streamSize)
, io_(io)
{
writeStatusLine();
writeHeaders();
}
void writeStatusLine();
void writeHeaders();
std::shared_ptr<Tcp::Peer> peer() const { std::shared_ptr<Tcp::Peer> peer() const {
if (peer_.expired()) if (peer_.expired())
...@@ -357,6 +346,8 @@ public: ...@@ -357,6 +346,8 @@ public:
return putOnWire(arr, N - 1); return putOnWire(arr, N - 1);
} }
Async::Promise<ssize_t> sendFile(Code code, const std::string& file);
ResponseStream stream(Code code, size_t streamSize = DefaultStreamSize) { ResponseStream stream(Code code, size_t streamSize = DefaultStreamSize) {
code_ = code; code_ = code;
...@@ -387,6 +378,7 @@ private: ...@@ -387,6 +378,7 @@ private:
} }
Async::Promise<ssize_t> putOnWire(const char* data, size_t len); Async::Promise<ssize_t> putOnWire(const char* data, size_t len);
Async::Promise<ssize_t> putOnWire(Fd fd);
std::weak_ptr<Tcp::Peer> peer_; std::weak_ptr<Tcp::Peer> peer_;
DynamicStreamBuf buf_; DynamicStreamBuf buf_;
......
...@@ -10,6 +10,7 @@ ...@@ -10,6 +10,7 @@
#include "peer.h" #include "peer.h"
#include "os.h" #include "os.h"
#include <sys/timerfd.h> #include <sys/timerfd.h>
#include <sys/sendfile.h>
namespace Net { namespace Net {
...@@ -339,7 +340,7 @@ IoWorker::handleNotify() { ...@@ -339,7 +340,7 @@ IoWorker::handleNotify() {
} }
Async::Promise<ssize_t> Async::Promise<ssize_t>
IoWorker::asyncWrite(Fd fd, const void* buf, size_t len) { IoWorker::asyncWrite(Fd fd, const void* buf, size_t len, int flags) {
return Async::Promise<ssize_t>([=](Async::Resolver& resolve, Async::Rejection& reject) { return Async::Promise<ssize_t>([=](Async::Resolver& resolve, Async::Rejection& reject) {
auto it = toWrite.find(fd); auto it = toWrite.find(fd);
...@@ -352,7 +353,7 @@ IoWorker::asyncWrite(Fd fd, const void* buf, size_t len) { ...@@ -352,7 +353,7 @@ IoWorker::asyncWrite(Fd fd, const void* buf, size_t len) {
for (;;) { for (;;) {
auto *bufPtr = static_cast<const char *>(buf) + totalWritten; auto *bufPtr = static_cast<const char *>(buf) + totalWritten;
auto bufLen = len - totalWritten; auto bufLen = len - totalWritten;
ssize_t bytesWritten = ::send(fd, bufPtr, bufLen, 0); ssize_t bytesWritten = ::send(fd, bufPtr, bufLen, flags);
if (bytesWritten < 0) { if (bytesWritten < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) { if (errno == EAGAIN || errno == EWOULDBLOCK) {
toWrite.insert( toWrite.insert(
...@@ -376,6 +377,35 @@ IoWorker::asyncWrite(Fd fd, const void* buf, size_t len) { ...@@ -376,6 +377,35 @@ IoWorker::asyncWrite(Fd fd, const void* buf, size_t len) {
}); });
} }
Async::Promise<ssize_t>
IoWorker::asyncWriteFile(Fd peer_fd, Fd fd, size_t size) {
return Async::Promise<ssize_t>([=](Async::Resolver& resolve, Async::Rejection& reject) {
ssize_t totalWritten = 0;
off_t off = 0;
for (;;) {
ssize_t bytes = ::sendfile(peer_fd, fd, &off, size - off);
if (bytes < 0) {
if (errno == EAGAIN || errno == EWOULDBLOCK) {
reject(Net::Error("Incomplete"));
}
else {
reject(Net::Error::system("Could not send file"));
}
break;
}
else {
totalWritten += bytes;
if (totalWritten == size) {
resolve(totalWritten);
break;
}
off = totalWritten;
}
}
});
}
} // namespace Tcp } // namespace Tcp
......
...@@ -108,6 +108,11 @@ private: ...@@ -108,6 +108,11 @@ private:
std::unique_ptr<std::thread> thread; std::unique_ptr<std::thread> thread;
mutable std::mutex peersMutex; mutable std::mutex peersMutex;
std::unordered_map<Fd, std::shared_ptr<Peer>> peers; std::unordered_map<Fd, std::shared_ptr<Peer>> peers;
/* @Incomplete: this should be a std::dequeue.
If an asyncWrite on a particular fd is initiated whereas the fd is not write-ready
yet and some writes are still on-hold, writes should queue-up so that when the
fd becomes ready again, we can write everything
*/
std::unordered_map<Fd, OnHoldWrite> toWrite; std::unordered_map<Fd, OnHoldWrite> toWrite;
Optional<Timer> timer; Optional<Timer> timer;
...@@ -124,7 +129,8 @@ private: ...@@ -124,7 +129,8 @@ private:
std::shared_ptr<Peer>& getPeer(Fd fd); std::shared_ptr<Peer>& getPeer(Fd fd);
std::shared_ptr<Peer>& getPeer(Polling::Tag tag); std::shared_ptr<Peer>& getPeer(Polling::Tag tag);
Async::Promise<ssize_t> asyncWrite(Fd fd, const void *buf, size_t len); Async::Promise<ssize_t> asyncWrite(Fd fd, const void *buf, size_t len, int flags = 0);
Async::Promise<ssize_t> asyncWriteFile(Fd peer_fd, Fd fd, size_t size);
void handlePeerDisconnection(const std::shared_ptr<Peer>& peer); void handlePeerDisconnection(const std::shared_ptr<Peer>& peer);
......
...@@ -78,10 +78,14 @@ Peer::tryGetData(std::string(name)) const { ...@@ -78,10 +78,14 @@ Peer::tryGetData(std::string(name)) const {
} }
Async::Promise<ssize_t> Async::Promise<ssize_t>
Peer::send(const void* buf, size_t len) { Peer::send(const void* buf, size_t len, int flags) {
return io_->asyncWrite(fd_, buf, len); return io_->asyncWrite(fd_, buf, len, flags);
} }
Async::Promise<ssize_t>
Peer::sendFile(Fd fd, size_t len) {
return io_->asyncWriteFile(this->fd_, fd, len);
}
std::ostream& operator<<(std::ostream& os, const Peer& peer) { std::ostream& operator<<(std::ostream& os, const Peer& peer) {
const auto& addr = peer.address(); const auto& addr = peer.address();
......
...@@ -50,7 +50,8 @@ public: ...@@ -50,7 +50,8 @@ public:
return std::static_pointer_cast<T>(data); return std::static_pointer_cast<T>(data);
} }
Async::Promise<ssize_t> send(const void* buf, size_t len); Async::Promise<ssize_t> send(const void* buf, size_t len, int flags = 0);
Async::Promise<ssize_t> sendFile(Fd fd, size_t len);
private: private:
IoWorker* io_; IoWorker* io_;
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment