我可以使用 RxJS 进行 inter-panel 通信吗？

Question

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

我的问题/任务：我们有一个 AngularJS 应用程序，我们想要有侧面板（和一个中央面板），每个侧面板可能有 sub-panels 可以添加、删除等

这些面板必须在它们之间进行通信：不仅 parent-child 交换，而且任何面板到任何面板（侧/副/中央...）。

我觉得经典的 Angular 方式（事件总线：$emit / $broadcast / $on）在这里不太合适。即使是简单的 parent / child 通信，当 parent 在启动时触发事件，但 child 尚未收听时，我也遇到了问题。用 $timeout 解决了这个问题，但这很脆弱。另外，为了让两个children进行通信，他们发送给发送的parent，这是笨拙的。

我认为这个问题是在项目中引入响应式编程的机会（在很早的阶段，不会在这里造成破坏），但是如果我已经阅读了很多关于该主题的文章，那么到目前为止我还没有什么经验.

因此我的问题是：是否有一种干净的方法来使用 FRP 来管理它？

我正在考虑设置一个服务（因此是一个单例）来监听新的面板、广播可观察对象、接受观察者等。但我不太确定该怎么做。

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

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

谢谢。

Answer 1

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

我的实验游乐场在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();
})
;

我必须预先创建频道（在 .运行中），否则收听未创建的频道会崩溃。不确定我是否会解除限制...

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

希望对大家有用。

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

在拍摄 2 中：http://plnkr.co/edit/0yZ86a 我清理了 API 并制作了一个隐藏通道服务的中介服务。频道是预先创建的。我让清理订阅变得更容易了。

在 Take 3 中：http://plnkr.co/edit/UqdyB2 我将主题添加到频道（受 Postal.js 启发）允许更好的交流和更少的主题。

我可以使用 RxJS 进行 inter-panel 通信吗？

Can I use RxJS to do inter-panel communication?

javascript

frp

reactive-programming

rxjs

angularjs