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

Begin work on new streams, chunked response streams, etc

Browse Source
This commit is contained in:
Alex Hultman
2018-09-26 20:42:26 +02:00
parent 18367b58fd
commit c25a2785e5
2 changed files with 167 additions and 120 deletions
Download Patch File Download Diff File Expand all files Collapse all files
main.cpp
+86 -43
View File
@@ -12,64 +12,107 @@ int main(int argc, char **argv) {
.dh_params_file_name = "/home/alexhultman/dhparams.pem",
.passphrase = "1234"
}*/).get("/", [](auto *res, auto *req) {
res->writeStatus(uWS::HTTP_200_OK)->write("Hello world!");
}).get("/endless", [](auto *res, auto *req) {
/* This route doesn't specify any length and only returns 1 char per chunk */
res->write([](int offset) {
std::string_view data("<html><body><h1>Hello, world</h1></body></html>");
if (offset < data.length()) {
return std::make_pair<bool, std::string_view>(offset < data.length() - 1, data.substr(offset, 1));
}
return uWS::HTTP_STREAM_FIN;
});
res->writeStatus(uWS::HTTP_200_OK);
res->writeHeader("Content-Type", "text/html;charset=utf-8");
res->write("<h1>Hallå!</h1>Din user-agent är: ");
res->end(req->getHeader("user-agent"));
// }).get("/endless", [](auto *res, auto *req) {
// /* This route doesn't specify any length and only returns 1 char per chunk */
// res->write([](int offset) {
// std::string_view data("<html><body><h1>Hello, world</h1></body></html>");
// if (offset < data.length()) {
// return std::make_pair<bool, std::string_view>(offset < data.length() - 1, data.substr(offset, 1));
// }
// return uWS::HTTP_STREAM_FIN;
// });
}).get("/async/sintel.mkv", [](auto *res, auto *req) {
// res->write(asyncFileReader.stream(res))
// asyncFileReader.getAsStream()
// res->write(data, length) -> bool (identical to nodejs)
// res->tryWrite(data, length) -> int (what we will use mostly)
// both operate like uSockets, they are paused by default and will trigger onWritable if set
// res->onWritable(lambda) -> the same as "drain" even in Node.js, call tryWrite here, or do nothing to hang/pause
// resume is gone, it will resume automatically as needed, just like uSockets do automatically
// res will be somewhat like AsyncSocket but with certain middle functions for formatting the data (think Content-Length and Chunked)
// res->onAborted(lambda) will be called when closed, for now all events lie on the res
// res->onData() will probably be the same?
//res->writeStatus(uWS::HTTP_200_OK)->end("Hello Sintel!");
// om vi håller headers i en buffer och data i en annan?
// om end eller tryEnd inte kommer före write swappar den till chunked!
res->write("Hallå! Din user-agent är: "); // -> skriver till cork-buffern "Transfer-Encoding: chunked" och ett chunk
res->end(req->getHeader("user-agent"));
// write före end går över till chunked, från där kan man extenda till SSE men websockets är bättre format (binärt, mindre)
// hur fungerar detta med server-sent-events?
// end efter write blir ett noll-segment i chunked!
/*if (!res->tryEnd()) {
}*/
/* This route streams back chunks of data in delayed fashion */
res->writeStatus(uWS::HTTP_200_OK)->write([res](int offset) {
// res->writeStatus(uWS::HTTP_200_OK)->write([res](int offset) {
/* Handle broken stream */
if (offset == -1) {
std::cout << "Stream was closed by peer!" << std::endl;
return uWS::HTTP_STREAM_IGNORE;
}
// /* Handle broken stream */
// if (offset == -1) {
// std::cout << "Stream was closed by peer!" << std::endl;
// return uWS::HTTP_STREAM_IGNORE;
// }
/* 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 {
// /* 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 {
//std::string_view outerChunk;
// //std::string_view outerChunk;
/* We had nothing readily available right now, request async chunk and pause the stream until we have */
asyncFileReader.request(offset, [res](std::string_view chunk) {
// /* We had nothing readily available right now, request async chunk and pause the stream until we have */
// asyncFileReader.request(offset, [res](std::string_view chunk) {
//std::cout << "We came here!" << std::endl;
// //std::cout << "We came here!" << std::endl;
/* 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);
}
});
// /* 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);
// }
// });
//std::cout << "Returning chunk of size: " << outerChunk.length() << std::endl;
//return std::pair<bool, std::string_view>(false, outerChunk);
// //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!
// // 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());
// //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/HttpResponse.h
+81 -77
View File
@@ -11,15 +11,6 @@ namespace uWS {
/* Some pre-defined status constants to use with writeStatus */
const char *HTTP_200_OK = "200 OK";
/* Return this from a stream callback to signal pause */
const std::pair<bool, std::string_view> HTTP_STREAM_PAUSE = {false, std::string_view(nullptr, 0)};
/* 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:
@@ -27,6 +18,33 @@ private:
return (HttpResponseData<SSL> *) AsyncSocket<SSL>::getExt();
}
int u32toaHex(uint32_t value, char *dst) {
char palette[] = "0123456789abcdef";
char temp[10];
char *p = temp;
do {
*p++ = palette[value % 16];
value /= 16;
} while (value > 0);
int ret = p - temp;
do {
*dst++ = *--p;
} while (p != temp);
return ret;
}
/* Write an unsigned 32-bit integer in hex */
void writeUnsignedHex(unsigned int value) {
char buf[10];
int length = u32toaHex(value, buf);
/* For now we do this copy */
AsyncSocket<SSL>::write(buf, length);
}
int u32toa(uint32_t value, char *dst) {
char temp[10];
char *p = temp;
@@ -90,58 +108,66 @@ public:
return this;
}
/* Resume response streaming as far as possible */
void resume(std::string_view chunk = {}) {
/* End the response with an optional data chunk */
void end(std::string_view data = {}) {
HttpResponseData<SSL> *httpResponseData = getHttpResponseData();
/* Do nothing if not even paused */
if (!(httpResponseData->state & HttpResponseData<SSL>::HTTP_PAUSED_STREAM_OUT)) {
std::cout << "Resue called but we are not even in paused state!" << std::endl;
return;
if (httpResponseData->state & HttpResponseData<SSL>::HTTP_WRITE_CALLED) {
/* Do not allow sending 0 chunk here */
if (data.length()) {
AsyncSocket<SSL>::write("\r\n", 2);
writeUnsignedHex(data.length());
AsyncSocket<SSL>::write("\r\n", 2);
AsyncSocket<SSL>::write(data.data(), data.length());
}
/* Terminating 0 chunk */
AsyncSocket<SSL>::write("\r\n0\r\n\r\n", 7);
} else {
/* We have a known send size */
AsyncSocket<SSL>::write("Content-Length: ", 16);
writeUnsigned(data.length());
AsyncSocket<SSL>::write("\r\n\r\n", 4);
AsyncSocket<SSL>::write(data.data(), data.length());
}
//std::cout << "Resume called and we really are paused" << std::endl;
/* Remove paused status */
httpResponseData->state &= ~HttpResponseData<SSL>::HTTP_PAUSED_STREAM_OUT;
/*if (chunk.length()) {
int written = AsyncSocket<SSL>::write(chunk.data(), chunk.length(), true);
if (written == chunk.length()) {
// pull a new chunk from the callback (basically call onWritable)
std::cout << "Wrote everything off!" << std::endl;
}
}*/
AsyncSocket<SSL> *asyncSocket = this;
// again, this path is shared with onwritable, write and here!
while (true) {
auto [msg_more, chunk] = httpResponseData->outStream(httpResponseData->offset);
// break on pause!
if (chunk.length() == 0) {
//std::cout << "Resume paused!" << std::endl;
httpResponseData->state |= HttpResponseData<SSL>::HTTP_PAUSED_STREAM_OUT;
break;
}
int written = asyncSocket->mergeDrain(chunk);
httpResponseData->offset += written;
// this is not correct, we can reach the end!
if (written < chunk.length()) {
break;
}
}
// no, basically just write this off and if all written, call streamOut callback
}
/* Write parts of the response in chunking fashion */
bool write(std::string_view data) {
/* Do not allow sending 0 chunks, they mark end of response */
if (!data.length()) {
return true; // are we corked still?
}
HttpResponseData<SSL> *httpResponseData = getHttpResponseData();
if (!(httpResponseData->state & HttpResponseData<SSL>::HTTP_WRITE_CALLED)) {
writeHeader("Transfer-Encoding", "chunked");
httpResponseData->state |= HttpResponseData<SSL>::HTTP_WRITE_CALLED;
}
AsyncSocket<SSL>::write("\r\n", 2);
writeUnsignedHex(data.length());
AsyncSocket<SSL>::write("\r\n", 2);
AsyncSocket<SSL>::write(data.data(), data.length());
// are we corked still?
return true;
}
// we really want tryEnd(data) integer to try and stream something with known size
// write/tryWrite called first should enter into chunked?
// tryWrite(char *, length) int
// write(char *, length) bool
/* Attach an output stream function. Chunks may be read more than once. Negative offset mean broken stream */
void write(std::function<std::pair<bool, std::string_view>(int)> cb, int length = 0) {
void writeOldRemoveMe(std::function<std::pair<bool, std::string_view>(int)> cb, int length = 0) {
HttpResponseData<SSL> *httpResponseData = getHttpResponseData();
/* Do not allow write if already called */
@@ -209,28 +235,6 @@ public:
}
}
/* Convenience function for static data */
void write(std::string_view data, std::function<void(std::string_view)> cb = nullptr) {
if (cb) {
// todo: think about how the stream will signal done (streams API challenge overall)
write([data](int offset) {
// if offset == length then we know it is end
// what if we want to return both fin and data? can't do that
return std::make_pair<bool, std::string_view>(false, data.substr(offset));//{false, data.substr(offset)};
}, data.length());
} else {
// requires no extra alloc
write([data](int offset) {
return std::make_pair<bool, std::string_view>(false, data.substr(offset));//{false, data.substr(offset)};
}, data.length());
}
}
/* Attach a read handler for data sent. Will be called with a chunk of size 0 when FIN */
void read(std::function<void(std::string_view)> handler) {
HttpResponseData<SSL> *data = getHttpResponseData();