feat(WebWorkers): Add WebSocket MessageBuses for debugging apps

Closes #3858

modules/angular2/src/web_workers/debug_tools/multi_client_server_message_bus.dart

@@ -0,0 +1,216 @@
+library angular2.src.web_workers.debug_tools.multi_client_server_message_bus;
+
+import "package:angular2/src/web_workers/shared/message_bus.dart"
+    show MessageBus, MessageBusSink, MessageBusSource;
+import 'dart:io';
+import 'dart:convert' show JSON;
+import 'dart:async';
+import 'package:angular2/src/core/facade/async.dart' show EventEmitter;
+import 'package:angular2/src/web_workers/shared/messaging_api.dart';
+
+// TODO(jteplitz602): Remove hard coded result type and
+// clear messageHistory once app is done with it #3859
+class MultiClientServerMessageBus implements MessageBus {
+  final MultiClientServerMessageBusSink sink;
+  MultiClientServerMessageBusSource source;
+  bool hasPrimary = false;
+
+  MultiClientServerMessageBus(this.sink, this.source);
+
+  MultiClientServerMessageBus.fromHttpServer(HttpServer server)
+      : sink = new MultiClientServerMessageBusSink() {
+    source = new MultiClientServerMessageBusSource(resultReceived);
+    server.listen((HttpRequest request) {
+      if (request.uri.path == "/ws") {
+        WebSocketTransformer.upgrade(request).then((WebSocket socket) {
+          var wrapper = new WebSocketWrapper(
+              sink.messageHistory, sink.resultMarkers, socket);
+          if (!hasPrimary) {
+            wrapper.setPrimary(true);
+            hasPrimary = true;
+          }
+          sink.addConnection(wrapper);
+          source.addConnection(wrapper);
+
+          wrapper.stream.listen(null, onDone: _handleDisconnect(wrapper));
+        });
+      }
+    });
+  }
+
+  void resultReceived() {
+    sink.resultReceived();
+  }
+
+  EventEmitter from(String channel) {
+    return source.from(channel);
+  }
+
+  EventEmitter to(String channel) {
+    return sink.to(channel);
+  }
+
+  Function _handleDisconnect(WebSocketWrapper wrapper) {
+    return () {
+      sink.removeConnection(wrapper);
+      if (wrapper.isPrimary) {
+        hasPrimary = false;
+      }
+    };
+  }
+}
+
+class WebSocketWrapper {
+  WebSocket socket;
+  Stream stream;
+  int _numResultsReceived = 0;
+  bool _isPrimary = false;
+  bool caughtUp = false;
+  List<String> _messageHistory;
+  List<int> _resultMarkers;
+
+  WebSocketWrapper(this._messageHistory, this._resultMarkers, this.socket) {
+    stream = socket.asBroadcastStream();
+    stream.listen((encodedMessage) {
+      var message = JSON.decode(encodedMessage)['message'];
+      if (message is Map && message.containsKey("type")) {
+        if (message['type'] == 'result') {
+          resultReceived();
+        }
+      }
+    });
+  }
+
+  bool get isPrimary => _isPrimary;
+
+  void resultReceived() {
+    if (!isPrimary && !caughtUp) {
+      _numResultsReceived++;
+      sendToMarker(_numResultsReceived);
+    }
+  }
+
+  void setPrimary(bool primary) {
+    _isPrimary = primary;
+    if (primary) {
+      caughtUp = true;
+    }
+  }
+
+  // Sends up to the given result marker
+  void sendToMarker(int markerIndex) {
+    int numMessages;
+    int curr;
+    if (markerIndex >= _resultMarkers.length) {
+      // we're past the final result marker so send all messages in history
+      curr = (_resultMarkers.length > 0)
+          ? _resultMarkers[_resultMarkers.length - 1]
+          : 0;
+      numMessages = _messageHistory.length - curr;
+      caughtUp = true;
+    } else {
+      curr = (markerIndex == 0) ? 0 : _resultMarkers[markerIndex - 1];
+      var end = _resultMarkers[markerIndex];
+      numMessages = end - curr;
+    }
+    while (numMessages > 0) {
+      socket.add(_messageHistory[curr]);
+      curr++;
+      numMessages--;
+    }
+  }
+}
+
+class MultiClientServerMessageBusSink implements MessageBusSink {
+  final List<String> messageHistory = new List<String>();
+  final Set<WebSocketWrapper> openConnections = new Set<WebSocketWrapper>();
+  final Map<String, EventEmitter> _channels = new Map<String, EventEmitter>();
+  final List<int> resultMarkers = new List<int>();
+
+  void resultReceived() {
+    resultMarkers.add(messageHistory.length);
+  }
+
+  void addConnection(WebSocketWrapper webSocket) {
+    openConnections.add(webSocket);
+    // send messages up to the first result marker to this socket
+    webSocket.sendToMarker(0);
+  }
+
+  void removeConnection(WebSocketWrapper webSocket) {
+    openConnections.remove(webSocket);
+  }
+
+  EventEmitter to(String channel) {
+    if (_channels.containsKey(channel)) {
+      return _channels[channel];
+    } else {
+      var emitter = new EventEmitter();
+      emitter.listen((message) {
+        _send({'channel': channel, 'message': message});
+      });
+      return emitter;
+    }
+  }
+
+  void _send(dynamic message) {
+    String encodedMessage = JSON.encode(message);
+    openConnections.forEach((WebSocketWrapper webSocket) {
+      if (webSocket.caughtUp) {
+        webSocket.socket.add(encodedMessage);
+      }
+    });
+    messageHistory.add(encodedMessage);
+  }
+}
+
+class MultiClientServerMessageBusSource implements MessageBusSource {
+  final Map<String, EventEmitter> _channels = new Map<String, EventEmitter>();
+  Function onResultReceived;
+
+  MultiClientServerMessageBusSource(this.onResultReceived);
+
+  EventEmitter from(String channel) {
+    if (_channels.containsKey(channel)) {
+      return _channels[channel];
+    } else {
+      var emitter = new EventEmitter();
+      _channels[channel] = emitter;
+      return emitter;
+    }
+  }
+
+  void addConnection(WebSocketWrapper webSocket) {
+    if (webSocket.isPrimary) {
+      webSocket.stream.listen((encodedMessage) {
+        var decodedMessage = decodeMessage(encodedMessage);
+        var channel = decodedMessage['channel'];
+        var message = decodedMessage['message'];
+        if (message is Map && message.containsKey("type")) {
+          if (message['type'] == 'result') {
+            // tell the bus that a result was received on the primary
+            onResultReceived();
+          }
+        }
+
+        if (_channels.containsKey(channel)) {
+          _channels[channel].add(message);
+        }
+      });
+    } else {
+      webSocket.stream.listen((encodedMessage) {
+        // handle events from non-primary browser
+        var decodedMessage = decodeMessage(encodedMessage);
+        var channel = decodedMessage['channel'];
+        var message = decodedMessage['message'];
+        if (_channels.containsKey(EVENT_CHANNEL) && channel == EVENT_CHANNEL) {
+          _channels[channel].add(message);
+        }
+      });
+    }
+  }
+
+  Map<String, dynamic> decodeMessage(dynamic message) {
+    return JSON.decode(message);
+  }
+}

modules/angular2/src/web_workers/debug_tools/single_client_server_message_bus.dart

@@ -0,0 +1,140 @@
+library angular2.src.web_workers.debug_tools.single_client_server_message_bus;
+
+import "package:angular2/src/web_workers/shared/message_bus.dart"
+    show MessageBus, MessageBusSink, MessageBusSource;
+import 'dart:io';
+import 'dart:convert' show JSON;
+import 'dart:async';
+import "package:angular2/src/core/facade/async.dart" show EventEmitter;
+
+class SingleClientServerMessageBus implements MessageBus {
+  final SingleClientServerMessageBusSink sink;
+  SingleClientServerMessageBusSource source;
+  bool connected = false;
+
+  SingleClientServerMessageBus(this.sink, this.source);
+
+  SingleClientServerMessageBus.fromHttpServer(HttpServer server)
+      : sink = new SingleClientServerMessageBusSink() {
+    source = new SingleClientServerMessageBusSource();
+    server.listen((HttpRequest request) {
+      if (request.uri.path == "/ws") {
+        if (!connected) {
+          WebSocketTransformer.upgrade(request).then((WebSocket socket) {
+            sink.setConnection(socket);
+
+            var stream = socket.asBroadcastStream();
+            source.setConnectionFromStream(stream);
+            stream.listen(null, onDone: _handleDisconnect);
+          }).catchError((error) {
+            throw error;
+            connected = false;
+          });
+          connected = true;
+        } else {
+          // refuse additional clients
+          request.response.statusCode = HttpStatus.SERVICE_UNAVAILABLE;
+          request.response.write("Maximum number of clients connected.");
+          request.response.close();
+        }
+      }
+    });
+  }
+
+  void _handleDisconnect() {
+    sink.removeConnection();
+    source.removeConnection();
+    connected = false;
+  }
+
+  EventEmitter from(String channel) {
+    return source.from(channel);
+  }
+
+  EventEmitter to(String channel) {
+    return sink.to(channel);
+  }
+}
+
+class SingleClientServerMessageBusSink implements MessageBusSink {
+  final List<String> _messageBuffer = new List<String>();
+  WebSocket _socket = null;
+  final Map<String, EventEmitter> _channels = new Map<String, EventEmitter>();
+
+  void setConnection(WebSocket webSocket) {
+    _socket = webSocket;
+    _sendBufferedMessages();
+  }
+
+  EventEmitter to(String channel) {
+    if (_channels.containsKey(channel)) {
+      return _channels[channel];
+    } else {
+      var emitter = new EventEmitter();
+      emitter.listen((message) {
+        _send({'channel': channel, 'message': message});
+      });
+      return emitter;
+    }
+  }
+
+  void removeConnection() {
+    _socket = null;
+  }
+
+  void _send(dynamic message) {
+    String encodedMessage = JSON.encode(message);
+    if (_socket != null) {
+      _socket.add(encodedMessage);
+    } else {
+      _messageBuffer.add(encodedMessage);
+    }
+  }
+
+  void _sendBufferedMessages() {
+    _messageBuffer.forEach((message) => _socket.add(message));
+    _messageBuffer.clear();
+  }
+}
+
+class SingleClientServerMessageBusSource implements MessageBusSource {
+  final Map<String, EventEmitter> _channels = new Map<String, EventEmitter>();
+  Stream _stream;
+
+  SingleClientServerMessageBusSource();
+
+  EventEmitter from(String channel) {
+    if (_channels.containsKey(channel)) {
+      return _channels[channel];
+    } else {
+      var emitter = new EventEmitter();
+      _channels[channel] = emitter;
+      return emitter;
+    }
+  }
+
+  void setConnectionFromWebSocket(WebSocket socket) {
+    setConnectionFromStream(socket.asBroadcastStream());
+  }
+
+  void setConnectionFromStream(Stream stream) {
+    _stream = stream;
+    _stream.listen((encodedMessage) {
+      var decodedMessage = decodeMessage(encodedMessage);
+      var channel = decodedMessage['channel'];
+      var message = decodedMessage['message'];
+
+      if (_channels.containsKey(channel)) {
+        _channels[channel].add(message);
+      }
+    });
+  }
+
+  void removeConnection() {
+    _stream = null;
+  }
+
+  Map<String, dynamic> decodeMessage(dynamic message) {
+    return JSON.decode(message);
+  }
+}

modules/angular2/src/web_workers/debug_tools/web_socket_message_bus.dart

@@ -0,0 +1,78 @@
+library angular2.src.web_workers.worker.web_socket_message_bus;
+
+import 'dart:html';
+import 'dart:convert' show JSON;
+import "package:angular2/src/web_workers/shared/message_bus.dart"
+    show MessageBus, MessageBusSink, MessageBusSource;
+import 'package:angular2/src/core/facade/async.dart' show EventEmitter;
+
+class WebSocketMessageBus implements MessageBus {
+  final WebSocketMessageBusSink sink;
+  final WebSocketMessageBusSource source;
+
+  WebSocketMessageBus(this.sink, this.source);
+
+  WebSocketMessageBus.fromWebSocket(WebSocket webSocket)
+      : sink = new WebSocketMessageBusSink(webSocket),
+        source = new WebSocketMessageBusSource(webSocket);
+
+  EventEmitter from(String channel) {
+    return source.from(channel);
+  }
+
+  EventEmitter to(String channel) {
+    return sink.to(channel);
+  }
+}
+
+class WebSocketMessageBusSink implements MessageBusSink {
+  final WebSocket _webSocket;
+  final Map<String, EventEmitter> _channels = new Map<String, EventEmitter>();
+
+  WebSocketMessageBusSink(this._webSocket);
+
+  EventEmitter to(String channel) {
+    if (_channels.containsKey(channel)) {
+      return _channels[channel];
+    } else {
+      var emitter = new EventEmitter();
+      emitter.listen((message) {
+        _send({'channel': channel, 'message': message});
+      });
+      _channels[channel] = emitter;
+      return emitter;
+    }
+  }
+
+  void _send(message) {
+    _webSocket.send(JSON.encode(message));
+  }
+}
+
+class WebSocketMessageBusSource implements MessageBusSource {
+  final Map<String, EventEmitter> _channels = new Map<String, EventEmitter>();
+
+  WebSocketMessageBusSource(WebSocket webSocket) {
+    webSocket.onMessage.listen((MessageEvent encodedMessage) {
+      var message = decodeMessage(encodedMessage.data);
+      var channel = message['channel'];
+      if (_channels.containsKey(channel)) {
+        _channels[channel].add(message['message']);
+      }
+    });
+  }
+
+  EventEmitter from(String channel) {
+    if (_channels.containsKey(channel)) {
+      return _channels[channel];
+    } else {
+      var emitter = new EventEmitter();
+      _channels[channel] = emitter;
+      return emitter;
+    }
+  }
+
+  Map<String, dynamic> decodeMessage(dynamic message) {
+    return JSON.decode(message);
+  }
+}