refactor(WebWorker): Abstract message passing and serialization to UIMessageBroker

closes #3703

Closes #3815

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

@@ -0,0 +1,136 @@
+/// <reference path="../../../globals.d.ts" />
+import {MessageBus} from "angular2/src/web_workers/shared/message_bus";
+import {print, isPresent, DateWrapper, stringify} from "../../facade/lang";
+import {
+  Promise,
+  PromiseCompleter,
+  PromiseWrapper,
+  ObservableWrapper,
+  EventEmitter
+} from "angular2/src/facade/async";
+import {ListWrapper, StringMapWrapper, MapWrapper} from "../../facade/collection";
+import {Serializer} from "angular2/src/web_workers/shared/serializer";
+import {Injectable} from "angular2/di";
+import {Type} from "angular2/src/facade/lang";
+
+@Injectable()
+export class ClientMessageBrokerFactory {
+  constructor(private _messageBus: MessageBus, protected _serializer: Serializer) {}
+
+  createMessageBroker(channel: string): ClientMessageBroker {
+    return new ClientMessageBroker(this._messageBus, this._serializer, channel);
+  }
+}
+
+export class ClientMessageBroker {
+  private _pending: Map<string, PromiseCompleter<any>> = new Map<string, PromiseCompleter<any>>();
+  private _sink: EventEmitter;
+
+  constructor(messageBus: MessageBus, protected _serializer: Serializer, public channel) {
+    this._sink = messageBus.to(channel);
+    var source = messageBus.from(channel);
+    ObservableWrapper.subscribe(source, (message) => this._handleMessage(message));
+  }
+
+  private _generateMessageId(name: string): string {
+    var time: string = stringify(DateWrapper.toMillis(DateWrapper.now()));
+    var iteration: number = 0;
+    var id: string = name + time + stringify(iteration);
+    while (isPresent(this._pending[id])) {
+      id = `${name}${time}${iteration}`;
+      iteration++;
+    }
+    return id;
+  }
+
+  runOnUiThread(args: UiArguments, returnType: Type): Promise<any> {
+    var fnArgs = [];
+    if (isPresent(args.args)) {
+      ListWrapper.forEach(args.args, (argument) => {
+        if (argument.type != null) {
+          fnArgs.push(this._serializer.serialize(argument.value, argument.type));
+        } else {
+          fnArgs.push(argument.value);
+        }
+      });
+    }
+
+    var promise: Promise<any>;
+    var id: string = null;
+    if (returnType != null) {
+      var completer: PromiseCompleter<any> = PromiseWrapper.completer();
+      id = this._generateMessageId(args.method);
+      this._pending.set(id, completer);
+      PromiseWrapper.catchError(completer.promise, (err, stack?) => {
+        print(err);
+        completer.reject(err, stack);
+      });
+
+      promise = PromiseWrapper.then(completer.promise, (value: any) => {
+        if (this._serializer == null) {
+          return value;
+        } else {
+          return this._serializer.deserialize(value, returnType);
+        }
+      });
+    } else {
+      promise = null;
+    }
+
+    // TODO(jteplitz602): Create a class for these messages so we don't keep using StringMap #3685
+    var message = {'method': args.method, 'args': fnArgs};
+    if (id != null) {
+      message['id'] = id;
+    }
+    ObservableWrapper.callNext(this._sink, message);
+
+    return promise;
+  }
+
+  private _handleMessage(message: StringMap<string, any>): void {
+    var data = new MessageData(message);
+    // TODO(jteplitz602): replace these strings with messaging constants #3685
+    if (data.type === "result" || data.type === "error") {
+      var id = data.id;
+      if (this._pending.has(id)) {
+        if (data.type === "result") {
+          this._pending.get(id).resolve(data.value);
+        } else {
+          this._pending.get(id).reject(data.value, null);
+        }
+        this._pending.delete(id);
+      }
+    }
+  }
+}
+
+class MessageData {
+  type: string;
+  value: any;
+  id: string;
+
+  constructor(data: StringMap<string, any>) {
+    this.type = StringMapWrapper.get(data, "type");
+    this.id = this._getValueIfPresent(data, "id");
+    this.value = this._getValueIfPresent(data, "value");
+  }
+
+  /**
+   * Returns the value from the StringMap if present. Otherwise returns null
+   */
+  _getValueIfPresent(data: StringMap<string, any>, key: string) {
+    if (StringMapWrapper.contains(data, key)) {
+      return StringMapWrapper.get(data, key);
+    } else {
+      return null;
+    }
+  }
+}
+
+export class FnArg {
+  constructor(public value, public type) {}
+}
+
+export class UiArguments {
+  constructor(public method: string, public args?: List<FnArg>) {}
+}

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

@@ -40,6 +40,10 @@ import {
   RenderViewWithFragmentsStore
 } from 'angular2/src/web_workers/shared/render_view_with_fragments_store';
 
+// PRIMITIVE is any type that does not need to be serialized (string, number, boolean)
+// We set it to String so that it is considered a Type.
+export const PRIMITIVE: Type = String;
+
 @Injectable()
 export class Serializer {
   private _enumRegistry: Map<any, Map<number, any>>;
@@ -76,7 +80,7 @@ export class Serializer {
       ListWrapper.forEach(obj, (val) => { serializedObj.push(this.serialize(val, type)); });
       return serializedObj;
     }
-    if (type == String) {
+    if (type == PRIMITIVE) {
       return obj;
     }
     if (type == ViewDefinition) {
@@ -119,7 +123,7 @@ export class Serializer {
       ListWrapper.forEach(map, (val) => { obj.push(this.deserialize(val, type, data)); });
       return obj;
     }
-    if (type == String) {
+    if (type == PRIMITIVE) {
       return map;
     }
 

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

@@ -0,0 +1,78 @@
+import {Injectable} from 'angular2/di';
+import {ListWrapper, Map, MapWrapper} from 'angular2/src/facade/collection';
+import {Serializer} from "angular2/src/web_workers/shared/serializer";
+import {isPresent, Type, FunctionWrapper} from "angular2/src/facade/lang";
+import {MessageBus} from "angular2/src/web_workers/shared/message_bus";
+import {EventEmitter, Promise, PromiseWrapper, ObservableWrapper} from 'angular2/src/facade/async';
+
+@Injectable()
+export class ServiceMessageBrokerFactory {
+  constructor(private _messageBus: MessageBus, protected _serializer: Serializer) {}
+
+  createMessageBroker(channel: string): ServiceMessageBroker {
+    return new ServiceMessageBroker(this._messageBus, this._serializer, channel);
+  }
+}
+
+/**
+ * Helper class for UIComponents that allows components to register methods.
+ * If a registered method message is received from the broker on the worker,
+ * the UIMessageBroker desererializes its arguments and calls the registered method.
+ * If that method returns a promise, the UIMessageBroker returns the result to the worker.
+ */
+export class ServiceMessageBroker {
+  private _sink: EventEmitter;
+  private _methods: Map<string, Function> = new Map<string, Function>();
+
+  constructor(messageBus: MessageBus, private _serializer: Serializer, public channel) {
+    this._sink = messageBus.to(channel);
+    var source = messageBus.from(channel);
+    ObservableWrapper.subscribe(source, (message) => this._handleMessage(message));
+  }
+
+  registerMethod(methodName: string, signature: List<Type>, method: Function, returnType?: Type):
+      void {
+    this._methods.set(methodName, (message: ReceivedMessage) => {
+      var serializedArgs = message.args;
+      var deserializedArgs: List<any> = ListWrapper.createFixedSize(signature.length);
+      for (var i = 0; i < signature.length; i++) {
+        var serializedArg = serializedArgs[i];
+        deserializedArgs[i] = this._serializer.deserialize(serializedArg, signature[i]);
+      }
+
+      var promise = FunctionWrapper.apply(method, deserializedArgs);
+      if (isPresent(returnType) && isPresent(promise)) {
+        this._wrapWebWorkerPromise(message.id, promise, returnType);
+      }
+    });
+  }
+
+  private _handleMessage(map: StringMap<string, any>): void {
+    var message = new ReceivedMessage(map);
+    if (this._methods.has(message.method)) {
+      this._methods.get(message.method)(message);
+    }
+  }
+
+  private _wrapWebWorkerPromise(id: string, promise: Promise<any>, type: Type): void {
+    PromiseWrapper.then(promise, (result: any) => {
+      ObservableWrapper.callNext(
+          this._sink,
+          {'type': 'result', 'value': this._serializer.serialize(result, type), 'id': id});
+    });
+  }
+}
+
+export class ReceivedMessage {
+  method: string;
+  args: List<any>;
+  id: string;
+  type: string;
+
+  constructor(data: StringMap<string, any>) {
+    this.method = data['method'];
+    this.args = data['args'];
+    this.id = data['id'];
+    this.type = data['type'];
+  }
+}