talksik/uWebSockets
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Begin work on AsyncFileReader demo

Browse Source
This commit is contained in:
Alex Hultman
2018-09-23 20:32:47 +02:00
parent 7f3753e9db
commit 2fb87b67c7
5 changed files with 140 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
examples/helpers/AsyncFileReader.h
+82
View File
@@ -0,0 +1,82 @@
#include <map>
#include <cstring>
#include <fstream>
#include <sstream>
#include <iostream>
/* This is just a very simple and inefficient demo of async responses,
* please do roll your own variant or use a database or Node.js's async
* features instead of this really bad demo */
struct AsyncFileReader {
private:
/* The cache we have in memory for this file */
std::string cache;
int cacheOffset = 0;
/* The pending async file read (yes we only support one pending read) */
std::function<void(std::string_view)> pendingReadCb;
int fileSize;
std::ifstream fin;
uWS::Loop *loop;
public:
/* Construct a demo async. file reader for fileName */
AsyncFileReader(std::string fileName) {
fin.open(fileName, std::ios::binary);
// get fileSize
fin.seekg(0, fin.end);
fileSize = fin.tellg();
//std::cout << "File size is: " << fileSize << std::endl;
// cache up 1 mb!
cache.resize(1024 * 1024);
//std::cout << "Caching 1 MB at offset = " << 0 << std::endl;
fin.seekg(0, fin.beg);
fin.read(cache.data(), cache.length());
cacheOffset = 0;
// get loop for thread
loop = uWS::Loop::defaultLoop();
}
/* Returns any data already cached for this offset */
std::string_view peek(int offset) {
/* Did we hit the cache? */
if (offset >= cacheOffset && ((offset - cacheOffset) < cache.length())) {
/* Cache hit */
//std::cout << "Cache hit!" << std::endl;
return std::string_view(cache.data() + offset - cacheOffset, cache.length() - offset + cacheOffset);
} else {
/* Cache miss */
//std::cout << "Cache miss!" << std::endl;
return std::string_view(nullptr, 0);
}
}
/* Asynchronously request more data at offset */
void request(int offset, std::function<void(std::string_view)> cb) {
// std::async this
//std::cout << "Caching 1 MB at offset = " << offset << std::endl;
fin.seekg(offset, fin.beg);
fin.read(cache.data(), cache.length());
cacheOffset = offset;
// Loop.defer this
cb(std::string_view(cache.data(), cache.length()));
}
/* Abort any pending async. request */
void abort() {
}
int getFileSize() {
return fileSize;
}
};
examples/helpers/FileCache.h
-49
View File
@@ -1,49 +0,0 @@
#include <experimental/filesystem>
#include <map>
#include <fstream>
struct FileCache {
private:
std::map<std::string_view, std::string_view> cache;
void cacheFile(std::string file, std::string root) {
std::cout << "Caching file: " << file << std::endl;
std::ifstream fin(root + file, std::ios::binary);
std::ostringstream oss;
oss << fin.rdbuf();
if (file == "/index.html") {
file = "/";
}
char *cachedFile = (char *) malloc(oss.str().size());
memcpy(cachedFile, oss.str().data(), oss.str().size());
char *key = (char *) malloc(file.length());
memcpy(key, file.data(), file.length());
std::cout << "Size: " << oss.str().size() << std::endl;
cache[std::string_view(key, file.length())] = std::string_view(cachedFile, oss.str().size());
}
public:
FileCache(std::string root) {
for(auto &p : std::experimental::filesystem::recursive_directory_iterator(root)) {
cacheFile(p.path().string().substr(root.length()), root);
}
}
std::string_view getFile(std::string_view file) {
auto it = cache.find(file);
if (it == cache.end()) {
//std::cout << "Did not find file: " << file << std::endl;
return "<h1>Nope!</h1>";
} else {
//std::cout << "Did find file: " << file << std::endl;
return it->second;
}
}
};
main.cpp
+42 -62
View File
@@ -1,59 +1,11 @@
#include "App.h"
#include <map>
#include <cstring>
#include <fstream>
#include <sstream>
#include <iostream>
#include "examples/helpers/AsyncFileReader.h"
// should probably fix this one up a bit some time
std::string_view getFile(std::string_view file) {
static std::map<std::string_view, std::string_view> cache;
auto it = cache.find(file);
if (it == cache.end()) {
std::cout << "Did not have file: " << file << std::endl;
std::ifstream fin(std::string(file), std::ios::binary);
std::ostringstream oss;
oss << fin.rdbuf();
char *cachedFile = (char *) malloc(oss.str().size());
memcpy(cachedFile, oss.str().data(), oss.str().size());
char *key = (char *) malloc(file.length());
memcpy(key, file.data(), file.length());
std::cout << "Size: " << oss.str().size() << std::endl;
cache[std::string_view(key, file.length())] = std::string_view(cachedFile, oss.str().size());
return getFile(file);
} else {
return it->second;
}
}
#include <set>
std::set<uWS::HttpResponse<false> *> delayedResponses;
AsyncFileReader asyncFileReader("/home/alexhultman/v0.15/sintel.mkv");
int main(int argc, char **argv) {
// create a timer that resumes sockets
auto *timer = us_create_timer((us_loop *) uWS::Loop::defaultLoop(), 1, 0);
us_timer_set(timer, [](auto *timer) {
for (auto *x : delayedResponses) {
std::cout << "Resuming a response now!" << std::endl;
x->resume();
}
delayedResponses.clear();
}, 1000, 1000);
uWS::/*SSL*/App(/*{
.key_file_name = "/home/alexhultman/uWebSockets/misc/ssl/key.pem",
.cert_file_name = "/home/alexhultman/uWebSockets/misc/ssl/cert.pem",
@@ -70,25 +22,53 @@ int main(int argc, char **argv) {
}
return uWS::HTTP_STREAM_FIN;
});
}).get("/delayed", [](auto *res, auto *req) {
}).get("/async/sintel.mkv", [](auto *res, auto *req) {
// res->write(asyncFileReader.stream(res))
// asyncFileReader.getAsStream()
/* This route streams back chunks of data in delayed fashion */
res->writeStatus(uWS::HTTP_200_OK)->write([res](int offset) {
/* Handle aborted stream out */
/* Handle broken stream */
if (offset == -1) {
std::cout << "Removing closed stream!" << std::endl;
delayedResponses.erase(res);
return uWS::HTTP_STREAM_FIN;
std::cout << "Stream was closed by peer!" << std::endl;
return uWS::HTTP_STREAM_IGNORE;
}
/* Delay stream out */
std::cout << "Delaying stream now" << std::endl;
delayedResponses.insert(res);
return uWS::HTTP_STREAM_PAUSE;
/* Peek from cache */
std::string_view chunk = asyncFileReader.peek(offset);
if (chunk.length()) {
/* We had parts of this file cached already */
return std::pair<bool, std::string_view>(false, chunk);
} else {
}, 100);
}).get("/:folder/:file", [](auto *res, auto *req) {
res->writeStatus(uWS::HTTP_200_OK)->write(getFile((req->getUrl() == "/" ? "/rocket_files/rocket.html" : req->getUrl()).substr(1)));
std::string_view outerChunk;
/* We had nothing readily available right now, request async chunk and pause the stream until we have */
asyncFileReader.request(offset, [&outerChunk, res](std::string_view chunk) {
/* We were aborted */
if (!chunk.length()) {
std::cout << "Async File Read request was aborted!" << std::endl;
// close the socket here?
// we need a way to NOT resume a paused socket! essentially close!
} else {
/* We finally got the data, resume stream with this chunk */
//res->resume(chunk);
outerChunk = chunk;
}
});
std::cout << "Returning chunk of size: " << outerChunk.length() << std::endl;
return std::pair<bool, std::string_view>(false, outerChunk);
// what if we resumed before we paused! we cannot do that!
std::cout << "PAusing stream out due to empty cache!" << std::endl;
return uWS::HTTP_STREAM_PAUSE;
}
}, asyncFileReader.getFileSize());
}).listen(3000, [](auto *token) {
if (token) {
std::cout << "Listening on port " << 3000 << std::endl;
src/HttpContext.h
+13 -2
View File
@@ -135,6 +135,8 @@ private:
/* Handle HTTP write out */
static_dispatch(us_ssl_socket_context_on_writable, us_socket_context_on_writable)(getSocketContext(), [](auto *s) {
std::cout << "Writable event!" << std::endl;
/* Silence any spurious writable events due to SSL_read failing to write */
AsyncSocket<SSL> *asyncSocket = (AsyncSocket<SSL> *) s;
HttpResponseData<SSL> *httpResponseData = (HttpResponseData<SSL> *) asyncSocket->getExt();
@@ -162,8 +164,17 @@ private:
if (httpResponseData->outStream) {
/* Regular path, request more data */
auto [msg_more, chunk] = httpResponseData->outStream(httpResponseData->offset);
httpResponseData->offset += asyncSocket->mergeDrain(chunk);
// todo: share this path with HttpResponse::write (it is exatly the same logic!)
while (true) {
auto [msg_more, chunk] = httpResponseData->outStream(httpResponseData->offset);
int written = asyncSocket->mergeDrain(chunk);
httpResponseData->offset += written;
// this is not correct, we can reach the end!
if (written < chunk.length()) {
break;
}
}
// todo: we should loop until we cannot send anymore just like we do in HttpResponse::write(stream)!
} else {
src/HttpResponse.h
+3
View File
@@ -17,6 +17,9 @@ const std::pair<bool, std::string_view> HTTP_STREAM_PAUSE = {false, std::string_
/* Return this from a stream callback to signal FIN */
const std::pair<bool, std::string_view> HTTP_STREAM_FIN = {false, std::string_view((const char *) 1, 0)};
/* Nobody cares what value this one has */
const auto HTTP_STREAM_IGNORE = HTTP_STREAM_FIN;
template <bool SSL>
struct HttpResponse : public AsyncSocket<SSL> {
private: