RxJS入门（2）---Observable的介绍

创建Observable

有好多种方式创建Observable，但是最常见的方式之一就是通过create操作符。在Rx.Observalbe对象的create操作符中需要传递一个回调函数，这个回调函数中需要传递一个订阅者参数（Observer，相当于（1）中传统观察者模式中的listener）。在这个方法中定义了Observable如何发射值，下面就举个例子来说明：

var observable = Rx.Observable.create(function(observer) {
observer.onNext('Simon');
observer.onNext('Jen');
observer.onNext('Sergi');
observer.onCompleted(); // 成功完成
});

当我们订阅这个Observable，它通过调用onNext（）发射三个字符串给他的订阅者（监听者），最后调用onCompleted()标识这个序列完成了，但是我们是如何准确的订阅一个Observable了？通过Observers（也就是订阅者)。（补充：Observable可翻译为，可被订阅者，但是感觉怪怪的，就直译了）

与observer的关联

observers监听observable，不论何时在Observable中发生了什么事件，它会调用它的observers中相应的方法。

observers有三个方法：onNext，onCompleted，onError。

onNext：相当于在（1）中observer模式中的Update方法，当observable发射了一个新值它就会被调用。注意到那个名称是如何反应到我们订阅的序列中的，而不是仅仅分离的值。

onCompleted：再没有任何有效数据的信号，在onCompleted被调用后，之后的onNext调用将会无效。

onError：当observable中发生了错误后将会被调用，在onError被调用后，之后的onNext调用将会无效。

下面可以举个例子说明我们是如何创建observer的：

var observer = Rx.Observer.create(
function onNext(x) { console.log('Next: ' + x); },
function onError(err) { console.log('Error: ' + err); },
function onCompleted() { console.log('Completed'); }
);

在Rx.Observer对象的create方法中有onNext，onCompleted，onError三个方法，并且他返回的也是一个Observer的对象实例。这三个方法是可选的，你根据你的需求来决定。例如，我们订阅一个无限的序列（比如按钮的点击事件，用户可以一直保持点击），onCompleted函数将永不会调用；再如我们确定某个序列将不会发生错误（例如数字数组的observable），我们也是不需要onError方法的。

通过observable使用ajax

如何创建observable获取远程内容，我们将会使用Rx.Observable.create来封装XMLHttpRequest对象：

function get(url) {
return Rx.Observable.create(function(observer) {
// Make a traditional Ajax request
var req = new XMLHttpRequest();
req.open('GET', url);
req.onload = function() {
if (req.status == 200) {
// If the status is 200, meaning there have been no problems,
// Yield the result to listeners and complete the sequence
observer.onNext(req.response);
observer.onCompleted();
}
else {
// Otherwise, signal to listeners that there has been an error
observer.onError(new Error(req.statusText));
}
};
req.onerror = function() {
observer.onError(new Error("Unknown Error"));
};
req.send();
});
}
// Create an Ajax Observable
var test = get('/api/contents.json');

在上面的代码中，使用create封装XMLHttpRequest的get函数，如果HTTP GET请求成功了，我们将发送它的内容并完成那个序列（我们的observable讲仅仅发送一个结果），否则我们将发射一个错误。在最后一行，我是使用一个特定的url去调用这个函数。这将会产生一个observable，但是它不会发出任何请求，这个很重要：observable不会做任何事，直到最少有一个observer订阅它，所以让我们接着如下：

// Subscribe an Observer to it
test.subscribe(
function onNext(x) { console.log('Result: ' + x); },
function onError(err) { console.log('Error: ' + err); },
function onCompleted() { console.log('Completed'); }
);

首先我们必须注意的是：我们不是很明确地创建一个observer像上面的代码中那样，大部分的时候我们会使用一个局部的版本（在observable中我们调用subscribe操作符，使用observer的三种可选函数：onNext，onCompleted，onError)。subscribe设置所有的动态的设置所有的事情。在订阅（subscription）之前，我们仅仅定义声明observable和observer的交互，他们只会在我们调用subscribe后才开始交互。

一直存在的操作符（operator）

在rxjs中，改变或者查询序列（sequence）的方法叫做操作符。操作符可以在静态的Rx.Observalbe对象或者是Observable的实例中找到。在上面的例子中，create就是一个操作符。

create是一个很好的选择当我们创建指定的Observable，但是Rxjs提供了大量的其他的操作符使得从普通资源创建Observable变得容易了。

让我们再次重新看下之前的example，就像这个ajax请求这个普通的操作在这里就提供了一个操作符以供我们使用。Rxjs DOM library提供了好多方法根据dom相关的资源去创建Observable。所以我们就可以使用Rx.DOM.Request.get来处理一个get请求了。我们的代码可以变得如下了：

Rx.DOM.get('/api/contents.json').subscribe(
function onNext(data) { console.log(data.response); },
function onError(err) { console.error(err); }
);

这一点代码的功能和我们之前的功能是一样的，但是我们没有有必要围绕这个XMLHttpRequest创建一个包装器（wrapper）；这里已经有了。请注意，这次我们在这里遗漏了onCompleted这个回调函数，因为我们没有计划当Observable完成的时候要去交互（作出发应）。这个序列仅仅产生一个结果，并且这个它已经在onNext函数中被使用了。

Rxjs有一系列的操作符，事实上，这也是它的主要优点。

在rxjs编码中，我们要力求所有的数据都在Observable中，而不仅仅是来自异步资源的。这样做很容易组合来源不同的各种数据，像回到函数返回的存在的数组，或者是某些用户的事件触发的XMLHttpRequest请求。

例如，一个数组的某些值需要在另外的一些组合数据中被使用，最好的方式就是把这个数组放到Observable中（显而易见，如果这个数组仅仅是一个组合用不到的中间变量，完全没有必要这样做）。在接下来的文章中，你将会学到在哪些情况下需要改变数据的类型到Observable中。

rxjs提供了操作符去创建大部分JavaScript数据类型的Observable。让我们走下你一直在使用的大部分的：arrays、events、callbacks。

根据Arrays创建Observable

我们可以使用任何array类似或者Iterable对象，通过多功能的from操作符转化为Observable。from操作符使用一个array作为参数，并且返回一个Observable（发射array的每一个元素）。

Rx.Observable
.from(['Adrià', 'Jen', 'Sergi'])
.subscribe(
function(x) { console.log('Next: ' + x); },
function(err) { console.log('Error:', err); }
function() { console.log('Completed'); }
);

from操作符伴随着fromEvent，在RxJS中这是最方便和使用频率最高的操作符。

通过js的Event创建Observable

当我们把一个event转化为了一个Observable，它变成了一个很好的值，可以被用来组合和通过。例如，这里有一个Observable，它可以发射鼠标指针移动到任何位置的坐标：

var allMoves = Rx.Observable.fromEvent(document, 'mousemove');
allMoves.subscribe(function(e) {
console.log(e.clientX, e.clientY);
});

把event转化为Observable解开了event本身的强制约束。更重要的是，我们可以基于原始的Observable创建一个新的Observable，并且这些新的Observable是独立的，可以别用于其他的任务：

var movesOnTheRight = allMoves.filter(function(e) {
return e.clientX > window.innerWidth / 2;
});
var movesOnTheLeft = allMoves.filter(function(e) {
return e.clientX < window.innerWidth / 2;
});
movesOnTheRight.subscribe(function(e) {
console.log('Mouse is on the right:', e.clientX);
});
movesOnTheLeft.subscribe(function(e) {
console.log('Mouse is on the left:', e.clientX);
});

在上面的代码中，我们根据原始的allMoves创建了两个Observable，这些从初始Observable生成的特定Observable包含仅仅一个过滤器选项：movesOnTheRight包含屏幕右边发生的的event事件，movesOnTheLeft包含屏幕左边发生的event事件。他们都没有改变原始的Observable：allMoves依然保持发射所有的鼠标移动。Observable是不可改变的，每一涉及到它的操作符都是创建的一个新的Observable。

根据Callback函数创建Observable

如果使用第三方的js库基于callback编写代码进行交互总有好多意外。使用fromCallback和fromNodeCallback两个函数我们可以把我们的Callback转换为Observable。Node.js总是遵循着调用回调函数首先使用一个error的参数告诉回调函数，发生了错误。当我们使用fromNodeCallback去创建指定的Node.js风格的回调函数：

var Rx = require('rx'); // Load RxJS
var fs = require('fs'); // Load Node.js Filesystem module
// Create an Observable from the readdir method
var readdir = Rx.Observable.fromNodeCallback(fs.readdir);
// Send a delayed message
var source = readdir('/Users/sergi');
var subscription = source.subscribe(
function(res) { console.log('List of directories: ' + res); },
function(err) { console.log('Error: ' + err); },
function() { console.log('Done!'); });

在上面的代码中，我们创建一个Observable readdir 使用了Node.js的 fs.readdir方法。fs.readdir接受一个目录路径和一个回调函数delayMsg，一旦这个目录内容重置回调函数就会被调用。

我们使用readdir在同样的参数时我们传给原始的fs.readdir,减去了那个回调函数。它返会一个Observable，可以合适的使用onNext，onError，onCompleted，当我们订阅一个Observer到它的时候。