rx-java - 在不同的计算上下文的情况下如何避免订阅订阅

Question

我写了一个程序，它从套接字 io 接收消息，然后播放媒体文件。因此，当程序收到下一条消息时，媒体播放器会停止并播放另一个消息。我SocketIO将消息流实现为Observable<String>和MediaPlayeras Complatable。所以代码包含“订阅中的订阅”：

class Main {

    Disposable d = null;

    public void start() {    
        ScoketIO.connect("localhost:9000").subscribe(event -> {
            if (d != null)
                d.dispose();
            d = MediaPlayer.play(event).subscribe();
        }
    }
}

也许有办法避免“订阅订阅”？删除所有可空值并使代码更具“功能性”也很好

更新

似乎switchMap适用于那个简单的例子：

ScoketIO.connect("localhost:9000")
                .switchMapCompletable(event -> MediaPlayer.play(event))
                .subscribe();

当新事件到来时，媒体播放器被释放（停止播放）。但是如果媒体播放器出现故障，所有链都会失败 - 套接字连接也会失败，但事实并非如此。添加retry到媒体播放器也会触发重新连接。“不同的计算环境”是什么意思。

标签： rx-javareactive-programmingrx-java2

Answer 1

如果您可以在 MediaPlayer 中构建 Completable stop 方法，您可以执行以下操作：

        SocketIO.connect("localhost:9000")
                .flatMapCompletable(event ->
                        MediaPlayer.stop()
                                .andThen(MediaPlayer.play(event))
                ).subscribe();