摘要：響應(yīng)式組件上節(jié)中錯誤的把添加到了中，實際中它返回的是一個，先調(diào)整過來。將這兩條流合并成一條請求流。利用上面實現(xiàn)的方法，可以很方便的拿到響應(yīng)流，即。由于不同于，請求和響應(yīng)一一對應(yīng)的很好，所以這里可能需要一定的條件來拿到請求對應(yīng)的響應(yīng)。

上回搭建了一個組件以及它所依賴的服務(wù)的基本結(jié)構(gòu)，這節(jié)接著它繼續(xù)。另外從本節(jié)開始，統(tǒng)一采用rxjs6的風格，6和5在寫法上最大的不同就是棄用鏈式調(diào)用，而采用pipe的方法，當然也有一些其它的變更，請自行翻閱文檔。

上節(jié)中錯誤的把generateRandomCode 添加到了 subscription中，實際中它返回的是一個Observable，先調(diào)整過來。

initialModel() {
        this.randomCode = this.auth.generateRandomCode(this.generateCode$);
}

刪除 launch方法中的 .add(this.auth.generateRandomCode(this.generateCode$));

驗證碼變更的邏輯有兩處，第一是程序設(shè)定好的時間周期到達后，第二是用戶點點擊時。我們的組件中已經(jīng)設(shè)置了一個subject來獲取用戶的輸入，并且已經(jīng)傳給了服務(wù)，現(xiàn)在我們來完善服務(wù)中獲取驗證碼的邏輯，也就是那個generateRandomCode方法。

generateRandomCode(signal: Observable): Observable {
    // 每30秒向后臺獲取一次驗證碼
    const period = timer(0, 1000 * 30) // 第一個參數(shù)延遲時間，第二個參數(shù)間隔周期。
            .pipe(
                mapTo(true)
            );

    // 將這兩條流合并成一條請求流。
    const request = merge(signal, period);

    // 修改返回，當請求流中產(chǎn)生數(shù)據(jù)后，向服務(wù)器請求并將結(jié)果返回。
    return request.pipe(
        switchMapTo(this.http.get(url)),
        map((res: Response)) => {
            // 假設(shè)數(shù)據(jù)保存在random 字段下
            const body = res.json()；

            return body.data.random;
        }
    )
}

隨便碼請求的過程就完成了，當然還可以添加其它邏輯，如限制用戶10秒內(nèi)最多可以獲取一次，可以給傳入的 signal 加 debounce time

const signal2 = signal.pipe(
    debounceTime(10 * 1000)
)

另外還有對于錯誤的處理等，可以參照之前 前端大耍 的那篇 《Angular Http 請求出錯后重試》。

既然說到了http的失敗重試，其實websocket也一樣。

首先我們實現(xiàn)一個建議websocket連接的方法，代碼如下：

// url 和 protocols 不需要解釋; input: 請求數(shù)據(jù)的流，通過它來向服務(wù)端發(fā)送數(shù)據(jù)。
function connect(url, input, protocols) {

    const connectionStatus = new BehaviorSubject(0); // 用來查看當前連接的狀態(tài)。

    const messages = new Observable(function (observer) {
        
        const socket = new WebSocket(url, protocols);

        const inputSubscription;

        const open = false;

        const forcedClose = false;

        const closed = () => {
            if (!open) return;
            connectionStatus.next(connectionStatus.getValue() - 1);
            open = false;
        };

        socket.onopen = () => {
            open = true;
            connectionStatus.next(connectionStatus.getValue() + 1);
            inputSubscription = input.subscribe( data => socket.send(data));
        };

        socket.onmessage = message => observer.next(message.data);

        socket.onerror = error => {
            closed();
            observer.error(error);
        };

        socket.onclose = event => {
            closed();
            if (forcedClose)
                observer.complete();
            else
                observer.error(new Error(event.reason));
        };

        return function () {
            forcedClose = true;
            
            if (inputSubscription)
                inputSubscription.unsubscribe();
            if (open) {
                closed();
                socket.close();
            }
        };
    });

    return { messages: messages, connectionStatus: connectionStatus };
}

利用上面實現(xiàn)的方法，可以很方便的拿到響應(yīng)流，即 message。

@Injectable()
export class WebSocketService {
    inputStream: Subject = new Subject(); // 和上面寫好的方法進行通信

    message: Observable; // 暴露給調(diào)用者來獲取數(shù)據(jù)

    constructor() {
        this.connect()
    }

    // 暴露給其它服務(wù)或組件來發(fā)送數(shù)據(jù)，同時獲得請求對應(yīng)的響應(yīng)。
    send(data) {
        this.inputStream.next(data);

        // 由于websocket 不同于http，請求和響應(yīng)一一對應(yīng)的很好，所以這里可能需要一定的條件來拿到請求對應(yīng)的響應(yīng)。
        return this.message.pipe(
            filter(message => message.flag === data.flag)
        );
    }

    private connect(): void {
        if(this.message) return;

        const { message, connectStatus } = connect(youUrl, this.inputSteam);

        this.message = message.pipe(
            map(response => /*處理數(shù)據(jù)邏輯*/),
            retryWhen(error => error.pipe(
                tap(err => console.log(err)),  // 打印一下錯誤，可以換成其它的邏輯。
                delay(500) // 500毫秒后發(fā)起重試
            )),
            share() // 將這個流變?yōu)閔ot，至于流的hot 和 cold可能需要多帶帶的文章來解釋。
        );

        connectStatus.subscribe(status => console.log("當前連接的狀態(tài):" + status))
    }
}

可以看出在響應(yīng)式的代碼中，很少需要維護一些中間數(shù)據(jù)狀態(tài)，數(shù)據(jù)都是在流中獲取，轉(zhuǎn)換和傳遞，訂閱者對數(shù)據(jù)最好可以實現(xiàn)開箱即用，無需另外的加工。