fix(WebWorker): Add zone support to MessageBus

Closes #4053

modules/angular2/src/web_workers/shared/client_message_broker.ts

@@ -18,7 +18,11 @@ export {Type} from "angular2/src/core/facade/lang";
 export class ClientMessageBrokerFactory {
   constructor(private _messageBus: MessageBus, protected _serializer: Serializer) {}
 
-  createMessageBroker(channel: string): ClientMessageBroker {
+  /**
+   * Initializes the given channel and attaches a new {@link ClientMessageBroker} to it.
+   */
+  createMessageBroker(channel: string, runInZone: boolean = true): ClientMessageBroker {
+    this._messageBus.initChannel(channel, runInZone);
     return new ClientMessageBroker(this._messageBus, this._serializer, channel);
   }
 }

modules/angular2/src/web_workers/shared/generic_message_bus.dart

@@ -0,0 +1,151 @@
+library angular2.src.web_workers.shared.generic_message_bus;
+
+import 'dart:async';
+import 'package:angular2/src/core/facade/async.dart' show EventEmitter;
+import 'package:angular2/src/web_workers/shared/message_bus.dart'
+    show MessageBus, MessageBusSink, MessageBusSource;
+import 'package:angular2/src/core/zone/ng_zone.dart';
+import 'package:angular2/src/core/facade/lang.dart';
+
+class GenericMessageBus implements MessageBus {
+  final MessageBusSink _sink;
+  final MessageBusSource _source;
+
+  MessageBusSink get sink => _sink;
+  MessageBusSource get source => _source;
+
+  GenericMessageBus(MessageBusSink sink, MessageBusSource source)
+      : _sink = sink,
+        _source = source;
+
+  void attachToZone(NgZone zone) {
+    sink.attachToZone(zone);
+    source.attachToZone(zone);
+  }
+
+  void initChannel(String channel, [bool runInZone = true]) {
+    sink.initChannel(channel, runInZone);
+    source.initChannel(channel, runInZone);
+  }
+
+  EventEmitter from(String channel) {
+    return source.from(channel);
+  }
+
+  EventEmitter to(String channel) {
+    return sink.to(channel);
+  }
+}
+
+abstract class GenericMessageBusSink implements MessageBusSink {
+  NgZone _zone;
+  final _channels = new Map<String, _Channel>();
+  final _messageBuffer = new List<dynamic>();
+
+  void attachToZone(NgZone zone) {
+    _zone = zone;
+    _zone.overrideOnEventDone(() {
+      sendMessages(_messageBuffer);
+      _messageBuffer.clear();
+    }, false);
+  }
+
+  void initChannel(String channelName, [bool runInZone = true]) {
+    if (_channels.containsKey(channelName)) {
+      throw new BaseException("${channelName} has already been initialized.");
+    }
+
+    var emitter = new EventEmitter();
+    var channel = new _Channel(emitter, runInZone);
+
+    emitter.listen((data) {
+      var message = {'channel': channelName, 'message': data};
+      if (runInZone) {
+        _messageBuffer.add(message);
+      } else {
+        sendMessages([message]);
+      }
+    });
+
+    _channels[channelName] = channel;
+  }
+
+  EventEmitter to(String channelName) {
+    if (_channels.containsKey(channelName)) {
+      return _channels[channelName].emitter;
+    } else {
+      throw new BaseException(
+          "${channelName} is not set up. Did you forget to call initChannel?");
+    }
+  }
+
+  void sendMessages(List<dynamic> messages);
+}
+
+abstract class GenericMessageBusSource implements MessageBusSource {
+  Stream _stream;
+  final _channels = new Map<String, _Channel>();
+  NgZone _zone;
+
+  Stream get stream => _stream;
+
+  GenericMessageBusSource(Stream stream) {
+    attachTo(stream);
+  }
+
+  void attachTo(Stream stream) {
+    _stream = stream;
+    if (stream != null) {
+      stream.listen((messages) {
+        List<dynamic> decodedMessages = decodeMessages(messages);
+        if (decodedMessages != null) {
+          decodedMessages.forEach((message) => _handleMessage(message));
+        }
+      });
+    }
+  }
+
+  void attachToZone(NgZone zone) {
+    _zone = zone;
+  }
+
+  void initChannel(String channelName, [bool runInZone = true]) {
+    if (_channels.containsKey(channelName)) {
+      throw new BaseException("${channelName} has already been initialized.");
+    }
+
+    var emitter = new EventEmitter();
+    var channelInfo = new _Channel(emitter, runInZone);
+    _channels[channelName] = channelInfo;
+  }
+
+  EventEmitter from(String channelName) {
+    if (_channels.containsKey(channelName)) {
+      return _channels[channelName].emitter;
+    } else {
+      throw new BaseException(
+          "${channelName} is not set up. Did you forget to call initChannel?");
+    }
+  }
+
+  void _handleMessage(dynamic data) {
+    var channelName = data['channel'];
+    if (_channels.containsKey(channelName)) {
+      var channelInfo = _channels[channelName];
+      if (channelInfo.runInZone) {
+        _zone.run(() => channelInfo.emitter.add(data['message']));
+      } else {
+        channelInfo.emitter.add(data['message']);
+      }
+    }
+  }
+
+  List<dynamic> decodeMessages(dynamic message);
+}
+
+class _Channel {
+  EventEmitter emitter;
+  bool runInZone;
+
+  _Channel(this.emitter, this.runInZone);
+}

modules/angular2/src/web_workers/shared/isolate_message_bus.dart

@@ -1,76 +1,33 @@
 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/core/facade/async.dart';
-
-class IsolateMessageBus implements MessageBus {
-  final IsolateMessageBusSink sink;
-  final IsolateMessageBusSource source;
+import 'package:angular2/src/web_workers/shared/generic_message_bus.dart';
 
+class IsolateMessageBus extends GenericMessageBus {
   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);
-  }
+      : super(sink, source);
 }
 
-class IsolateMessageBusSink implements MessageBusSink {
+class IsolateMessageBusSink extends GenericMessageBusSink {
   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;
-    }
+  @override
+  void sendMessages(List<dynamic> messages) {
+    _port.send(messages);
   }
 }
 
-class IsolateMessageBusSource extends MessageBusSource {
-  final Stream rawDataStream;
-  final Map<String, EventEmitter> _channels = new Map<String, EventEmitter>();
+class IsolateMessageBusSource extends GenericMessageBusSource {
+  IsolateMessageBusSource(ReceivePort port) : super(port.asBroadcastStream());
 
-  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;
+  @override
+  List<dynamic> decodeMessages(dynamic messages) {
+    if (messages is SendPort) {
+      return null;
     }
+
+    return messages;
   }
 }

modules/angular2/src/web_workers/shared/message_bus.ts

@@ -1,5 +1,6 @@
 import {EventEmitter} from 'angular2/src/core/facade/async';
 import {BaseException} from 'angular2/src/core/facade/lang';
+import {NgZone} from 'angular2/src/core/zone/ng_zone';
 export {EventEmitter, Observable} from 'angular2/src/core/facade/async';
 
 function _abstract() {
@@ -13,6 +14,23 @@ function _abstract() {
  * by the corresponding MessageBusSource.
  */
 export /* abstract (with TS 1.6) */ class MessageBus implements MessageBusSource, MessageBusSink {
+  /**
+   * Sets up a new channel on the MessageBus.
+   * MUST be called before calling from or to on the channel.
+   * If runInZone is true then the source will emit events inside the angular zone
+   * and the sink will buffer messages and send only once the zone exits.
+   * if runInZone is false then the source will emit events inside the global zone
+   * and the sink will send messages immediatly.
+   */
+  initChannel(channel: string, runInZone: boolean = true): void { throw _abstract(); }
+
+  /**
+   * Assigns this bus to the given zone.
+   * Any callbacks attached to channels where runInZone was set to true on initialization
+   * will be executed in the given zone.
+   */
+  attachToZone(zone: NgZone): void { throw _abstract(); }
+
   /**
    * Returns an {@link EventEmitter} that emits every time a messsage
    * is received on the given channel.
@@ -28,6 +46,21 @@ export /* abstract (with TS 1.6) */ class MessageBus implements MessageBusSource
 }
 
 export interface MessageBusSource {
+  /**
+   * Sets up a new channel on the MessageBusSource.
+   * MUST be called before calling from on the channel.
+   * If runInZone is true then the source will emit events inside the angular zone.
+   * if runInZone is false then the source will emit events inside the global zone.
+   */
+  initChannel(channel: string, runInZone: boolean): void;
+
+  /**
+   * Assigns this source to the given zone.
+   * Any channels which are initialized with runInZone set to true will emit events that will be
+   * executed within the given zone.
+   */
+  attachToZone(zone: NgZone): void;
+
   /**
    * Returns an {@link EventEmitter} that emits every time a messsage
    * is received on the given channel.
@@ -36,6 +69,21 @@ export interface MessageBusSource {
 }
 
 export interface MessageBusSink {
+  /**
+   * Sets up a new channel on the MessageBusSink.
+   * MUST be called before calling to on the channel.
+   * If runInZone is true the sink will buffer messages and send only once the zone exits.
+   * if runInZone is false the sink will send messages immediatly.
+   */
+  initChannel(channel: string, runInZone: boolean): void;
+
+  /**
+   * Assigns this sink to the given zone.
+   * Any channels which are initilialized with runInZone set to true will wait for the given zone
+   * to exit before sending messages.
+   */
+  attachToZone(zone: NgZone): void;
+
   /**
    * 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

modules/angular2/src/web_workers/shared/post_message_bus.ts

@@ -3,9 +3,11 @@ import {
   MessageBusSource,
   MessageBusSink
 } from "angular2/src/web_workers/shared/message_bus";
+import {BaseException} from 'angular2/src/core/facade/lang';
 import {EventEmitter} from 'angular2/src/core/facade/async';
 import {StringMap, StringMapWrapper} from 'angular2/src/core/facade/collection';
 import {Injectable} from "angular2/src/core/di";
+import {NgZone} from 'angular2/src/core/zone/ng_zone';
 
 /**
  * A TypeScript implementation of {@link MessageBus} for communicating via JavaScript's
@@ -13,62 +15,131 @@ import {Injectable} from "angular2/src/core/di";
  */
 @Injectable()
 export class PostMessageBus implements MessageBus {
-  constructor(private _sink: PostMessageBusSink, private _source: PostMessageBusSource) {}
+  constructor(public sink: PostMessageBusSink, public source: PostMessageBusSource) {}
 
-  from(channel: string): EventEmitter { return this._source.from(channel); }
+  attachToZone(zone: NgZone): void {
+    this.source.attachToZone(zone);
+    this.sink.attachToZone(zone);
+  }
 
-  to(channel: string): EventEmitter { return this._sink.to(channel); }
+  initChannel(channel: string, runInZone: boolean = true): void {
+    this.source.initChannel(channel, runInZone);
+    this.sink.initChannel(channel, runInZone);
+  }
+
+  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();
+  private _zone: NgZone;
+  private _channels: StringMap<string, _Channel> = StringMapWrapper.create();
+  private _messageBuffer: Array<Object> = [];
 
   constructor(private _postMessageTarget: PostMessageTarget) {}
 
-  public to(channel: string): EventEmitter {
+  attachToZone(zone: NgZone): void {
+    this._zone = zone;
+    this._zone.overrideOnEventDone(() => this._handleOnEventDone(), false);
+  }
+
+  initChannel(channel: string, runInZone: boolean = true): void {
     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});
+      throw new BaseException(`${channel} has already been initialized`);
+    }
+
+    var emitter = new EventEmitter();
+    var channelInfo = new _Channel(emitter, runInZone);
+    this._channels[channel] = channelInfo;
+    emitter.observer({
+      next: (data: Object) => {
+        var message = {channel: channel, message: data};
+        if (runInZone) {
+          this._messageBuffer.push(message);
+        } else {
+          this._sendMessages([message]);
         }
-      });
-      return emitter;
+      }
+    });
+  }
+
+  to(channel: string): EventEmitter {
+    if (StringMapWrapper.contains(this._channels, channel)) {
+      return this._channels[channel].emitter;
+    } else {
+      throw new BaseException(`${channel} is not set up. Did you forget to call initChannel?`);
+    }
+  }
+
+  private _handleOnEventDone() {
+    // TODO: Send all buffered messages in one postMessage call
+    this._sendMessages(this._messageBuffer);
+    this._messageBuffer = [];
+  }
+
+  private _sendMessages(messages: Array<Object>) { this._postMessageTarget.postMessage(messages); }
+}
+
+export class PostMessageBusSource implements MessageBusSource {
+  private _zone: NgZone;
+  private _channels: StringMap<string, _Channel> = StringMapWrapper.create();
+
+  constructor(eventTarget?: EventTarget) {
+    if (eventTarget) {
+      eventTarget.addEventListener("message", (ev: MessageEvent) => this._handleMessages(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._handleMessages(ev));
+    }
+  }
+
+  attachToZone(zone: NgZone) { this._zone = zone; }
+
+  initChannel(channel: string, runInZone: boolean = true) {
+    if (StringMapWrapper.contains(this._channels, channel)) {
+      throw new BaseException(`${channel} has already been initialized`);
+    }
+
+    var emitter = new EventEmitter();
+    var channelInfo = new _Channel(emitter, runInZone);
+    this._channels[channel] = channelInfo;
+  }
+
+  from(channel: string): EventEmitter {
+    if (StringMapWrapper.contains(this._channels, channel)) {
+      return this._channels[channel].emitter;
+    } else {
+      throw new BaseException(`${channel} is not set up. Did you forget to call initChannel?`);
+    }
+  }
+
+  private _handleMessages(ev: MessageEvent): void {
+    var messages = ev.data;
+    for (var i = 0; i < messages.length; i++) {
+      this._handleMessage(messages[i]);
+    }
+  }
+
+  private _handleMessage(data: any): void {
+    var channel = data.channel;
+    if (StringMapWrapper.contains(this._channels, channel)) {
+      var channelInfo = this._channels[channel];
+      if (channelInfo.runInZone) {
+        this._zone.run(() => { channelInfo.emitter.next(data.message); });
+      } else {
+        channelInfo.emitter.next(data.message);
+      }
     }
   }
 }
 
-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;
-    }
-  }
+/**
+ * Helper class that wraps a channel's {@link EventEmitter} and
+ * keeps track of if it should run in the zone.
+ */
+class _Channel {
+  constructor(public emitter: EventEmitter, public runInZone: boolean) {}
 }
 
 // TODO(jteplitz602) Replace this with the definition in lib.webworker.d.ts(#3492)

modules/angular2/src/web_workers/shared/service_message_broker.ts

@@ -14,7 +14,11 @@ import {
 export class ServiceMessageBrokerFactory {
   constructor(private _messageBus: MessageBus, protected _serializer: Serializer) {}
 
-  createMessageBroker(channel: string): ServiceMessageBroker {
+  /**
+   * Initializes the given channel and attaches a new {@link ServiceMessageBroker} to it.
+   */
+  createMessageBroker(channel: string, runInZone: boolean = true): ServiceMessageBroker {
+    this._messageBus.initChannel(channel, runInZone);
     return new ServiceMessageBroker(this._messageBus, this._serializer, channel);
   }
 }