FlowableProcessor rsocket-js 打字稿

Question

我正在尝试使用 rsocket-websocket-client 构建一个聊天前端。我可以使用requestChannel(new Flowable(source...)) 从前端发送消息并使用requestChannel(new Flowable.just({metatdata})) 接收消息。

我试图使用FlowableProcessor 将requestChannel 的两次调用减少为一次。

在 FlowableProcessor 上找不到 rsocket 的文档。

这是我的尝试：

const processor = new FlowableProcessor(
    new Flowable(source => {
        source.onSubscribe({
            cancel: () => {},
            request: n => {}
        });
        source.onNext({
            metadata: constructMetadataWithChannelId(channelId),
        });
    })
);
sock.requestChannel(processor.map(item => item))
    .subscribe({
        onComplete: () => {
            console.log(
                `complted subscribe`,
            );
        },
        onError: error1 => {
            console.log(
                `subscriber err: ${error1}`,
            );
        },
        onSubscribe: subscription => {
            console.log(
                `onSubscribe`,
            );
            setConnectStatus('connected');
            setChannelIdDone(true);
            subscription.request(1000);
        },
        onNext: (val: any) => {
            const value = JSON.parse(val) as Message;
            console.log(
                `received event from channel: ${JSON.stringify(
                                            value,
                                        )}`,
            );
        }
    })

我了解这是类型问题。无法确定processor.map(item => item) 出错的位置。

TS2345: Argument of type 'IPublisher<unknown>' is not assignable to parameter of type 'Flowable<Payload<Buffer, Buffer>>'.
Type 'IPublisher<unknown>' is missing the following properties from type 'Flowable<Payload<Buffer, Buffer>>': lift, take

Answer 1

错误是微不足道的。 FlawableProcessor 不能使用，因为它没有实现与Flawable 相同的接口。

目前rsocket-js 打磨得不好，有一些瑕疵。其中一些缺陷是类型使用不一致。 AFAIU 应该在所有其他公共接口中使用IPublisher 和ISubscriber 接口。但是为了作者的简单（我猜）它们被替换为Flowable 和Single 类型。

根据源码FlowableProcessor没有扩展Flowable而是实现了IPublisher、ISubscriber和ISubscription接口本身并且没有实现lift和take实现的Flowable方法.所以它不能直接用来代替Flowable，虽然它应该被用作IPublisher。

在您的示例中，我认为没有理由使用 FlowableProcessor。相反，您可以将 Flowable 用作构造 FlowableProcessor 的参数直接传递给 requestChannel 方法：

const requestSource = new Flowable(source => {
    source.onSubscribe({
        cancel: () => {},
        request: n => {}
    });
    source.onNext({
        metadata: constructMetadataWithChannelId(channelId),
    });
});
sock.requestChannel(requestSource.map(item => item))
    ...

如果您真的需要在这段代码中使用FlowableProcessor 处理器，那么您可以强制将其强制转换为Flowable，但它可能会成为未来意外错误的来源：

sock.requestChannel(processor.map(item => item) as any as Flowable)

还请注意您错误地使用了Flowable。当尚未请求数据时，您在订阅时发送数据。这违反了 RSocket 合同。正确的实现应该是这样的：

    let requestsSink: {
        sendRequest(myRequest: unknown): void,
        complete(): void
    };
    const requestsSource = new Flowable((requestsSubscriber) => {
        // Number of the requests requested by subscriber.
        let requestedRequests = 0;
        // Buffer for requests which should be sent but not requested yet.
        const pendingRequests: unknown[] = [];
        let completed = false;

        requestsSink = {
            sendRequest(myRequest: unknown) {
                if (completed) {
                    // It's completed, nobody expects this request.
                    return;
                }
                if (requestedRequests > 0) {
                    --requestedRequests;
                    requestsSubscriber.onNext(myRequest);
                } else {
                    pendingRequests.push(myRequest);
                }
            },
            complete() {
                if (!completed) {
                    completed = true;
                    requestsSubscriber.onComplete();
                }
            },
        };

        requestsSubscriber.onSubscribe({
            cancel: () => {
                // TODO: Should be handled somehow.
            },
            request(n: number) {
                const toSend = pendingRequests.splice(n);
                requestedRequests += n - toSend.length;
                for (const pending of toSend) {
                    requestsSubscriber.onNext(pending);
                }
            }
        });
    });

    sock.requestChannel(requestsSource.map(item => item))
        ...
    
    // Somewhere else the data is provided:
    if (requestsSink != null) {
        requestsSink.sendRequest({});
        requestsSink.sendRequest({});
        requestsSink.sendRequest({});
        requestsSink.sendRequest({});
        requestsSink.complete();
    }