feat(WebWorker) Add channel support to MessageBus

closes angular#3661 and angular#3686

Author: jteplitz

modules/angular2/src/facade/async.dart

@@ -81,13 +81,13 @@ class ObservableWrapper {
 }
 
 class EventEmitter extends Stream {
-  StreamController<String> _controller;
+  StreamController<dynamic> _controller;
 
   EventEmitter() {
     _controller = new StreamController.broadcast();
   }
 
-  StreamSubscription listen(void onData(String line),
+  StreamSubscription listen(void onData(dynamic line),
       {void onError(Error error), void onDone(), bool cancelOnError}) {
     return _controller.stream.listen(onData,
         onError: onError, onDone: onDone, cancelOnError: cancelOnError);

modules/angular2/src/web-workers/shared/isolate_message_bus.dart

@@ -0,0 +1,76 @@
+library angular2.src.web_workers.shared.isolate_message_bus;
+
+import 'dart:isolate';
+import 'dart:async';
+import 'dart:core';
+import 'package:angular2/src/web-workers/shared/message_bus.dart'
+    show MessageBus, MessageBusSink, MessageBusSource;
+import 'package:angular2/src/facade/async.dart';
+
+class IsolateMessageBus implements MessageBus {
+  final IsolateMessageBusSink sink;
+  final IsolateMessageBusSource source;
+
+  IsolateMessageBus(IsolateMessageBusSink sink, IsolateMessageBusSource source)
+      : sink = sink,
+        source = source;
+
+  EventEmitter from(String channel) {
+    return source.from(channel);
+  }
+
+  EventEmitter to(String channel) {
+    return sink.to(channel);
+  }
+}
+
+class IsolateMessageBusSink implements MessageBusSink {
+  final SendPort _port;
+  final Map<String, EventEmitter> _channels = new Map<String, EventEmitter>();
+
+  IsolateMessageBusSink(SendPort port) : _port = port;
+
+  EventEmitter to(String channel) {
+    if (_channels.containsKey(channel)) {
+      return _channels[channel];
+    } else {
+      var emitter = new EventEmitter();
+      emitter.listen((message) {
+        _port.send({'channel': channel, 'message': message});
+      });
+      _channels[channel] = emitter;
+      return emitter;
+    }
+  }
+}
+
+class IsolateMessageBusSource extends MessageBusSource {
+  final Stream rawDataStream;
+  final Map<String, EventEmitter> _channels = new Map<String, EventEmitter>();
+
+  IsolateMessageBusSource(ReceivePort port)
+      : rawDataStream = port.asBroadcastStream() {
+    rawDataStream.listen((message) {
+      if (message is SendPort){
+        return;
+      }
+
+      if (message.containsKey("channel")) {
+        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;
+    }
+  }
+}

modules/angular2/src/web-workers/shared/message_bus.ts

@@ -1,26 +1,52 @@
-// TODO(jteplitz602) to be idiomatic these should be releated to Observable's or Streams
+import {EventEmitter} from 'angular2/src/facade/async';
+import {BaseException} from 'angular2/src/facade/lang';
+// TODO(jteplitz602): Replace both the interface and the exported class with an abstract class #3683
+
+function _abstract() {
+  throw new BaseException("This method is abstract");
+}
+
 /**
- * Message Bus is a low level API used to communicate between the UI and the worker.
- * It smooths out the differences between Javascript's postMessage and Dart's Isolate
- * allowing you to work with one consistent API.
+ * Message Bus is a low level API used to communicate between the UI and the background.
+ * Communication is based on a channel abstraction. Messages published in a
+ * given channel to one MessageBusSink are received on the same channel
+ * by the corresponding MessageBusSource.
+ * TODO(jteplitz602): This should just extend both the source and the sink once
+ * https://github.com/angular/ts2dart/issues/263 is closed.
  */
-export interface MessageBus {
-  sink: MessageBusSink;
-  source: MessageBusSource;
-}
+export interface MessageBusInterface {
+  /**
+   * Returns an {@link EventEmitter} that emits every time a messsage
+   * is received on the given channel.
+   */
+  from(channel: string): EventEmitter;
 
-export interface SourceListener {
-  (data: any): void;  // TODO: Replace this any type with the type of a real messaging protocol
+  /**
+   * Returns an {@link EventEmitter} for the given channel
+   * To publish methods to that channel just call next (or add in dart) on the returned emitter
+   */
+  to(channel: string): EventEmitter;
 }
 
 export interface MessageBusSource {
   /**
-   * Attaches the SourceListener to this source.
-   * The SourceListener will get called whenever the bus receives a message
-   * Returns a listener id that can be passed to {removeListener}
+   * Returns an {@link EventEmitter} that emits every time a messsage
+   * is received on the given channel.
    */
-  addListener(fn: SourceListener): number;
-  removeListener(index: number);
+  from(channel: string): EventEmitter;
 }
 
-export interface MessageBusSink { send(message: Object): void; }
+export interface MessageBusSink {
+  /**
+   * Returns an {@link EventEmitter} for the given channel
+   * To publish methods to that channel just call next (or add in dart) on the returned emitter
+   */
+  to(channel: string): EventEmitter;
+}
+
+// TODO(jteplitz602): Remove this class once we have abstract classes #3683
+export class MessageBus implements MessageBusInterface {
+  from(channel: string): EventEmitter { throw _abstract(); }
+
+  to(channel: string): EventEmitter { throw _abstract(); }
+}

modules/angular2/src/web-workers/shared/messaging_api.ts

@@ -0,0 +1,9 @@
+/**
+ * All channels used by angular's WebWorker components are listed here.
+ * You should not use these channels in your application code.
+ */
+export const SETUP_CHANNEL = "ng-WebWorkerSetup";
+export const RENDER_COMPILER_CHANNEL = "ng-RenderCompiler";
+export const RENDERER_CHANNEL = "ng-Renderer";
+export const XHR_CHANNEL = "ng-XHR";
+export const EVENT_CHANNEL = "ng-events";

modules/angular2/src/web-workers/shared/post_message_bus.dart

@@ -0,0 +1,3 @@
+// PostMessageBus can't be implemented in dart since dart doesn't use postMessage
+// This file is only here to prevent ts2dart from trying to transpile the PostMessageBus
+library angular2.src.web_workers.shared.post_message_bus;

modules/angular2/src/web-workers/shared/post_message_bus.ts

@@ -0,0 +1,75 @@
+import {
+  MessageBusInterface,
+  MessageBusSource,
+  MessageBusSink
+} from "angular2/src/web-workers/shared/message_bus";
+import {EventEmitter} from 'angular2/src/facade/async';
+import {StringMap, StringMapWrapper} from 'angular2/src/facade/collection';
+import {Injectable} from "angular2/di";
+
+/**
+ * A TypeScript implementation of {@link MessageBus} for communicating via JavaScript's
+ * postMessage API.
+ */
+@Injectable()
+export class PostMessageBus implements MessageBusInterface {
+  constructor(private _sink: PostMessageBusSink, private _source: PostMessageBusSource) {}
+
+  from(channel: string): EventEmitter { return this._source.from(channel); }
+
+  to(channel: string): EventEmitter { return this._sink.to(channel); }
+}
+
+export class PostMessageBusSink implements MessageBusSink {
+  private _channels: StringMap<string, EventEmitter> = StringMapWrapper.create();
+
+  constructor(private _postMessageTarget: PostMessageTarget) {}
+
+  public to(channel: string): EventEmitter {
+    if (StringMapWrapper.contains(this._channels, channel)) {
+      return this._channels[channel];
+    } else {
+      var emitter = new EventEmitter();
+      emitter.observer({
+        next: (message: Object) => {
+          this._postMessageTarget.postMessage({channel: channel, message: message});
+        }
+      });
+      return emitter;
+    }
+  }
+}
+
+export class PostMessageBusSource implements MessageBusSource {
+  private _channels: StringMap<string, EventEmitter> = StringMapWrapper.create();
+
+  constructor(eventTarget?: EventTarget) {
+    if (eventTarget) {
+      eventTarget.addEventListener("message", (ev: MessageEvent) => this._handleMessage(ev));
+    } else {
+      // if no eventTarget is given we assume we're in a WebWorker and listen on the global scope
+      addEventListener("message", (ev: MessageEvent) => this._handleMessage(ev));
+    }
+  }
+
+  private _handleMessage(ev: MessageEvent) {
+    var data = ev.data;
+    var channel = data.channel;
+    if (StringMapWrapper.contains(this._channels, channel)) {
+      this._channels[channel].next(data.message);
+    }
+  }
+
+  public from(channel: string): EventEmitter {
+    if (StringMapWrapper.contains(this._channels, channel)) {
+      return this._channels[channel];
+    } else {
+      var emitter = new EventEmitter();
+      this._channels[channel] = emitter;
+      return emitter;
+    }
+  }
+}
+
+// TODO(jteplitz602) Replace this with the definition in lib.webworker.d.ts(#3492)
+export interface PostMessageTarget { postMessage: (message: any, transfer?:[ArrayBuffer]) => void; }

modules/angular2/src/web-workers/ui/application.dart

@@ -4,8 +4,9 @@ import 'dart:isolate';
 import 'dart:async';
 import 'dart:core';
 import 'package:angular2/src/web-workers/shared/message_bus.dart'
-    show MessageBus, MessageBusSink, MessageBusSource;
+    show MessageBus;
 import 'package:angular2/src/web-workers/ui/impl.dart' show bootstrapUICommon;
+import 'package:angular2/src/web-workers/shared/isolate_message_bus.dart';
 
 /**
  * Bootstrapping a WebWorker
@@ -23,63 +24,22 @@ Future<MessageBus> bootstrap(String uri) {
 /**
  * To be called from the main thread to spawn and communicate with the worker thread
  */
-Future<UIMessageBus> spawnWebWorker(Uri uri) {
+Future<MessageBus> spawnWebWorker(Uri uri) {
   var receivePort = new ReceivePort();
   var isolateEndSendPort = receivePort.sendPort;
   return Isolate.spawnUri(uri, const [], isolateEndSendPort).then((_) {
     var source = new UIMessageBusSource(receivePort);
     return source.sink.then((sendPort) {
-      var sink = new UIMessageBusSink(sendPort);
-      return new UIMessageBus(sink, source);
+      var sink = new IsolateMessageBusSink(sendPort);
+      return new IsolateMessageBus(sink, source);
     });
   });
 }
 
-class UIMessageBus extends MessageBus {
-  final UIMessageBusSink sink;
-  final UIMessageBusSource source;
-
-  UIMessageBus(UIMessageBusSink sink, UIMessageBusSource source)
-      : sink = sink,
-        source = source;
-}
-
-class UIMessageBusSink extends MessageBusSink {
-  final SendPort _port;
-
-  UIMessageBusSink(SendPort port) : _port = port;
-
-  void send(message) {
-    _port.send(message);
-  }
-}
-
-class UIMessageBusSource extends MessageBusSource {
-  final ReceivePort _port;
-  final Stream rawDataStream;
-  Map<int, StreamSubscription> _listenerStore =
-      new Map<int, StreamSubscription>();
-  int _numListeners = 0;
-
-  UIMessageBusSource(ReceivePort port)
-      : _port = port,
-        rawDataStream = port.asBroadcastStream();
+class UIMessageBusSource extends IsolateMessageBusSource {
+  UIMessageBusSource(ReceivePort port) : super(port);
 
   Future<SendPort> get sink => rawDataStream.firstWhere((message) {
         return message is SendPort;
       });
-
-  int addListener(Function fn) {
-    var subscription = rawDataStream.listen((message) {
-      fn({"data": message});
-    });
-
-    _listenerStore[++_numListeners] = subscription;
-    return _numListeners;
-  }
-
-  void removeListener(int index) {
-    _listenerStore[index].cancel();
-    _listenerStore.remove(index);
-  }
 }

modules/angular2/src/web-workers/ui/application.ts

@@ -1,9 +1,9 @@
 import {
-  MessageBus,
-  MessageBusSource,
-  MessageBusSink,
-  SourceListener
-} from "angular2/src/web-workers/shared/message_bus";
+  PostMessageBus,
+  PostMessageBusSink,
+  PostMessageBusSource
+} from 'angular2/src/web-workers/shared/post_message_bus';
+import {MessageBus} from 'angular2/src/web-workers/shared/message_bus';
 import {BaseException} from "angular2/src/facade/lang";
 import {bootstrapUICommon} from "angular2/src/web-workers/ui/impl";
 
@@ -23,33 +23,7 @@ export function bootstrap(uri: string): MessageBus {
 
 export function spawnWebWorker(uri: string): MessageBus {
   var webWorker: Worker = new Worker(uri);
-  return new UIMessageBus(new UIMessageBusSink(webWorker), new UIMessageBusSource(webWorker));
-}
-
-export class UIMessageBus implements MessageBus {
-  constructor(public sink: UIMessageBusSink, public source: UIMessageBusSource) {}
-}
-
-export class UIMessageBusSink implements MessageBusSink {
-  constructor(private _webWorker: Worker) {}
-
-  send(message: Object): void { this._webWorker.postMessage(message); }
-}
-
-export class UIMessageBusSource implements MessageBusSource {
-  private _listenerStore: Map<int, SourceListener> = new Map<int, SourceListener>();
-  private _numListeners: int = 0;
-
-  constructor(private _webWorker: Worker) {}
-
-  public addListener(fn: SourceListener): int {
-    this._webWorker.addEventListener("message", fn);
-    this._listenerStore[++this._numListeners] = fn;
-    return this._numListeners;
-  }
-
-  public removeListener(index: int): void {
-    removeEventListener("message", this._listenerStore[index]);
-    this._listenerStore.delete(index);
-  }
+  var sink = new PostMessageBusSink(webWorker);
+  var source = new PostMessageBusSource(webWorker);
+  return new PostMessageBus(sink, source);
 }