TimeoutException when a period of time elapses without any emitted items

DMJessieSP

unread,

Apr 11, 2017, 7:22:34 AM4/11/17

to RxJava

I am trying to listen to a streaming API, but because this API is a bit slow with emitting items I get a TimeoutException when an item hasn't been emitted for a while. I don't want there to be any limit for how long I can wait for an emission. I have tried setting the timeout option to 100 seconds, but this does not work. It does work for 10 seconds however. I have also tried using retry which works because it does resubscribe, but then I still get the exception message. I have also tried using onErrorResumeNext, but this does not work either. How can I solve this problem so that I can listen to an endless stream even if the stream is a bit slow?

This is what I have used:

.flatMap(responseBody -> events(responseBody.source()))
.timeout(100, TimeUnit.SECONDS)
.onErrorResumeNext(Observable.empty())

.subscribe(System.out::println);

Dávid Karnok

unread,

Apr 11, 2017, 7:29:08 AM4/11/17

to DMJessieSP, RxJava

Hi. I don't understand the problem. Are you saying that timeout(100, TimeUnit.SECONDS) doesn't wait long enough? What is the exact stacktrace of the TimeoutException you are getting. Have you tried reproducing the issue with some offline sequences of simple data?

--

Best regards,

David Karnok

DMJessieSP

unread,

Apr 11, 2017, 7:41:50 AM4/11/17

to RxJava, sin4d...@hotmail.com

Hi,

Thank you for your reply!

Yes, I get a TimeoutException after only a couple of seconds even when setting the timeout to 100 seconds. I also tried setting it to 1 day but it makes no difference. I don't want to get a timeout at all as this creates a big problem when the stream is a bit slow.

Stack trace:

java.net.SocketTimeoutException: timeout

at okio.Okio$3.newTimeoutException(Okio.java:207)

at okio.AsyncTimeout.exit(AsyncTimeout.java:261)

at okio.AsyncTimeout$2.read(AsyncTimeout.java:215)

at okio.RealBufferedSource.request(RealBufferedSource.java:71)

at okio.RealBufferedSource.require(RealBufferedSource.java:64)

at okio.RealBufferedSource.readHexadecimalUnsignedLong(RealBufferedSource.java:270)

at okhttp3.internal.http1.Http1Codec$ChunkedSource.readChunkSize(Http1Codec.java:444)

at okhttp3.internal.http1.Http1Codec$ChunkedSource.read(Http1Codec.java:425)

at okio.RealBufferedSource.read(RealBufferedSource.java:50)

at okio.ForwardingSource.read(ForwardingSource.java:35)

at retrofit2.OkHttpCall$ExceptionCatchingRequestBody$1.read(OkHttpCall.java:285)

at okio.RealBufferedSource.exhausted(RealBufferedSource.java:60)

at com.example.meetup.MeetupListener$1.call(MeetupListener.java:141)

at com.example.meetup.MeetupListener$1.call(MeetupListener.java:135)

at rx.Observable.unsafeSubscribe(Observable.java:8666)

at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:250)

at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:147)

at rx.internal.operators.OperatorMap$MapSubscriber.onNext(OperatorMap.java:74)

at rx.internal.operators.OperatorMerge$MergeSubscriber.emitScalar(OperatorMerge.java:505)

at rx.internal.operators.OperatorMerge$MergeSubscriber.tryEmit(OperatorMerge.java:463)

at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:246)

at rx.internal.operators.OperatorMerge$MergeSubscriber.onNext(OperatorMerge.java:147)

at rx.internal.operators.OperatorMap$MapSubscriber.onNext(OperatorMap.java:74)

at retrofit2.adapter.rxjava.RxJavaCallAdapterFactory$CallOnSubscribe.call(RxJavaCallAdapterFactory.java:146)

at retrofit2.adapter.rxjava.RxJavaCallAdapterFactory$CallOnSubscribe.call(RxJavaCallAdapterFactory.java:125)

at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:50)

at rx.internal.operators.OnSubscribeLift.call(OnSubscribeLift.java:30)