AngularJS Rx replacement to $broadcast | $emit | $on

CommunicationService.js

/**
 * Created by kdavin on 24/11/2015.
 */
import {Service, Module} from '../../../decorators';
import Rx from 'rx';

@Module({
  name : "app.common.services.communication"
})
@Service("CommunicationService")
class CommunicationService {

  constructor () {
    this.subjects = new Map();
  }

  $on(subject, $scope) {
    let observable = this.$$getOrCreateSubject(subject);
    $scope.$on(subject, (_, data) => observable.onNext(data));

    return this.$sub(subject, $scope, true);
  }

  $sub(subject, $scope, needApply = false) {
    return {
      $do : (func) => {
        let subscriber = needApply ? (data) => $scope.$apply(() => func(data)) : data => func(data);
        let dispose = this.$$getOrCreateSubject(subject).subscribe(subscriber);
        this.$$declareDestruction ($scope, dispose);
      }
    };
  }

  $$declareDestruction($scope, dispose) {
    $scope.$on ('$destroy', () => {
      dispose.dispose ();
      this.$$removeUnusedTopic();
    });
  }

  $emit(subject) {
    return {
      $send : (val) => this.$$getOrCreateSubject(subject).onNext(val)
    };
  }

  $$getOrCreateSubject(subject) {
    this.subjects.has(subject) ? this.subjects.get(subject) : this.subjects.set(subject, new Rx.ReplaySubject(0));
    return this.subjects.get(subject);
  }

  $$removeUnusedTopic() {
    this.subjects.forEach((value, key, map) => {
      if (!value.hasObservers())
        map.delete(key);
    });
  }
}

export default CommunicationService;