rxjava和终止流

Question

我对使用rxjava进行反应性编程是新手，在经历了更简单的示例之后，我现在试图弄清楚如何使用连续流。下面这个例子的问题是，在我接受了3个元素后，程序不会终止。我的假设是，我不知何故需要取消订阅我的可观察的，但我不完全掌握如何终止while循环并使程序退出。

我遇到了下面的RxJava帖子--终止无限流，但我仍然不知道我遗漏了什么。

class MyTwitterDataProvider {
/*
This example is written in Groovy

Instance variables and constructor omitted
*/

public Observable<String> getTweets() {
    BufferedReader reader = new BufferedReader(new InputStreamReader(getTwitterStream()))

    Observable.create({ observer ->
        executor.execute(new Runnable() {
            def void run() {
                String newLine
                while ((newLine = reader.readLine()) != null) {
                    System.out.println("printing tweet: $newLine")
                    observer.onNext(newLine)
                }

                observer.onCompleted()
            }
        })
    })
}

def InputStream getTwitterStream() {
// code omitted
}

public static void main (String [] args) {
    MyTwitterDataProvider provider = new MyTwitterDataProvider()
    Observable<String> myTweetsObservable = provider.getTweets().take(3)

    Subscription myTweetSubscription = myTweetsObservable.subscribe({tweet-> println("client prints: $tweet")})
   // myTweetSubscription.unsubscribe()
}
}

Answer 1

您必须在循环中添加一个检查，以查看观察者是否仍被订阅：

            while ((newLine = reader.readLine()) != null && !observer.isUnsubsribed()) {
                System.out.println("printing tweet: $newLine")
                observer.onNext(newLine)
            }