Rxjs写个简单的http请求类

我们先来写一个Rx.js的简单Demo

let observable$ = Observable.create( ( observer ) => {
    // internal interaction with observable
    observer.next( 1 );
    observer.next( 2 );
    observer.next( 3 );
    //==> observer.error( 'error-message' );
    //==> observer.complete();
} );
let observer = {
    next: data => console.log( '[data] => ', data ),
    complete: data => console.log( '[complete]' ),
};
let subscription = observable$.subscribe( observer );
// outside interaction with observable
subscription.next( 4 );
subscription.next( 5 );
subscription.complete();

Observable对象可以通过Observable.create创建并且接收一个observer参数.我们可以在observer这个对象上面调用next,error或者complete。
我们可以在Observable.create里面执行一个简单的http请求并且当成功时调用next或者失败时调用error方法。

import { Observable } from 'rxjs';
import axios from 'axios';
let observable$ = Observable.create( ( observer ) => {
    axios.get( 'https://www.baidu.com' )
    .then( ( response ) => {
        observer.next( response.data );
        observer.complete();
    } )
    .catch( ( error ) => {
        observer.error( error );
    } );
} );
let subscription = observable$.subscribe( {
    next: data => console.log( '[data] => ', data ),
    complete: data => console.log( '[complete]' ),
} );

下来我们看看该怎样取消订阅

let subscription = observable$.subscribe( {
    next: data => console.log( '[data] => ', data ),
    complete: data => console.log( '[complete]' ),
} );
setTimeout( () => {
    subscription.unsubscribe();
} );

这一次，我们没有看到在console中输出任何内容，当我们订阅时，一个ajax请求已经发送出去。
但是我们在获得response之前，就取消了订阅。
请求执行成功并且已经调用了next和call方法，但是由于unsubscribe()调用，observer停止监听observable。
但是这样很容易造成资源浪费，虽然没有监听了，可是后台仍然发送了ajax请求。下来我们看看该怎么解决这个问题。

import { Observable, Subscriber } from 'rxjs';
let observable$ = new Observable( ( observer ) => {
    return new Subscriber( observer );
} );

Subscriber实现了observer接口并且集成了Subscription类。言外之意就是有了next,error,complete和unsubscribe方法。
下面我们写个类来继承Subscriber

class AxiosSubscriber extends Subscriber {
    constructor( observer ) {
        super( observer );

        observer.next( 'HELLO' );
        observer.complete();
    }
   unsubscribe() { // 在complete和error后会自动触发unsubscribe
        console.log( 'unsubscribed' );
        super.unsubscribe();
    }
}
let observable$ = new Observable( ( observer ) => {
    return new AxiosSubscriber( observer );
} );
let subscription = observable$.subscribe( console.log );

为了取消axios的底层XHR请求，我们将使用axios-cancel模块。该模块在axios上增加了cancel原型方法，该方法接受requestId参数。我们需要在发出axios请求时创建并传递requestId字符串。

import React from 'react';
import { fromEvent, Observable, Subscriber } from 'rxjs';
import { switchMap } from 'rxjs/operators';
import axios from 'axios';
const { CancelToken } = axios;

class AxiosSubscriber extends Subscriber {
    constructor(observer) {

        super(observer);
        // create sample request id
        this.requestId = Math.random() + '-xhr-id';
        this.cancelSource = CancelToken.source()

        // XHR abort pointer
        this.aborted = false;
        this.token = this.cancelSource.token

        // make axios request on subscription
        axios.get('https://jsonplaceholder.typicode.com/users', { cancelToken: this.token })
            .then((response) => {
                observer.next(response.data);
                observer.complete();
            })
            .catch((error) => {
                observer.error(error);
            });
    }

    unsubscribe() {
        super.unsubscribe();
        // cancel XHR
        if (this.aborted === false) {
            this.cancelSource.cancel('取消上次请求')
            this.aborted = true;
        }
    }
}

export class Click extends React.Component {
    constructor() {
        super()
        this.state = {
            clickNum: 0
        }
    }
    render() {
        let observable$ = new Observable((observer) => {
            return new AxiosSubscriber(observer);
        });

        fromEvent(document.getElementById('typeahead-input'), 'input')
            .pipe(
                switchMap(() => observable$)
            )
            .subscribe(console.log);

        return <div>
        </div>
    }
}