rx-java - RxJava:取消订阅可从另一个异步生产中观察到的异步

Question

取消订阅另一个异步Observable的最佳(干净)方法是什么？
说，我有一个演示文稿，显示了用户的聊天列表。
每个“聊天”项目应显示其自身的最新消息。
这是原始代码说明:

fun AsyncInsideAsync() {
    /**
     * A chat that 'transmits' latest message
     */
    class Chat(val name: Int) {
        val lastMessage: PublishSubject<String> = PublishSubject.create()

        fun postMessage(message: Int) {
            lastMessage.onNext("Chat: $name, Message: $message")
        }
    }

    // List of chats for user.
    val chats: PublishSubject<Chat> = PublishSubject.create()

    ///////////////////////////////////
    //        ORIGINAL SEQUENCE      //
    ///////////////////////////////////
    val sequence = chats.flatMap { it.lastMessage }

    val subscriber = TestSubscriber<String>()
    sequence.subscribe(subscriber)

    // User has single chat in a chat-list
    val chat1 = Chat(1)
    chats.onNext(chat1)

    // Someone posts a message in Chat 1
    chat1.postMessage(1)
    subscriber.assertValues(
            "Chat: 1, Message: 1"
    )

    // Someone posts another message
    chat1.postMessage(2)
    subscriber.assertValues(
            "Chat: 1, Message: 1",
            "Chat: 1, Message: 2"
    )

    // Chat 1 disappears FROM USER CHAT LIST, Chat 2 is created
    val chat2 = Chat(2)
    chats.onNext(chat2)

    // Someone posts a message to Chat 2
    chat2.postMessage(1)
    subscriber.assertValues(
            "Chat: 1, Message: 1",
            "Chat: 1, Message: 2",
            "Chat: 2, Message: 1"
    )

    // Someone out there posts a message to Chat 1 that is not visible to user anymore
    chat1.postMessage(3)

    // The answer looks like this 
    //      "Chat: 1, Message: 1",
    //      "Chat: 1, Message: 2",
    //      "Chat: 2, Message: 1",
    //      "Chat: 1, Message: 3"
    // Chat 1 is still subscribed and test fails
    subscriber.assertValues(
            "Chat: 1, Message: 1",
            "Chat: 1, Message: 2",
            "Chat: 2, Message: 1",
    )
}

我想到的是使用一个主题(或可观察的共享对象)来制动一系列内部订阅。但是看起来很奇怪:

    ///////////////////////////////////
    //        MODIFIED SEQUENCE      //
    ///////////////////////////////////
    val unsubscribe: PublishSubject<Boolean> = PublishSubject.create()
    val sequence = chats
            .doOnNext({ unsubscribe.onNext(true) })
            .doAfterTerminate({ unsubscribe.onNext(true) })
            .flatMap {
                it.lastMessage.takeUntil(unsubscribe)
            }

这种方法有效，但看起来令人恐惧。
非常感谢!

Answer 1

假设评论

// Chat 1 disappears FROM USER CHAT LIST, Chat 2 is created

表示您希望当前的Chat(在这种情况下为chat1)在Chat上发送新的chats时停止发射，您可以使用switchMap运算符完成此操作。

val sequence = chats.switchMap { it.lastMessage }

从ReactiveX文档中:

RxJava also implements the switchMap operator. It behaves much like flatMap, except that whenever a new item is emitted by the source Observable, it will unsubscribe to and stop mirroring the Observable that was generated from the previously-emitted item, and begin only mirroring the current one.