Replacing RxJava's Observable by Flowable

bitrich-info / xchange-stream

XChange-stream is a Java library providing a simple and consistent streaming API for interacting with Bitcoin and other crypto currency exchanges via WebSocket protocol. It is build on top of of XChange library providing new interfaces for streaming API. User can subscribe for live updates via reactive streams of RxJava library.

Apache License 2.0

413 stars 220 forks source link

Replacing RxJava's Observable by Flowable #148

Open sanderdg opened 6 years ago

sanderdg commented 6 years ago

Throughout the source the Observable class is used but actually Flowable should be used instead to prevent out of memory exceptions and allow for backpressure support. According to the documentation it should be used when (amongst other things) one of the following is applicable:

Potentially 10k+ of elements (ticker data for sure).
Network (Streaming) IO that requests a logical amount.

It shouldn't be to much work to change this. Thanks for this great project!

caleb-allen commented 6 years ago

+1 to this. Financial streaming data is a textbook example of the benefits of a backpressure supported stream, like Flowable

davidjirovec commented 6 years ago

Sound reasonable, are there any benefits to just calling io.reactivex.Observable#toFlowable?

badgerwithagun commented 6 years ago

Do you really want xchange-stream specifying its own backpressure strategy? What if xchange-stream decides LATEST and I want DROP?

As @davidjirovec suggests, I usually just use toFlowable().