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

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

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

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

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

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

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

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

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

            return body.data.random;
        }
    )
}

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

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

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

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

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

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

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

    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 };
}

利用上面實現的方法，可以很方便的拿到響應流，即 message。

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

    message: Observable; // 暴露給調用者來獲取數據

    constructor() {
        this.connect()
    }

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

        // 由于websocket 不同于http，請求和響應一一對應的很好，所以這里可能需要一定的條件來拿到請求對應的響應。
        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 => /*處理數據邏輯*/),
            retryWhen(error => error.pipe(
                tap(err => console.log(err)),  // 打印一下錯誤，可以換成其它的邏輯。
                delay(500) // 500毫秒后發起重試
            )),
            share() // 將這個流變為hot，至于流的hot 和 cold可能需要多帶帶的文章來解釋。
        );

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

可以看出在響應式的代碼中，很少需要維護一些中間數據狀態，數據都是在流中獲取，轉換和傳遞，訂閱者對數據最好可以實現開箱即用，無需另外的加工。