Basic README

Makefile

@@ -5,7 +5,7 @@ CXX_INCLUDES=-I$(HTTP_PARSER) -I$(DART_SDK)/include
 .PHONY: clean debug macos all
 
 all:
-	printf 'Available targets:\n'\
+	//printf 'Available targets:\n'\
 	'  * `debug` - Builds a debug library on MacOS\n'\
 	'  * `example` - Runs example/main.dart in LLDB on MacOS\n'\
 	'  * `macos` - Builds a release-mode library on MacOS\n'

README.md

@@ -0,0 +1,51 @@
+# wings
+Native HTTP driver for Angel, for a nice speed boost.
+
+Not ready for production.
+
+## How does it work?
+Typically, Angel uses the `AngelHttp` driver, which is wrapper over the `HttpServer` functionality in
+`dart:io`, which in turns uses the `ServerSocket` and `Socket` functionality. This is great - Dart's standard
+library comes with an HTTP server, which saves a lot of difficult in implementation.
+
+However, abstraction tends to come with a cost. Wings seeks to minimize abstraction entirely. Rather than
+using the built-in Dart network stack, Wings' HTTP server is implemented in C++ as a Dart native extension,
+and the `AngelWings` driver listens to events from the extension and converts them directly into
+`RequestContext` objects, without any additional abstraction within. This reduces the amount of computation
+performed on each request, and helps to minimize response latency. Sending data from the response buffer in plain
+Dart surprisingly is the most expensive operation, as is revealed by the Observatory.
+
+By combining Dart's powerful VM with a native code server based on
+[the same one used in Node.js](https://github.com/nodejs/http-parser),
+`AngelWings` trims several milliseconds off every request, both saving resources and reducing
+load times for high-traffic applications.
+
+## How can I use it?
+The intended way to use `AngelWings` is via
+[`package:build_native`](https://github.com/thosakwe/build_native);
+however, the situation surrounding distributing native extensions is yet far from ideal,
+so this package includes pre-built binaries out-of-the-box.
+
+Thanks to this, you can use it like any other Dart package, by installing it via Pub.
+
+## Brief example
+Using `AngelWings` is almost identical to using `AngelHttp`; however, it does
+not support SSL, and therefore should be placed behind a reverse proxy like `nginx` in production.
+
+```dart
+main() async {
+  var app = new Angel();
+  var wings = new AngelWings(app, shared: true, useZone: false);
+
+  app.injectEncoders({'gzip': gzip.encoder, 'deflate': zlib.encoder});
+
+  app.get('/hello', 'Hello, native world! This is Angel WINGS.');
+
+  var fs = const LocalFileSystem();
+  var vDir = new VirtualDirectory(app, fs, source: fs.directory('web'));
+  app.use(vDir.handleRequest);
+
+  await wings.startServer('127.0.0.1', 3000);
+  print('Listening at http://${wings.address.address}:${wings.port}');
+}
+```

example/main.dart

@@ -2,7 +2,9 @@ import 'dart:async';
 import 'dart:io';
 import 'dart:isolate';
 import 'package:angel_framework/angel_framework.dart';
+import 'package:angel_static/angel_static.dart';
 import 'package:angel_wings/angel_wings.dart';
+import 'package:file/local.dart';
 
 main() async {
   for (int i = 1; i < Platform.numberOfProcessors; i++) {
@@ -19,6 +21,8 @@ void isolateMain(int id) {
   var app = new Angel();
   var wings = new AngelWings(app, shared: true, useZone: false);
 
+  app.injectEncoders({'gzip': gzip.encoder, 'deflate': zlib.encoder});
+
   var old = app.errorHandler;
   app.errorHandler = (e, req, res) {
     print(e);
@@ -26,7 +30,11 @@ void isolateMain(int id) {
     return old(e, req, res);
   };
 
-  app.get('/', 'Hello, native world! This is isolate #$id.');
+  app.get('/hello', 'Hello, native world! This is isolate #$id.');
+
+  var fs = const LocalFileSystem();
+  var vDir = new VirtualDirectory(app, fs, source: fs.directory('web'));
+  app.use(vDir.handleRequest);
 
   wings.startServer('127.0.0.1', 3000).then((_) {
     print(

lib/angel_wings.dart

@@ -13,7 +13,7 @@ import 'package:http_parser/http_parser.dart';
 import 'package:json_god/json_god.dart' as god;
 import 'package:mock_request/mock_request.dart';
 import 'package:pool/pool.dart';
-//import 'package:pooled_map/pooled_map.dart';
+import 'package:pooled_map/pooled_map.dart';
 import 'package:stack_trace/stack_trace.dart';
 import 'package:tuple/tuple.dart';
 import 'package:uuid/uuid.dart';

lib/src/bind_socket.cc

@@ -37,8 +37,8 @@ void wings_BindSocket(Dart_NativeArguments arguments)
 
     if (shared)
     {
-        #if __APPLE__
+        //#if __APPLE__
-        #else
+        //#else
         for (unsigned long i = 0; i < serverInfoVector.size(); i++)
         {
             WingsServerInfo *server_info = serverInfoVector.at(i);
@@ -49,7 +49,7 @@ void wings_BindSocket(Dart_NativeArguments arguments)
                 break;
             }
         }
-        #endif
+        //#endif
     }
 
     if (existingIndex > -1)
@@ -117,6 +117,7 @@ void wings_BindSocket(Dart_NativeArguments arguments)
             return;
         }
 
+/*
 #if __APPLE__
         ret = setsockopt(sockfd, SOL_SOCKET, SO_REUSEPORT, &i, sizeof(i));
 
@@ -126,6 +127,7 @@ void wings_BindSocket(Dart_NativeArguments arguments)
             return;
         }
 #endif
+*/
 
         if (addressLength > 4)
         {

lib/src/dart_debug.h

@@ -11,17 +11,17 @@ Dart_Handle ToCString(Dart_Handle obj, const char** out) {
   return Dart_StringToCString(string, out);
 }
 
-Dart_Handle Dart_PrintToFile(Dart_Handle obj, FILE* stream) {
+Dart_Handle Dart_//printToFile(Dart_Handle obj, FILE* stream) {
   const char *toString;
   Dart_Handle result = ToCString(obj, &toString);
 
   if (Dart_IsError(result))
     return result;
  
-  fprintf(stream, "%s\n", toString);
+  f//printf(stream, "%s\n", toString);
   return Dart_Null();
 }
 
-Dart_Handle Dart_Print(Dart_Handle obj) {
+Dart_Handle Dart_//print(Dart_Handle obj) {
-  return Dart_PrintToFile(obj, stdout);
+  return Dart_//printToFile(obj, stdout);
 }

lib/src/http_listener.cc

@@ -1,6 +1,6 @@
 #include <dart_native_api.h>
 #include <thread>
-//#include <iostream>
+#include <iostream>
 #include <cstdio>
 #include "wings.h"
 #include "wings_thread.h"

lib/src/wings.cc

@@ -1,5 +1,5 @@
 #include <cstdlib>
-//#include <iostream>
+#include <iostream>
 #include <string.h>
 #include <dart_api.h>
 #include "wings.h"

lib/src/wings.dart

@@ -76,9 +76,11 @@ class AngelWings {
 
   final RawReceivePort _recv = new RawReceivePort();
   final Map<String, MockHttpSession> _sessions = {};
-  final Map<int, WingsRequestContext> _staging = <int, WingsRequestContext>{};
+  //final Map<int, WingsRequestContext> _staging = <int, WingsRequestContext>{};
-  //final PooledMap<int, WingsRequestContext> _staging =
+  final PooledMap<int, WingsRequestContext> _staging =
-  //    new PooledMap<int, WingsRequestContext>();
+      new PooledMap<int, WingsRequestContext>();
+  final PooledMap<int, WingsRequestContext> _live =
+      new PooledMap<int, WingsRequestContext>();
   final Uuid _uuid = new Uuid();
   InternetAddress _address;
   int _port;
@@ -91,30 +93,42 @@ class AngelWings {
 
   //final Pool _pool = new Pool(1);
 
-  static void _send(int sockfd, Uint8List data) native "Send";
+  static void __send(int sockfd, Uint8List data) native "Send";
 
-  static void _closeSocket(int sockfd) native "CloseSocket";
+  static void __closeSocket(int sockfd) native "CloseSocket";
 
-  /*
+  void _send(WingsRequestContext req, Uint8List data) {
-  void _send(int sockfd, Uint8List data) {
+    _live.update(req._sockfd, (_) {
+      //print('Sending ${[req._sockfd, data]}');
+      __send(req._sockfd, data);
+      return req;
+    });
     // _pool.withResource(() {
-    print('Sending ${[sockfd, data]}');
+    ////print('Sending ${[sockfd, data]}');
-    _sendPort.send([sockfd, data]);
+    //_sendPort.send([sockfd, data]);
     //});
     //_pool.withResource(() => __send(sockfd, data));
   }
 
   void _closeSocket(WingsRequestContext req) {
+    _live.remove(req._sockfd).then((_) {
+      if (!req._closed) {
+        req._closed = true;
+        //print('Closing ${[req._sockfd]}');
+        __closeSocket(req._sockfd);
+      }
+      return req;
+    });
     //_pool.withResource(() {
-    if (!req._closed) {
+    //if (!req._closed) {
-      req._closed = true;
+    //  req._closed = true;
-      var sockfd = req._sockfd;
+    //  var sockfd = req._sockfd;
-      print('Sending ${[sockfd]}');
+    //  //print('Sending ${[sockfd]}');
-      _sendPort.send([sockfd]);
+    //  _sendPort.send([sockfd]);
-    }
+    //}
     //});
     //_pool.withResource(() => __closeSocket(sockfd));
-  }*/
+  }
 
   AngelWings(this.app, {this.shared: false, this.useZone: true}) {
     _recv.handler = _handleMessage;
@@ -172,16 +186,25 @@ class AngelWings {
     } else if (x is List && x.length >= 2) {
       int sockfd = x[0], command = x[1];
 
-      //WingsRequestContext _newRequest() =>
+      WingsRequestContext _newRequest() =>
-      //    new WingsRequestContext._(this, sockfd, app);
+          new WingsRequestContext._(this, sockfd, app);
       //print(x);
 
       switch (command) {
         case messageBegin:
           //print('BEGIN $sockfd');
-          _staging[sockfd] = new WingsRequestContext._(this, sockfd, app);
+          _staging.putIfAbsent(sockfd, _newRequest);
+          //_staging[sockfd] = new WingsRequestContext._(this, sockfd, app);
           break;
         case messageComplete:
+          _staging.remove(sockfd).then((rq) {
+            if (rq != null) {
+              rq._method = methodToString(x[2] as int);
+              rq._addressBytes = x[5] as Uint8List;
+              _live.put(sockfd, rq).then((_) => _handleRequest(rq));
+            }
+          });
+          /*
           //print('$sockfd in $_staging???');
           var rq = _staging.remove(sockfd);
           if (rq != null) {
@@ -189,13 +212,21 @@ class AngelWings {
             rq._addressBytes = x[5] as Uint8List;
             _handleRequest(rq);
           }
+          */
           break;
         case body:
+          _staging.update(sockfd, (rq) {
+            (rq._body ??= new StreamController<Uint8List>())
+                .add(x[2] as Uint8List);
+            return rq;
+          }, defaultValue: _newRequest);
+          /*
           var rq = _staging[sockfd];
           if (rq != null) {
             (rq._body ??= new StreamController<Uint8List>())
                 .add(x[2] as Uint8List);
           }
+          */
           break;
         //case upgrade:
         // TODO: Handle WebSockets...?
@@ -206,13 +237,25 @@ class AngelWings {
         //  onUpgradedMessage(sockfd, x[2]);
         //  break;
         case url:
-          _staging[sockfd]?.__url = x[2] as String;
+          _staging.update(sockfd, (rq) {
+            rq?.__url = x[2] as String;
+            return rq;
+          }, defaultValue: _newRequest);
+          //_staging[sockfd]?.__url = x[2] as String;
           break;
         case headerField:
-          _staging[sockfd]?._headerField = x[2] as String;
+          _staging.update(sockfd, (rq) {
+            rq?._headerField = x[2] as String;
+            return rq;
+          }, defaultValue: _newRequest);
+          //_staging[sockfd]?._headerField = x[2] as String;
           break;
         case headerValue:
-          _staging[sockfd]?._headerValue = x[2] as String;
+          _staging.update(sockfd, (rq) {
+            rq?._headerValue = x[2] as String;
+            return rq;
+          }, defaultValue: _newRequest);
+          //_staging[sockfd]?._headerValue = x[2] as String;
           break;
       }
     }
@@ -317,7 +360,7 @@ class AngelWings {
             return handleAngelHttpException(e, trace, req, res);
           }).catchError((e, StackTrace st) {
             var trace = new Trace.from(st ?? StackTrace.current).terse;
-            AngelWings._closeSocket(req._sockfd);
+            _closeSocket(req);
             // Ideally, we won't be in a position where an absolutely fatal error occurs,
             // but if so, we'll need to log it.
             if (app.logger != null) {
@@ -354,8 +397,8 @@ class AngelWings {
         b.writeln('HTTP/1.1 500 Internal Server Error');
         b.writeln();
 
-        _send(req._sockfd, _coerceUint8List(b.toString().codeUnits));
+        _send(req, _coerceUint8List(b.toString().codeUnits));
-        AngelWings._closeSocket(req._sockfd);
+        _closeSocket(req);
       } finally {
         return null;
       }
@@ -469,9 +512,9 @@ class AngelWings {
 
     return finalizers.then((_) {
       //print('A');
-      _send(req._sockfd, buf);
+      _send(req, buf);
       //print('B');
-      AngelWings._closeSocket(req._sockfd);
+      _closeSocket(req);
       //print('C');
 
       if (req.injections.containsKey(PoolResource)) {

lib/src/wings_response.dart

@@ -13,14 +13,14 @@ class WingsResponseContext extends ResponseContext {
     if (_isClosed && !_useStream)
       throw ResponseContext.closed();
     else if (_useStream)
-      AngelWings._send(correspondingRequest._sockfd, _coerceUint8List(data));
+      _wings._send(correspondingRequest, _coerceUint8List(data));
     else
       buffer.add(data);
   }
 
   @override
   Future close() {
-    AngelWings._closeSocket(correspondingRequest._sockfd);
+    _wings._closeSocket(correspondingRequest);
     _isClosed = true;
     _useStream = false;
     return super.close();
@@ -76,8 +76,8 @@ class WingsResponseContext extends ResponseContext {
       }
     }
 
-    return output.forEach(((data) =>
+    return output.forEach(
-        AngelWings._send(correspondingRequest._sockfd, _coerceUint8List(data))));
+        ((data) => _wings._send(correspondingRequest, _coerceUint8List(data))));
   }
 
   @override
@@ -147,8 +147,8 @@ class WingsResponseContext extends ResponseContext {
 
     b.writeln();
 
-    AngelWings._send(
+    _wings._send(
-        correspondingRequest._sockfd, _coerceUint8List(b.toString().codeUnits));
+        correspondingRequest, _coerceUint8List(b.toString().codeUnits));
   }
 }
 

lib/src/worker_thread.cc

@@ -1,5 +1,5 @@
-//#include <memory>
+#include <memory>
-//#include <iostream>
+#include <iostream>
 #include <utility>
 #include <dart_native_api.h>
 #include "wings_thread.h"
@@ -12,31 +12,32 @@ void wingsThreadMain(wings_thread_info *info)
 
     while (true)
     {
-        std::unique_lock<std::mutex> lock(serverInfo->mutex, std::defer_lock);
+        //std::unique_lock<std::mutex> lock(serverInfo->mutex, std::defer_lock);
+        std::lock_guard<std::mutex> lock(serverInfo->mutex);
 
         sockaddr client_addr{};
         socklen_t client_addr_len;
 
-        if (lock.try_lock())
+        //if (lock.try_lock())
+        //{
+        int client = accept(serverInfo->sockfd, &client_addr, &client_addr_len);
+        //lock.unlock();
+
+        if (client < 0)
         {
-            int client = accept(serverInfo->sockfd, &client_addr, &client_addr_len);
+            // send_error(info->port, "Failed to accept client socket.");
-            lock.unlock();
+            return;
-
-            if (client < 0)
-            {
-                // send_error(info->port, "Failed to accept client socket.");
-                return;
-            }
-
-            //auto rq = std::make_shared<requestInfo>();
-            auto *rq = new requestInfo;
-            rq->ipv6 = serverInfo->ipv6;
-            rq->sock = client;
-            rq->addr = client_addr;
-            rq->addr_len = client_addr_len;
-            rq->port = port;
-            handleRequest(rq);
         }
+
+        //auto rq = std::make_shared<requestInfo>();
+        auto *rq = new requestInfo;
+        rq->ipv6 = serverInfo->ipv6;
+        rq->sock = client;
+        rq->addr = client_addr;
+        rq->addr_len = client_addr_len;
+        rq->port = port;
+        handleRequest(rq);
+        //}
     }
 }
 
@@ -176,7 +177,7 @@ void handleRequest(requestInfo *rq)
     };
 
     settings.on_url = [](http_parser *parser, const char *at, size_t length) {
-        // std::cout << "url" << std::endl;
+        // //std::cout << "url" << std::endl;
         return send_string(parser, (char *)at, length, 2);
     };
 
@@ -198,7 +199,7 @@ void handleRequest(requestInfo *rq)
     unsigned int isUpgrade = 0;
 
     //std::cout << "start" << std::endl;
-    while ((recved = recv(rq->sock, buf, len, 0)) > 0)
+    while ((recved = recv(rq->sock, buf, len, 0)) >= 0)
     {
         if (isUpgrade)
         {

pubspec.yaml

@@ -3,9 +3,10 @@ dependencies:
   angel_framework: ^1.0.0
   build_native: ^0.0.9
   mock_request: ^1.0.0
-  #pooled_map: ^1.0.0
+  pooled_map: ^1.0.0
   uuid: ^1.0.0
 dev_dependencies:
+  angel_static: ^1.3.0
   build_runner:
     git:
       url: https://github.com/thosakwe/build.git

web/index.html

@@ -0,0 +1,13 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+    <meta charset="UTF-8">
+    <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <meta http-equiv="X-UA-Compatible" content="ie=edge">
+    <title>Angel Wings</title>
+    <link rel="stylesheet" href="site.css">
+</head>
+<body>
+    <h1>Hello!!!</h1>
+</body>
+</html>

web/site.css

@@ -0,0 +1,3 @@
+h1 {
+    color: blue;
+}