Fix travis

CHANGELOG.md

@@ -1,5 +1,6 @@
 # 2.0.3
 * Remove `WebSocketController.plugin`.
+* Remove any unawaited futures.
 
 # 2.0.2
 * Update `stream_channel` to `2.0.0`.

analysis_options.yaml

@@ -2,5 +2,3 @@ include: package:pedantic/analysis_options.yaml
 analyzer:
     strong-mode:
       implicit-casts: false
-    errors:
-      unawaited_futures: ignore

example/main.dart

@@ -46,7 +46,7 @@ main(List<String> args) async {
     }
 
     var http2 = new AngelHttp2(app, ctx);
-    http2.onHttp1.forEach(http.handleRequest);
+    http2.onHttp1.listen(http.handleRequest);
     await http2.startServer('127.0.0.1', 3000);
     print('Listening at ${http2.uri}');
   } else {

lib/base_websocket_client.dart

@@ -95,13 +95,15 @@ abstract class BaseWebSocketClient extends BaseAngelClient {
   @override
   Future close() async {
     on._close();
-    await _socket.sink.close(status.goingAway);
-    _onData.close();
-    _onAllEvents.close();
-    _onAuthenticated.close();
-    _onError.close();
-    _onServiceEvent.close();
-    _onWebSocketChannelException.close();
+    scheduleMicrotask(() async {
+      await _socket.sink.close(status.goingAway);
+      await _onData.close();
+      await _onAllEvents.close();
+      await _onAuthenticated.close();
+      await _onError.close();
+      await _onServiceEvent.close();
+      await _onWebSocketChannelException.close();
+    });
   }
 
   /// Connects the WebSocket. [timeout] is optional.
@@ -119,22 +121,24 @@ abstract class BaseWebSocketClient extends BaseAngelClient {
         }
       });
 
-      getConnectedWebSocket().then((socket) {
-        if (!c.isCompleted) {
-          if (timer.isActive) timer.cancel();
+      scheduleMicrotask(() {
+        return getConnectedWebSocket().then((socket) {
+          if (!c.isCompleted) {
+            if (timer.isActive) timer.cancel();
 
-          while (_queue.isNotEmpty) {
-            var action = _queue.removeFirst();
-            socket.sink.add(serialize(action));
+            while (_queue.isNotEmpty) {
+              var action = _queue.removeFirst();
+              socket.sink.add(serialize(action));
+            }
+
+            c.complete(socket);
           }
-
-          c.complete(socket);
-        }
-      }).catchError((e, StackTrace st) {
-        if (!c.isCompleted) {
-          if (timer.isActive) timer.cancel();
-          c.completeError(e, st);
-        }
+        }).catchError((e, StackTrace st) {
+          if (!c.isCompleted) {
+            if (timer.isActive) timer.cancel();
+            c.completeError(e, st);
+          }
+        });
       });
 
       return await c.future.then((socket) {
@@ -293,13 +297,13 @@ class WebSocketsService<Id, Data> extends Service<Id, Data> {
   }
 
   Future close() async {
-    _onAllEvents.close();
-    _onCreated.close();
-    _onIndexed.close();
-    _onModified.close();
-    _onRead.close();
-    _onRemoved.close();
-    _onUpdated.close();
+    await _onAllEvents.close();
+    await _onCreated.close();
+    await _onIndexed.close();
+    await _onModified.close();
+    await _onRead.close();
+    await _onRemoved.close();
+    await _onUpdated.close();
   }
 
   /// Serializes an [action] to be sent over a WebSocket.

lib/server.dart

@@ -387,7 +387,7 @@ class AngelWebSocket {
       var ws = await WebSocketTransformer.upgrade(req.rawRequest);
       var channel = new IOWebSocketChannel(ws);
       var socket = new WebSocketContext(channel, req, res);
-      handleClient(socket);
+      scheduleMicrotask(() => handleClient(socket));
       return false;
     } else if (req is Http2RequestContext && res is Http2ResponseContext) {
       var connection =
@@ -428,9 +428,9 @@ class AngelWebSocket {
         });
 
         if (req.hasParsedBody) {
-          ctrl.local.sink.close();
+          await ctrl.local.sink.close();
         } else {
-          req.body.pipe(ctrl.local.sink);
+          await req.body.pipe(ctrl.local.sink);
         }
 
         var sink = utf8.encoder.startChunkedConversion(ctrl.foreign.sink);
@@ -443,7 +443,7 @@ class AngelWebSocket {
 
         var ws = new WebSocketChannel(ctrl.foreign);
         var socket = new WebSocketContext(ws, req, res);
-        handleClient(socket);
+        scheduleMicrotask(() => handleClient(socket));
         return false;
       }
     } else {

lib/websocket_context.dart

@@ -40,12 +40,14 @@ class WebSocketContext {
 
   /// Closes the underlying [StreamChannel].
   Future close() async {
-    await channel.sink.close();
-    _onAction.close();
-    _onAuthenticated.close();
-    _onData.close();
-    _onClose.add(null);
-    _onClose.close();
+    scheduleMicrotask(() async {
+      await channel.sink.close();
+      await _onAction.close();
+      await _onAuthenticated.close();
+      await _onData.close();
+      await _onClose.add(null);
+      await _onClose.close();
+    });
   }
 
   /// Sends an arbitrary [WebSocketEvent];

test/service/common.dart

@@ -1,3 +1,5 @@
+import 'dart:async';
+
 import 'package:angel_framework/angel_framework.dart';
 import 'package:angel_model/angel_model.dart';
 import 'package:angel_websocket/base_websocket_client.dart';
@@ -23,7 +25,7 @@ class TodoService extends MapService {
 
 testIndex(BaseWebSocketClient client) async {
   var todoService = client.service('api/todos');
-  todoService.index();
+  scheduleMicrotask(() => todoService.index());
 
   var indexed = await todoService.onIndexed.first;
   print('indexed: $indexed');