refactor(worker): reduce number of rx operator deps, and replace heavyweight rxjs/Subject with custom LiteSubject

Author: alxhub

service-worker/worker/src/plugins/push/index.ts

@@ -1,12 +1,13 @@
 import {
+  LiteSubject,
   Operation,
   Plugin,
   PluginFactory,
   VersionWorker
 } from '@angular/service-worker/worker';
 
-import {Subject} from 'rxjs/Subject';
 import {Observable} from 'rxjs/Observable';
+import {Observer} from 'rxjs/Observer';
 
 interface PushManifest {
   showNotifications?: boolean;
@@ -40,21 +41,25 @@ export function Push(): PluginFactory<PushImpl> {
 export class PushImpl implements Plugin<PushImpl> {
 
   private pushBuffer: any[] = [];
-  private pushSubject: Subject<any> = new Subject<any>();
+  private pushSubject: LiteSubject<any> = new LiteSubject<any>();
   pushes: Observable<any>;
 
   private get pushManifest(): PushManifest {
     return this.worker.manifest['push'] as PushManifest || EMPTY_MANIFEST;
   }
 
   constructor(private worker: VersionWorker) {
-    this.pushes = Observable.create(observer => {
-      this.pushBuffer.forEach(data => observer.next(data));
+    this.pushes = Observable.create((observer: Observer<any>) => {
+      if (this.pushBuffer !== null) {
+        this.pushBuffer.forEach(data => observer.next(data));
+      }
       this.pushBuffer = null;
-      const sub = this.pushSubject.subscribe(observer);
+      const sub = this.pushSubject.observable.subscribe(observer);
       return () => {
         sub.unsubscribe();
-        this.pushBuffer = [];
+        if (!this.pushSubject.hasSubscribers) {
+          this.pushBuffer = [];
+        }
       };
     });
   }

service-worker/worker/src/worker/logging.ts

@@ -1,5 +1,5 @@
 import {Observable} from 'rxjs/Observable';
-import {Subject} from 'rxjs/Subject';
+import {LiteSubject} from './rxjs';
 
 export enum Verbosity {
   INFO = 1,
@@ -16,7 +16,7 @@ interface LogEntry {
 let logBuffer: LogEntry[] = [];
 
 // Subject which will be used to broadcast log messages once there's a subscriber.
-let logSubject: Subject<LogEntry> = null;
+let logSubject: LiteSubject<LogEntry> = null;
 
 let logLevel = Verbosity.INFO;
 
@@ -25,7 +25,7 @@ let logLevel = Verbosity.INFO;
 let logStream = Observable.create(observer => {
   // Create the subject if it doesn't exist already.
   if (logSubject === null) {
-    logSubject = new Subject<LogEntry>();
+    logSubject = new LiteSubject<LogEntry>();
   }
 
   // An Observable representing buffered messages. Initialized to empty.
@@ -40,7 +40,7 @@ let logStream = Observable.create(observer => {
   // Combine (possibly empty) buffered messages with the subject, and pipe them to the
   // subscriber.
   return buffered
-    .concat(logSubject)
+    .concat(logSubject.observable)
     .subscribe(observer);
 });
 

service-worker/worker/src/worker/rxjs.ts

@@ -1,16 +1,11 @@
-import 'rxjs/add/observable/combineLatest';
 import 'rxjs/add/observable/concat';
 import 'rxjs/add/observable/defer';
 import 'rxjs/add/observable/empty';
 import 'rxjs/add/observable/from';
 import 'rxjs/add/observable/fromEvent';
 import 'rxjs/add/observable/fromPromise';
-import 'rxjs/add/observable/merge';
 import 'rxjs/add/observable/of';
-import 'rxjs/add/observable/timer';
 
-import 'rxjs/add/operator/cache';
-import 'rxjs/add/operator/catch';
 import 'rxjs/add/operator/concat';
 import 'rxjs/add/operator/concatMap';
 import 'rxjs/add/operator/do';
@@ -20,14 +15,43 @@ import 'rxjs/add/operator/ignoreElements';
 import 'rxjs/add/operator/let';
 import 'rxjs/add/operator/map';
 import 'rxjs/add/operator/mergeMap';
-import 'rxjs/add/operator/publishReplay';
 import 'rxjs/add/operator/reduce';
-import 'rxjs/add/operator/share';
 import 'rxjs/add/operator/switchMap';
 import 'rxjs/add/operator/take';
 import 'rxjs/add/operator/toPromise';
 
 import {Observable} from 'rxjs/Observable';
+import {Observer} from 'rxjs/Observer';
+
+export class LiteSubject<T> {
+
+  observable: Observable<T>;
+  private subscribers: Observer<T>[] = [];
+
+  constructor() {
+    this.observable = Observable.create((obs: Observer<T>) => {
+      this.subscribers.push(obs);
+      return () => {
+        let index = this.subscribers.indexOf(obs);
+        if (index >= 0) {
+          this.subscribers.splice(index, 1);
+        }
+      };
+    });
+  }
+
+  next(value: T): void {
+    this.subscribers.forEach(obs => obs.next(value));
+  }
+
+  complete(): void {
+    this.subscribers.forEach(obs => obs.complete());
+  }
+
+  get hasSubscribers(): boolean {
+    return this.subscribers.length > 0;
+  }
+}
 
 export function doAsync<T>(fn: (T) => Observable<any>): any {
   return (obs: Observable<T>) => obs