javascript - 我可以使用RxJS进行面板间通信吗？

Question

注意:我提到了 RxJS，但任何反应式库都可以(Bacon、Kefir、Most 等)。我的上下文是 AngularJS，但解决方案可能是独立的(或多或少)。

我的问题/任务:我们有一个 AngularJS 应用程序，我们希望有侧面板(和中央面板)，每个侧面板可能有可以添加、删除等的子面板。

这些面板之间必须进行通信:不仅是父子交换，还包括任何面板到任何面板(侧面/子/中央...)。

我觉得经典的 Angular 方式(事件总线:$emit/$broadcast/$on)在这里是相当不够的。即使对于简单的父/子通信，当父级在启动时触发事件但子级还没有听时，我也遇到了问题。用 $timeout 解决了这个问题，但这很脆弱。另外，为了让两个子进程进行通信，他们将消息发送给进行传输的父进程，这很笨拙。

我认为这个问题是在项目中引入响应式编程的机会(在非常早期的阶段，这里不会造成破坏)，但如果我已经阅读了很多有关该主题的内容，那么到目前为止我的经验还很少。

因此我的问题是:有没有一种干净的方法来使用 FRP 来管理这个问题？

我正在考虑建立一个服务(因此是一个单例)来监听新的面板、广播可观察量、接受观察者等。但我不太确定如何做到这一点。

与其重新发明轮子，我更愿意问这个问题是否已经解决，没有太多的耦合，没有不灵活等等。

注意:如果一个好的解决方案不使用 FRP，那也没关系! :-)

谢谢。

Answer 1

感谢@xgrommx和@user3743222的评论，以及这本优秀的RxJS书籍，我能够实现我的目标。

我的实验 Playground 位于http://plnkr.co/edit/sGx4HH?p=preview

通信中心服务的主体是(精简):

var service = {};

service.channels = {};

/**
 * Creates a new channel with given behavior (options) and returns it.
 * If the channel already exists, just returns it.
 */
service.createChannel = function(name, behavior)
{
  checkName(name);
  if (_.isObject(service.channels[name]))
    return service.channels[name];

  behavior = behavior || {};
  _.defaults(behavior, { persistent: false });

  if (behavior.persistent)
  {
    _.defaults(behavior, { bufferSize: null /* unlimited */, windowSize: 5000 /* 5 s */ });
    service.channels[name] = new Rx.ReplaySubject(behavior.bufferSize, behavior.windowSize);
  }
  else
  {
    service.channels[name] = new Rx.Subject();
  }
  return service.channels[name];
};

/**
 * Returns the channel at given name, undefined if not existing.
 */
service.getChannel = function(name)
{
  checkName(name);
  return service.channels[name];
};

/**
 * Destroys an existing channel.
 */
service.destroyChannel = function(name)
{
  checkName(name);
  if (!_.isObject(service.channels[name]))
    return;

  service.channels[name].dispose();
  service.channels[name] = undefined;
};

/**
 * Emits an event with a value.
 */
service.emit = function(name, value)
{
  checkName(name);
  if (!_.isObject(service.channels[name]))
    return;

  service.channels[name].onNext(value);
};


function checkName(name)
{
  if (!_.isString(name))
    throw Error('Name of channel must be a string.');
}

return service;

我的使用方式如下:

angular.module('proofOfConceptApp', [ 'rx' ])
.run(function (CommunicationCenterService)
{
  CommunicationCenterService.createChannel('Center', { persistent: true });
  CommunicationCenterService.createChannel('Left');
  CommunicationCenterService.createChannel('Right');
})
.controller('CentralController', function ($scope, $http, rx, observeOnScope, CommunicationCenterService) 
{
  var vm = this;

  CommunicationCenterService.getChannel('Right')
    .safeApply($scope, function (color) 
    {
      vm.userInput = color;
    })
    .subscribe();

  observeOnScope($scope, function () { return vm.userInput; })
    .debounce(1000)
    .map(function(change)
    {
      return change.newValue || "";
    })
    .distinctUntilChanged() // Only if the value has changed
    .flatMapLatest(searchWikipedia)
    .safeApply($scope, function (result) 
    {
      // result: [0] = search term, [1] = found names, [2] = descriptions, [3] = links
      var grouped = _.zip(result.data[1], result.data[3]);
      vm.results = _.map(grouped, function(r)
      {
        return { title: r[0], url: r[1] };
      });
      CommunicationCenterService.emit('Center', vm.results.length);
    })
    .subscribe();

  function searchWikipedia(term) 
  {
    console.log('search ' + term);
    return rx.Observable.fromPromise($http(
      {
        url: "http://en.wikipedia.org/w/api.php?callback=JSON_CALLBACK",
        method: "jsonp",
        params: 
        {
          action: "opensearch",
          search: encodeURI(term),
          format: "json"
        }
      }));             
  }

  CommunicationCenterService.emit('Center', 42); // Emits immediately
})
.controller('SubController', function($scope, $http, rx, observeOnScope, CommunicationCenterService) 
{
  var vm = this;

  vm.itemNb = $scope.$parent.results;//.length;

  CommunicationCenterService.getChannel('Left')
    .safeApply($scope, function (toggle) 
    {
      vm.messageFromLeft = toggle ? 'Left is OK' : 'Left is KO';
    })
    .subscribe();

  CommunicationCenterService.getChannel('Center')
    .safeApply($scope, function (length) 
    {
      vm.itemNb = length;
    })
    .subscribe();
})
.controller('LeftController', function($scope, $http, rx, observeOnScope, CommunicationCenterService) 
{
  var vm = this;

  vm.toggle = true;

  vm.toggleValue = function ()
  {
    CommunicationCenterService.emit('Left', vm.toggle);
  };

  observeOnScope($scope, function () { return vm.toggle; })
    .safeApply($scope, function (toggleChange) 
    {
      vm.valueToDisplay = toggleChange.newValue ? 'On' : 'Off';
    })
    .subscribe();

  CommunicationCenterService.getChannel('Center')
    .safeApply($scope, function (length) 
    {
      vm.messageFromCenter = 'Search gave ' + length + ' results';
    })
    .subscribe();
})
.controller('RightController', function($scope, $http, rx, observeOnScope, CommunicationCenterService) 
{
  var vm = this;

  var display = { red: 'Pink', green: 'Aquamarine', blue: 'Sky' };

  vm.color = { value: "blue" }; // Initial value

  observeOnScope($scope, function () { return vm.color.value; })
    .tap(function(x) 
    { 
      CommunicationCenterService.emit('Right', vm.color.value);
    })
    .safeApply($scope, function (colorChange) 
    {
      vm.valueToDisplay = display[colorChange.newValue];
    })
    .subscribe();

  CommunicationCenterService.getChannel('Left')
    .safeApply($scope, function (toggle) 
    {
      vm.messageFromLeft = toggle ? 'Left is on' : 'Left is off';
    })
    .subscribe();
})
;

我必须预先创建 channel (在 .run 中)，否则监听未创建的 channel 会崩溃。不确定我是否会解除限制...

这只是一个草稿，可能很脆弱，但到目前为止它达到了我的预期。

我希望它对某人有用。

[编辑]我更新了我的 plunk。

拍摄 2:http://plnkr.co/edit/0yZ86a我清理了 API 并创建了一个隐藏 channel 服务的中间服务。 channel 是预先创建的。我使订阅的清理变得更加容易。

第 3 段:http://plnkr.co/edit/UqdyB2我向 channel 添加主题(受 Postal.js 启发)，从而实现更精细的沟通和更少的主题。