从命令式到响应式（七）

摘要：响应式组件上节中错误的把添加到了中，实际中它返回的是一个，先调整过来。将这两条流合并成一条请求流。利用上面实现的方法，可以很方便的拿到响应流，即。由于不同于，请求和响应一一对应的很好，所以这里可能需要一定的条件来拿到请求对应的响应。

上回搭建了一个组件以及它所依赖的服务的基本结构，这节接着它继续。另外从本节开始，统一采用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))
    }
}

可以看出在响应式的代码中，很少需要维护一些中间数据状态，数据都是在流中获取，转换和传递，订阅者对数据最好可以实现开箱即用，无需另外的加工。