8

README.md

@@ -1,5 +1,5 @@
 # angel_websocket
-[![1.0.0-dev+7](https://img.shields.io/badge/version-1.0.0--dev+7-red.svg)](https://pub.dartlang.org/packages/angel_websocket)
+[![1.0.0-dev+8](https://img.shields.io/badge/pub-1.0.0--dev+8-red.svg)](https://pub.dartlang.org/packages/angel_websocket)
 [![build status](https://travis-ci.org/angel-dart/websocket.svg)](https://travis-ci.org/angel-dart/websocket)
 
 WebSocket plugin for Angel.
@@ -27,6 +27,15 @@ main() async {
 
 ```
 
+Filtering events is easy with services. Just return a `bool`, whether
+synchronously or asynchronously.
+
+```dart
+myService.properties['ws:filter'] = (WebSocketContext socket) async {
+  return true;
+}
+```
+
 **Adding Handlers within a Controller**
 
 `WebSocketController` extends a normal `Controller`, but also listens to WebSockets.
@@ -48,6 +57,12 @@ class MyController extends WebSocketController {
   void sendMessage(WebSocketContext socket, Db db) async {
     socket.send("found_message", db.collection("messages").findOne(where.id("...")));
   }
+
+  // Event filtering
+  @ExposeWs("foo")
+  void foo() {
+    broadcast(new WebSocketEvent(...), filter: (socket) async => ...);
+  }
 }
 ```
 

lib/server.dart

@@ -20,7 +20,7 @@ typedef AngelWebSocketRegisterer(Angel app, RequestHandler handler);
 /// Broadcasts events from [HookedService]s, and handles incoming [WebSocketAction]s.
 class AngelWebSocket extends AngelPlugin {
   Angel _app;
-  List<WebSocket> _clients = [];
+  List<WebSocketContext> _clients = [];
   final List<String> _servicesAlreadyWired = [];
 
   final StreamController<WebSocketAction> _onAction =
@@ -38,7 +38,7 @@ class AngelWebSocket extends AngelPlugin {
   final AngelWebSocketRegisterer register;
 
   /// A list of clients currently connected to this server via WebSockets.
-  List<WebSocket> get clients => new List.unmodifiable(_clients);
+  List<WebSocketContext> get clients => new List.unmodifiable(_clients);
 
   /// Services that have already been hooked to fire socket events.
   List<String> get servicesAlreadyWired =>
@@ -65,15 +65,26 @@ class AngelWebSocket extends AngelPlugin {
     return (HookedServiceEvent e) async {
       var event = await transformEvent(e);
       event.eventName = "$path::${event.eventName}";
-      await batchEvent(event);
+
+      _filter(WebSocketContext socket) {
+        if (e.service.properties.containsKey('ws:filter'))
+          return e.service.properties['ws:filter'](socket);
+        else
+          return true;
+      }
+
+      await batchEvent(event, filter: _filter);
     };
   }
 
   /// Slates an event to be dispatched.
-  Future batchEvent(WebSocketEvent event) async {
+  Future batchEvent(WebSocketEvent event,
+      {filter(WebSocketContext socket)}) async {
     // Default implementation will just immediately fire events
-    _clients.forEach((client) {
+    _clients.forEach((client) async {
-      client.add(god.serialize(event));
+      var result = true;
+      if (filter != null) result = await filter(client);
+      if (result == true) client.io.add(god.serialize(event));
     });
   }
 
@@ -240,13 +251,15 @@ class AngelWebSocket extends AngelPlugin {
         ..end();
 
       var ws = await WebSocketTransformer.upgrade(req.io);
-      _clients.add(ws);
-
       var socket = new WebSocketContext(ws, req, res);
+      _clients.add(socket);
       await handleConnect(socket);
 
       _onConnection.add(socket);
-      req.properties['socket'] = socket;
+
+      req
+        ..properties['socket'] = socket
+        ..inject(WebSocketContext, socket);
 
       ws.listen((data) {
         _onData.add(data);

lib/websocket_controller.dart

@@ -1,28 +1,38 @@
 part of angel_websocket.server;
 
+/// Marks a method as available to WebSockets.
 class ExposeWs {
   final String eventName;
 
   const ExposeWs(this.eventName);
 }
 
+/// A special controller that also supports WebSockets.
 class WebSocketController extends Controller {
   Map<String, MethodMirror> _handlers = {};
   Map<String, Symbol> _handlerSymbols = {};
-  AngelWebSocket ws;
+
+  /// The plug-in instance powering this controller.
+  AngelWebSocket plugin;
 
   WebSocketController() : super();
 
-  void broadcast(String eventName, data) {
+  /// Sends an event to all clients.
-    ws.batchEvent(new WebSocketEvent(eventName: eventName, data: data));
+  void broadcast(String eventName, data, {filter(WebSocketContext socket)}) {
+    plugin.batchEvent(new WebSocketEvent(eventName: eventName, data: data),
+        filter: filter);
   }
 
+  /// Fired on new connections.
   onConnect(WebSocketContext socket) {}
 
+  /// Fired on disconnections.
   onDisconnect(WebSocketContext socket) {}
 
+  /// Fired on all incoming actions.
   onAction(WebSocketAction action, WebSocketContext socket) async {}
 
+  /// Fired on arbitrary incoming data.
   onData(data, WebSocketContext socket) {}
 
   @override

pubspec.yaml

@@ -2,7 +2,7 @@ name: angel_websocket
 description: WebSocket plugin for Angel
 environment:
   sdk: ">=1.19.0"
-version: 1.0.0-dev+7
+version: 1.0.0-dev+8
 author: Tobe O <thosakwe@gmail.com>
 homepage: https://github.com/angel-dart/angel_websocket
 dependencies:

test/controller/io_test.dart

@@ -1,5 +1,5 @@
 import 'dart:io';
-import 'package:angel_diagnostics/angel_diagnostics.dart' as srv;
+import 'package:angel_diagnostics/angel_diagnostics.dart';
 import 'package:angel_framework/angel_framework.dart' as srv;
 import 'package:angel_websocket/io.dart' as ws;
 import 'package:angel_websocket/server.dart' as srv;
@@ -23,9 +23,9 @@ main() {
 
     await app.configure(websockets);
     await app.configure(new GameController());
+    await app.configure(logRequests(new File('log.txt')));
 
-    server =
+    server = await app.startServer();
-        await new srv.DiagnosticsServer(app, new File('log.txt')).startServer();
     url = 'ws://${server.address.address}:${server.port}/ws';
 
     client = new ws.WebSockets(url);