Android Retrofit 2 + RxJava: listen to endless stream
Zella's answer is right for Retrofit2 with rxJava, For rxJava2, I modified custom observable like this:
//imports
import io.reactivex.Observable
import io.reactivex.disposables.Disposable
import io.reactivex.schedulers.Schedulers
import okio.BufferedSource
import java.io.IOException
fun events(source: BufferedSource): Observable<String> {
return Observable.create { emitter ->
var isCompleted = false
try {
while (!source.exhausted()) {
emitter.onNext(source.readUtf8Line()!!)
}
emitter.onComplete()
} catch (e: IOException) {
e.printStackTrace()
if (e.message == "Socket closed") {
isCompleted = true
emitter.onComplete()
} else {
throw IOException(e)
}
}
if (!isCompleted) {
emitter.onComplete()
}
}
}
Changes in module level build.gradle dependencies:
//retrofit rxJava2 adapter
implementation 'com.squareup.retrofit2:adapter-rxjava2:2.7.1'
//rx-java
implementation 'io.reactivex.rxjava2:rxjava:2.2.11'
implementation 'io.reactivex.rxjava2:rxandroid:2.1.1'
Retrofit Adapter Changes:
ITwitterAPI api = new Retrofit.Builder()
.baseUrl("http://stream.meetup.com")
.addCallAdapterFactory(RxJava2CallAdapterFactory.create())
.build().create(ITwitterAPI.class);
And Called the Streaming API as
api.twitterStream()
.subscribeOn(Schedulers.io())
.observeOn(Schedulers.io())
.flatMap { responseBody-> events(responseBody.source()) }
.subscribe({ t ->
Log.i(TAG, "onNext t=$t")
}, { e ->
Log.i(TAG, "onError e=$e")
}, {
Log.i(TAG, "onFinish")
})
Here my solution:
You can use the @Streaming annotation:
public interface ITwitterAPI {
@GET("/2/rsvps")
@Streaming
Observable<ResponseBody> twitterStream();
}
ITwitterAPI api = new Retrofit.Builder()
.baseUrl("http://stream.meetup.com")
.addCallAdapterFactory(RxJavaCallAdapterFactory.create())
.build().create(ITwitterAPI.class);
With @Streaming
we can get raw input From ResponseBody
.
Here my function to wrap body divided by lines with events:
public static Observable<String> events(BufferedSource source) {
return Observable.create(new Observable.OnSubscribe<String>() {
@Override
public void call(Subscriber<? super String> subscriber) {
try {
while (!source.exhausted()) {
subscriber.onNext(source.readUtf8Line());
}
subscriber.onCompleted();
} catch (IOException e) {
e.printStackTrace();
subscriber.onError(e);
}
}
});
}
And result usage:
api.twitterStream()
.flatMap(responseBody -> events(responseBody.source()))
.subscribe(System.out::println);
upd about gracefully stopping
When we unsubscribing, retrofit closes inputstream. But impossible to detect inputstream closed or not from inputstream themselves, so only way - try reading from stream - we gets exception with Socket closed
message.
We can interpret this exception as closing:
@Override
public void call(Subscriber<? super String> subscriber) {
boolean isCompleted = false;
try {
while (!source.exhausted()) {
subscriber.onNext(source.readUtf8Line());
}
} catch (IOException e) {
if (e.getMessage().equals("Socket closed")) {
isCompleted = true;
subscriber.onCompleted();
} else {
throw new UncheckedIOException(e);
}
}
//if response end we get here
if (!isCompleted) {
subscriber.onCompleted();
}
}
And if connection closed because response end, we haven't any exceptions. Here isCompleted
check for that. Let me know if i am wrong :)