Вот мое решение:
Вы можете использовать аннотацию @Streaming:
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);
С помощью @Streaming
мы можем получить исходные данные от ResponseBody
.
Вот моя функция для переноса тела, разделенного строками, с событиями:
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);
}
}
});
}
И использование результата:
api.twitterStream()
.flatMap(responseBody -> events(responseBody.source()))
.subscribe(System.out::println);
уведомление об изящной остановке
Когда мы отписываемся, модификация закрывает входной поток. Но невозможно определить закрытый входной поток или нет из самого входного потока, поэтому единственный способ - попробовать прочитать из потока - мы получим исключение с сообщением Socket closed
. Мы можем интерпретировать это исключение как закрытие:
@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();
}
}
И если соединение закрыто из-за окончания ответа, у нас нет никаких исключений. Вот isCompleted
проверьте это. Дайте мне знать, если я ошибаюсь :)
person
zella
schedule
15.04.2016