标签: reactive-extensions-js

在请求无扩展(RX)的清澈,风景如画的解释？我问到RX是什么,我想,多亏了提供的答案,我现在有了这个想法.

在引用的问题中,我引用了http://reactive-extensions.github.com/RxJS/中的一句话,其中说:

RxJS是承诺异步的事件.

虽然我认为我得到了RX背后的想法,但我根本没有得到这句话.我甚至不能说出我不理解的是什么.它更像是......我没有看到句子的第一个和第二个半节之间的联系.

对我来说,这句话听起来很重要且令人印象深刻,但我几乎无法判断它是否真实,是否是一个伟大的洞察力,等等......

任何人都可以解释这句话的意思是某些人(像我一样)可以理解谁对所有这些被动的东西都是新手？

我想使用RxJS与同步世界"桥接"事件的异步世界.具体来说,我想创建一个函数,该函数返回在某个时间间隔内收集的事件数组.

我可以创建Observable来做我想要的

var source = Rx.Observable
.interval(100 /* ms */)
.bufferWithTime(1000).take(1)

我可以打印正确的值

var subscription = source.subscribe(
    function (x) {
        console.log('Next: ' + JSON.stringify(x));
    },
    function () {
        console.log('Completed');   
    });

这打印

[0,1,2,3,4,5,6,7,8] 
Completed 

但我想要的是将此数组赋值给变量.从概念上讲,我想要的东西

var collectedDuringSecond = source.toPromise.getValue()

这个想法是getValue会阻塞所以在上面的行完成之后collectDuringSecond将包含[0,1,2,3,4,5,6,7,8]

我正在尝试使用RxJS创建一个可观察到的观察图像.

• 获取一个值并在获得下一个值之前等待一段固定的时间.
• 下一个将是等待期间发出的最后一个值,跳过其余的值.
• 如果等待时间间隔没有发出任何值,则应立即抓取下一个值,如图像的最后一个示例所示.

我正在尝试设置一个简单游戏的更新循环，并考虑到可观察性。顶层组件是一个模型，它接受输入命令并产生更新；和一个视图，它显示接收到的更新，并产生输入。孤立地看，两者都可以正常工作，有问题的部分是将两者放在一起，因为两者都依赖于另一个。

将组件简化为以下内容：

var view = function (updates) {
  return Rx.Observable.fromArray([1,2,3]);
};
var model = function (inputs) {
  return inputs.map(function (i) { return i * 10; });
};

我把事情联系在一起的方式是这样的：

var inputBuffer = new Rx.Subject();
var updates = model(inputBuffer);
var inputs = view(updates);
updates.subscribe(
    function (i) { console.log(i); },
    function (e) { console.log("Error: " + e); },
    function () { console.log("Completed"); }
);
inputs.subscribe(inputBuffer);

也就是说，我添加了一个主题作为输入流的占位符，并将模型附加到该主题上。然后，在构建视图之后，我将实际输入传递给占位符主题，从而关闭循环。

然而，我不禁感到这不是正确的做事方式。为此使用主题似乎有点矫枉过正。有没有办法用 publish() 或 defer() 或类似的方法做同样的事情？

更新：这是一个不太抽象的例子来说明我遇到的问题。下面你会看到一个简单的“游戏”的代码，玩家需要点击一个目标来击中它。目标可以出现在左侧或右侧，每当它被击中时，它就会切换到另一侧。看起来很简单，但我仍然觉得我错过了一些东西......

//-- Helper methods and whatnot
// …

在我的系统中，我有一个源，两个将源映射到新值的“步骤”，然后是组合这两个步骤以创建最终值的总和。该系统的初始运行如我所愿，生成了 3 的单个总和。

var source = new Rx.BehaviorSubject(0);    \nvar stepOne = source.map(function (value) {\n    return value + 1;\n});\nvar stepTwo = source.map(function (value) {\n    return value + 2;\n});    \nvar sum = Rx.Observable.combineLatest(\n    stepOne,\n    stepTwo,\n    function (s1, s2) {\n        console.log(\'calc sum: \' + (s1 + s2));\n        return s1 + s2;\n    }).subscribe(function (sum) {\n    });\n

输出：

> calc sum: 3\n

但是，如果我为源输入一个新值，我会得到如下两个结果：

source.onNext(1);\n\n> calc sum: 4\n> calc sum: 5\n

第一个是新的源值通过系统的一部分时的中间结果\xe2\x80\xa6，然后当所有值完成传播时得到最终结果。

所以我的问题是，推荐的配置方式是什么，以便推送到源中的新值将原子地通过系统并仅生成一个总和结果？

谢谢！

我正在使用 RxJs 来计算特定时间窗口内到达的数据包数量。我的代码基本上是这样的：

var packetSubject = new Rx.Subject();
var packetsInWindow = [];

function startMonitoring() {
    var subscription = packetSubject
        .windowWithTime(1000)
        .select(function(window) {
            window.toArray().subscribe(function(elements) {
                packetsInWindow.push(elements.length);
            });
        })
        .subscribe();
}

function newPacket(packet) {
    packetSubject.onNext(packet);
}

如何使用 Rx TestScheduler 对这段代码进行单元测试？我找不到任何合适的示例来测试主题。

我正试图进入反应式编程.我使用数组函数,如map,filter和减少所有的时间,并且喜欢我可以在不创建状态的情况下进行数组操作.

作为练习,我试图在不引入状态变量的情况下使用RxJS创建可过滤的列表.最后它应该与此类似:

我会知道如何使用天真的JavaScript或AngularJS/ReactJS来实现这一点,但我正在尝试使用RxJS而不创建状态变量:

var list = [
  'John',
  'Marie',
  'Max',
  'Eduard',
  'Collin'
];

Rx.Observable.fromEvent(document.querySelector('#filter'), 'keyup')
  .map(function(e) { return e.target.value; });

// i need to get the search value in here somehow:
Rx.Observable.from(list).filter(function() {}); 

现在如何在我从列表中创建的observable中将搜索值放入我的过滤器函数中？

非常感谢你的帮助!

我正在尝试为.Net学习新的Reactive Extensions框架,这听起来像是我的应用程序的完美解决方案.在研究了示例(并且仍然相当弱的LINQ)之后,我正在努力弄清楚如何利用RX框架来完成下面的任务.

目标是在自定义数据源和GUI之间创建可配置事件"中继".继电器将使用LINQ测试和过滤传入事件,在等待下一个时间间隔时将限定事件缓存在列表中,然后同步到GUI线程并按照接收顺序回放事件.

如何使用RX来协调缓存,过滤和中继多个事件,例如用于添加,更改和从自定义数据源中删除值的事件？

这可能要求很多,但是如何解决这个问题的任何指导都将非常感激.见下面的示例代码......

public delegate void EventDelegateAdd(Thing thing);
public delegate void EventDelegateChange(Thing thing);
public delegate void EventDelegateRemove(Thing thing);

public delegate void EventDelegateBulkChangesStart();
public delegate void EventDelegateBulkChangesEnd();

// The "Things" that are stored in MyCustomDataSource

public class Thing
{
    public string Key { get; set; }
    public string Title { get; set; }
    public object OtherStuff { get; set; }
}

// A custom observable data source with events that indicate when Things are
// added, changed, or removed. …

我需要实现一个行为:

• 点击元素时 - 有一件事发生
• 但是当它被点击并保持一秒以上时,会发生其他事情(例如元素变得可拖动)然后第一个事件永远不会触发

我想我知道如何捕捉点击和保持类型的事件,但如何区分第一和第二？

你能告诉我如何使用这个jsbin做到这一点.我已经完成了"点击,按住并拖动"部分,除了它在拖动元素后仍然触发'click'事件,它不应该.

再次:单击元素 - 一个事件,单击并按住 - 元素可拖动(即使在鼠标向上),再次单击时,它将恢复正常(不可碎片)状态.

我不是在寻找一个简单的解决方案,它必须使用Rx.Observable或至少使用培根的streamEvent对象

谢谢

我正在尝试创建一个流/可观察的......

1. 仅在有订阅者时输出事件
2. 为任何新订阅者提供最新值。

具体的情况是，我需要一个可观察对象，它可以在特定事件发生时进行异步 API 调用，但前提是它有订阅者。我试图避免不必要的 API 调用。

我已经设法创建了一个只有在有这样的订阅者时才会触发的流......

let dataStream = Rx.Observable
   .interval(1000) // Fire an event every second
   .singleInstance() // Only do something when we have subscribers
   .startWith(null) // kick start as soon as something subscribes
   .flatMapLatest(interval => SomeAPI.someDataGet()) // get data, returns a promise

这有效。如果我console.log(...)在SomeAPI.someDataGet方法中，我只会看到它在流有订阅者时触发。我的实现看起来非常好，因为我这样做是为了订阅和取消订阅，这非常适合 React 组件生命周期方法。

let sub1;
sub1 = dataStream.subscribe(x => console.log('sub1', x));
sub1.dispose();

我还希望任何新订阅者在订阅时立即收到最新值。这是我挣扎的地方。如果我这样做...

let sub1, sub2;
sub1 = dataStream.subscribe(x => console.log('sub1', x));

setTimeout( () => {
    sub2 = dataStream.subscribe(x …