Improve streaming in and out over quic

examples/Http3Server.cpp

@@ -3,27 +3,55 @@
 /* Do not rely on this API, it will change */
 #include "Http3App.h"
 #include <iostream>
+#include <fstream>
 
-/* Example of simple Http3 server handling both GET with headers and POST with body.
+/* This is an example serving a video over HTTP3, and echoing posted data back */
- * You might be surprised to find out you can replace uWS::H3App with uWS::SSLApp and
+/* Todo: use onWritable and tryEnd instead of end */
- * serve TCP/TLS-based HTTP instead of QUIC-based HTTP, using the same very code ;) */
 int main() {
+
+	/* Read video file to memory */
+	std::ifstream file("video.mp4", std::ios::binary | std::ios::ate);
+	std::streamsize size = file.tellg();
+	file.seekg(0, std::ios::beg);
+
+	std::vector<char> buffer(size);
+	if (!file.read(buffer.data(), size)) {
+		std::cout << "Failed to load video.mp4" << std::endl;
+		return 0;
+	}
+
+	/* We need a bootstrapping server that instructs
+	 * the web browser to use HTTP3 */
+	(*new uWS::SSLApp({
+	  .key_file_name = "misc/key.pem",
+	  .cert_file_name = "misc/cert.pem",
+	  .passphrase = "1234"
+	})).get("/video.mp4", [&buffer](auto *res, auto *req) {
+		res->writeHeader("Alt-Svc", "h3=\":9004\"");
+		res->writeHeader("Alternative-Protocol", "quic:9004");
+	    res->end({"This is not HTTP3!", 18});
+	}).listen(9004, [](auto *listen_socket) {
+	    if (listen_socket) {
+			std::cout << "Bootstrapping server Listening on port " << 9004 << std::endl;
+	    }
+	});
+
+	/* And we serve the video over HTTP3 */
 	uWS::H3App({
 	  .key_file_name = "misc/key.pem",
 	  .cert_file_name = "misc/cert.pem",
 	  .passphrase = "1234"
-	}).get("/*", [](auto *res, auto *req) {
+	}).get("/video.mp4", [&buffer](auto *res, auto *req) {
+		/* Send back a video */
+	    res->end({&buffer[0], buffer.size()});
+	}).post("/*", [](auto *res, auto *req) {
 
-		/* Printing these should obviously be disabled if doing benchmarking */
+		std::cout << "Got POST request at " << req->getHeader(":path") << std::endl;
-		std::cout << req->getHeader(":path") << std::endl;
-		std::cout << req->getHeader(":method") << std::endl;
-
-	    res->end("Hello H3 from uWS!");
-	}).post("/*", [](auto *res, auto */*req*/) {
 
 		/* You also need to set onAborted if receiving data */
 		res->onData([res, bodyBuffer = (std::string *)nullptr](std::string_view chunk, bool isLast) mutable {
 			if (isLast) {
+				std::cout << "Sending back posted body now" << std::endl;
 				if (bodyBuffer) {
 					/* Send back the (chunked) body we got, as response */
 					bodyBuffer->append(chunk);
@@ -52,11 +80,11 @@ int main() {
 		});
 	}).listen(9004, [](auto *listen_socket) {
 	    if (listen_socket) {
-			std::cout << "Listening on port " << 9004 << std::endl;
+			std::cout << "HTTP/3 server Listening on port " << 9004 << std::endl;
 	    }
 	}).run();
 
-	std::cout << "Failed to listen on port 3000" << std::endl;
+	std::cout << "Failed to listen on port 9004" << std::endl;
 }
 
 #else

src/Http3Context.h

@@ -9,60 +9,63 @@ extern "C" {
 namespace uWS {
     struct Http3Context {
         static Http3Context *create(us_loop_t *loop, us_quic_socket_context_options_t options) {
-            //return nullptr;
-
-            printf("Creating context now\n");
 
             /* Create quic socket context (assumes h3 for now) */
-            auto *context = us_create_quic_socket_context(loop, options, sizeof(Http3ContextData)); // sizeof(Http3ContextData)
+            auto *context = us_create_quic_socket_context(loop, options, sizeof(Http3ContextData));
 
             /* Specify application callbacks */
             us_quic_socket_context_on_stream_data(context, [](us_quic_stream_t *s, char *data, int length) {
 
-                // we don't have a way to know EOF?
                 Http3ResponseData *responseData = (Http3ResponseData *) us_quic_stream_ext(s);
                 
+                /* We never emit FIN here */
                 if (responseData->onData) {
-                    responseData->onData({data, length}, true);
+                    responseData->onData({data, (size_t) length}, false);
+                }
+            });
+            us_quic_socket_context_on_stream_end(context, [](us_quic_stream_t *s) {
+
+                Http3ResponseData *responseData = (Http3ResponseData *) us_quic_stream_ext(s);
+                
+                /* Emit FIN to app */
+                if (responseData->onData) {
+                    responseData->onData({nullptr, 0}, true);
+                }
+
+                /* Have we written our entire backpressure, if any? */
+                if (responseData->buffer.length() && (responseData->bufferOffset == (int) responseData->buffer.length())) {
+                    printf("We got FIN and we have no backpressure, closing stream now!\n");
+                    us_quic_stream_close(s);
+                } else {
+                    printf("We got FIN but we have data to write, so keeping connection half-closed!\n");
                 }
 
-                //printf("Body length is: %d\n", length);
             });
             us_quic_socket_context_on_stream_open(context, [](us_quic_stream_t *s, int is_client) {
-                //printf("Stream opened!\n");
-
-                // inplace initialize Http3ResponseData here
 
+                /* Inplace init our per stream data */
                 new (us_quic_stream_ext(s)) Http3ResponseData();
             });
             us_quic_socket_context_on_close(context, [](us_quic_socket_t *s) {
-                printf("Disconnected from on_close in uws!\n");
+                printf("QUIC socket disconnected!\n");
-
             });
             us_quic_socket_context_on_stream_writable(context, [](us_quic_stream_t *s) {
                 Http3ResponseData *responseData = (Http3ResponseData *) us_quic_stream_ext(s);
-                // responseData->onWritable();
 
-                int written = us_quic_stream_write(s, responseData->buffer.data() + responseData->bufferOffset, responseData->buffer.length() - responseData->bufferOffset);
+                int written = us_quic_stream_write(s, responseData->buffer.data() + responseData->bufferOffset, (int) responseData->buffer.length() - responseData->bufferOffset);
+                responseData->bufferOffset += written;
 
-                //printf("wrote %d bytes in writable callback\n", written);
+                //printf("remaingin bytes: %ld\n", responseData->buffer.length() - responseData->bufferOffset);
 
-                // this whole thing should use the BackpressureBuffer class
+                if ((int) responseData->buffer.length() - responseData->bufferOffset == 0) {
-                responseData->bufferOffset += written;//responseData->buffer.substr(written);
-
-                printf("remaingin bytes: %ld\n", responseData->buffer.length() - responseData->bufferOffset);
-
-                if (responseData->buffer.length() - responseData->bufferOffset == 0) {
                     printf("wrote until end, shutting down now!\n");
                     us_quic_stream_shutdown(s);
                     us_quic_stream_close(s);
                 }
-
-                //printf("stream is now writable!\n");
-                
             });
             us_quic_socket_context_on_stream_headers(context, [](us_quic_stream_t *s) {
 
+                /* This is the main place of start for requests */
                 Http3ContextData *contextData = (Http3ContextData *) us_quic_socket_context_ext(us_quic_socket_context(us_quic_stream_socket(s)));
 
                 Http3Request *req = nullptr;
@@ -75,7 +78,7 @@ namespace uWS {
 
             });
             us_quic_socket_context_on_open(context, [](us_quic_socket_t *s, int is_client) {
-                printf("Connection established!\n");
+                printf("QUIC socket connected!\n");
             });
             us_quic_socket_context_on_stream_close(context, [](us_quic_stream_t *s) {
 
@@ -85,6 +88,8 @@ namespace uWS {
                     responseData->onAborted();
                 }
 
+                printf("Freeing per stream data in on_stream_close in uws!\n");
+
                 responseData->~Http3ResponseData();
             });
 
@@ -97,7 +102,7 @@ namespace uWS {
             /* The listening socket is the actual UDP socket used */
             us_quic_listen_socket_t *listen_socket = us_quic_socket_context_listen((us_quic_socket_context_t *) this, "::", 9004, sizeof(Http3ResponseData));
 
-            printf("Listen socket is: %p\n", listen_socket);
+            //printf("Listen socket is: %p\n", listen_socket);
 
             return listen_socket;
         }
@@ -109,7 +114,7 @@ namespace uWS {
 
             Http3ContextData *contextData = (Http3ContextData *) us_quic_socket_context_ext((us_quic_socket_context_t *) this);
 
-            printf("init: %p\n", contextData);
+            //printf("init: %p\n", contextData);
 
             new (contextData) Http3ContextData();
 

src/Http3ContextData.h

@@ -14,7 +14,7 @@ namespace uWS {
         HttpRouter<RouterData> router;
 
         Http3ContextData() {
-            printf("Constructing http3contextdata: %p\n", this);
+            //printf("Constructing http3contextdata: %p\n", this);
         }
     };
     

src/Http3Request.h

@@ -10,8 +10,8 @@ namespace uWS {
             for (int i = 0, more = 1; more; i++) {
                 char *name, *value;
                 int name_length, value_length;
-                if (more = us_quic_socket_context_get_header(nullptr, i, &name, &name_length, &value, &value_length)) {
+                if ((more = us_quic_socket_context_get_header(nullptr, i, &name, &name_length, &value, &value_length))) {
-                    if (name_length == key.length() && !memcmp(name, key.data(), key.length())) {
+                    if (name_length == (int) key.length() && !memcmp(name, key.data(), key.length())) {
                         return {value, (size_t) value_length};
                     }
                 }

src/Http3Response.h

@@ -38,12 +38,12 @@ namespace uWS {
             us_quic_socket_context_send_headers(nullptr, (us_quic_stream_t *) this, 1, 1);
 
             /* Write body and shutdown (unknown if content-length must be present?) */
-            int written = us_quic_stream_write((us_quic_stream_t *) this, (char *) data.data(), data.length());
+            int written = us_quic_stream_write((us_quic_stream_t *) this, (char *) data.data(), (int) data.length());
 
             printf("Wrote %d bytes out of %ld\n", written, data.length());
 
             /* Buffer up remains */
-            if (written != data.length()) {
+            if (written != (int) data.length()) {
                 responseData->buffer.clear();
                 responseData->bufferOffset = written;
                 responseData->buffer.append(data);