kaushikgopal/rxjava-android-samples

Learning RxJava for Android by example

Learning RxJava for Android by example

This is a repository with real-world useful examples of using RxJava with Android. It usually will be in a constant state of "Work in Progress" (WIP).

I've also been giving talks about Learning Rx using many of the examples listed in this repo.

• Learning RxJava For Android by Example : Part 1 [slides] (SF Android Meetup 2015)
• Learning Rx by Example : Part 2 [slides] (Øredev 2016)

Examples:

1. Background work & concurrency (using Schedulers)
2. Accumulate calls (using buffer)
3. Instant/Auto searching text listeners (using Subjects & debounce)
4. Networking with Retrofit & RxJava (using zip, flatmap)
5. Two-way data binding for TextViews (using PublishSubject)
6. Simple and Advanced polling (using interval and repeatWhen)
7. Simple and Advanced exponential backoff (using delay and retryWhen)
8. Form validation (using combineLatest)
9. Pseudo caching : retrieve data first from a cache, then a network call (using concat, concatEager, merge or publish)
10. Simple timing demos (using timer, interval or delay)
11. RxBus : event bus using RxJava (using RxRelay (never terminating Subjects) and debouncedBuffer)
12. Persist data on Activity rotations (using Subjects and retained Fragments)
13. Networking with Volley
14. Pagination with Rx (using Subjects)
15. Orchestrating Observable: make parallel network calls, then combine the result into a single data point (using flatmap & zip)
16. Simple Timeout example (using timeout)
17. Setup and teardown resources (using using)
18. Multicast playground

Description

1. Background work & concurrency (using Schedulers)

A common requirement is to offload lengthy heavy I/O intensive operations to a background thread (non-UI thread) and feed the results back to the UI/main thread, on completion. This is a demo of how long-running operations can be offloaded to a background thread. After the operation is done, we resume back on the main thread. All using RxJava! Think of this as a replacement to AsyncTasks.

The long operation is simulated by a blocking Thread.sleep call (since this is done in a background thread, our UI is never interrupted).

To really see this example shine. Hit the button multiple times and see how the button click (which is a UI operation) is never blocked because the long operation only runs in the background.

2. Accumulate calls (using buffer)

This is a demo of how events can be accumulated using the "buffer" operation.

A button is provided and we accumulate the number of clicks on that button, over a span of time and then spit out the final results.

If you hit the button once, you'll get a message saying the button was hit once. If you hit it 5 times continuously within a span of 2 seconds, then you get a single log, saying you hit that button 5 times (vs 5 individual logs saying "Button hit once").

Note:

If you're looking for a more foolproof solution that accumulates "continuous" taps vs just the number of taps within a time span, look at the EventBus Demo where a combo of the publish and buffer operators is used. For a more detailed explanation, you can also have a look at this blog post.

3. Instant/Auto searching text listeners (using Subjects & debounce)

This is a demo of how events can be swallowed in a way that only the last one is respected. A typical example of this is instant search result boxes. As you type the word "Bruce Lee", you don't want to execute searches for B, Br, Bru, Bruce, Bruce, Bruce L ... etc. But rather intelligently wait for a couple of moments, make sure the user has finished typing the whole word, and then shoot out a single call for "Bruce Lee".

As you type in the input box, it will not shoot out log messages at every single input character change, but rather only pick the lastly emitted event (i.e. input) and log that.

This is the debounce/throttleWithTimeout method in RxJava.

4. Networking with Retrofit & RxJava (using zip, flatmap)

Retrofit from Square is an amazing library that helps with easy networking (even if you haven't made the jump to RxJava just yet, you really should check it out). It works even better with RxJava and these are examples hitting the GitHub API, taken straight up from the android demigod-developer Jake Wharton's talk at Netflix. You can watch the talk at this link. Incidentally, my motivation to use RxJava was from attending this talk at Netflix.

(Note: you're most likely to hit the GitHub API quota pretty fast so send in an OAuth-token as a parameter if you want to keep running these examples often).

5. Two-way data binding for TextViews (using PublishSubject)

Auto-updating views are a pretty cool thing. If you've dealt with Angular JS before, they have a pretty nifty concept called "two-way data binding", so when an HTML element is bound to a model/entity object, it constantly "listens" to changes on that entity and auto-updates its state based on the model. Using the technique in this example, you could potentially use a pattern like the Presentation View Model pattern with great ease.

While the example here is pretty rudimentary, the technique used to achieve the double binding using a Publish Subject is much more interesting.

6. Simple and Advanced polling (using interval and repeatWhen)

This is an example of polling using RxJava Schedulers. This is useful in cases, where you want to constantly poll a server and possibly get new data. The network call is "simulated" so it forces a delay before return a resultant string.

There are two variants for this:

1. Simple Polling: say when you want to execute a certain task every 5 seconds
2. Increasing Delayed Polling: say when you want to execute a task first in 1 second, then in 2 seconds, then 3 and so on.

The second example is basically a variant of Exponential Backoff.

Instead of using a RetryWithDelay, we use a RepeatWithDelay here. To understand the difference between Retry(When) and Repeat(When) I wouuld suggest Dan's fantastic post on the subject.

An alternative approach to delayed polling without the use of repeatWhen would be using chained nested delay observables. See startExecutingWithExponentialBackoffDelay in the ExponentialBackOffFragment example.

7. Simple and Advanced exponential backoff (using delay and retryWhen)

Exponential backoff is a strategy where based on feedback from a certain output, we alter the rate of a process (usually reducing the number of retries or increasing the wait time before retrying or re-executing a certain process).

The concept makes more sense with examples. RxJava makes it (relatively) simple to implement such a strategy. My thanks to Mike for suggesting the idea.

Retry (if error) with exponential backoff

Say you have a network failure. A sensible strategy would be to NOT keep retrying your network call every 1 second. It would be smart instead (nay... elegant!) to retry with increasing delays. So you try at second 1 to execute the network call, no dice? try after 10 seconds... negatory? try after 20 seconds, no cookie? try after 1 minute. If this thing is still failing, you got to give up on the network yo!

We simulate this behaviour using RxJava with the retryWhen operator.

RetryWithDelay code snippet courtesy:

• http://stackoverflow.com/a/25292833/159825
• Another excellent implementation via @sddamico : https://gist.github.com/sddamico/c45d7cdabc41e663bea1
• This one includes support for jittering, by @leandrofavarin : http://leandrofavarin.com/exponential-backoff-rxjava-operator-with-jitter

Also look at the Polling example where we use a very similar Exponential backoff mechanism.

"Repeat" with exponential backoff

Another variant of the exponential backoff strategy is to execute an operation for a given number of times but with delayed intervals. So you execute a certain operation 1 second from now, then you execute it again 10 seconds from now, then you execute the operation 20 seconds from now. After a grand total of 3 times you stop executing.

Simulating this behavior is actually way more simpler than the prevoius retry mechanism. You can use a variant of the delay operator to achieve this.

8. Form validation (using .combineLatest)

Thanks to Dan Lew for giving me this idea in the fragmented podcast - episode #4 (around the 4:30 mark).

.combineLatest allows you to monitor the state of multiple observables at once compactly at a single location. The example demonstrated shows how you can use .combineLatest to validate a basic form. There are 3 primary inputs for this form to be considered "valid" (an email, a password and a number). The form will turn valid (the text below turns blue :P) once all the inputs are valid. If they are not, an error is shown against the invalid inputs.

We have 3 independent observables that track the text/input changes for each of the form fields (RxAndroid's WidgetObservable comes in handy to monitor the text changes). After an event change is noticed from all 3 inputs, the result is "combined" and the form is evaluated for validity.

Note that the Func3 function that checks for validity, kicks in only after ALL 3 inputs have received a text change event.

The value of this technique becomes more apparent when you have more number of input fields in a form. Handling it otherwise with a bunch of booleans makes the code cluttered and kind of difficult to follow. But using .combineLatest all that logic is concentrated in a nice compact block of code (I still use booleans but that was to make the example more readable).

9. Pseudo caching : retrieve data first from a cache, then a network call (using concat, concatEager, merge or publish)

We have two source Observables: a disk (fast) cache and a network (fresh) call. Typically the disk Observable is much faster than the network Observable. But in order to demonstrate the working, we've also used a fake "slower" disk cache just to see how the operators behave.

This is demonstrated using 4 techniques:

1. .concat
2. .concatEager
3. .merge
4. .publish selector + merge + takeUntil

The 4th technique is probably what you want to use eventually but it's interesting to go through the progression of techniques, to understand why.

concat is great. It retrieves information from the first Observable (disk cache in our case) and then the subsequent network Observable. Since the disk cache is presumably faster, all appears well and the disk cache is loaded up fast, and once the network call finishes we swap out the "fresh" results.

The problem with concat is that the subsequent observable doesn't even start until the first Observable completes. That can be a problem. We want all observables to start simultaneously but produce the results in a way we expect. Thankfully RxJava introduced concatEager which does exactly that. It starts both observables but buffers the result from the latter one until the former Observable finishes. This is a completely viable option.

Sometimes though, you just want to start showing the results immediately. Assuming the first observable (for some strange reason) takes really long to run through all its items, even if the first few items from the second observable have come down the wire it will forcibly be queued. You don't necessarily want to "wait" on any Observable. In these situations, we could use the merge operator. It interleaves items as they are emitted. This works great and starts to spit out the results as soon as they're shown.

Similar to the concat operator, if your first Observable is always faster than the second Observable you won't run into any problems. However the problem with merge is: if for some strange reason an item is emitted by the cache or slower observable after the newer/fresher observable, it will overwrite the newer content. Click the "MERGE (SLOWER DISK)" button in the example to see this problem in action. @JakeWharton and @swankjesse contributions go to 0! In the real world this could be bad, as it would mean the fresh data would get overridden by stale disk data.

To solve this problem you can use merge in combination with the super nifty publish operator which takes in a "selector". I wrote about this usage in a blog post but I have Jedi JW to thank for reminding of this technique. We publish the network observable and provide it a selector which starts emitting from the disk cache, up until the point that the network observable starts emitting. Once the network observable starts emitting, it ignores all results from the disk observable. This is perfect and handles any problems we might have.

Previously, I was using the merge operator but overcoming the problem of results being overwritten by monitoring the "resultAge". See the old PseudoCacheMergeFragment example if you're curious to see this old implementation.

10. Simple timing demos (using timer, interval and delay)

This is a super simple and straightforward example which shows you how to use RxJava's timer, interval and delay operators to handle a bunch of cases where you want to run a task at specific intervals. Basically say NO to Android TimerTasks.

Cases demonstrated here:

1. run a single task after a delay of 2s, then complete
2. run a task constantly every 1s (there's a delay of 1s before the first task fires off)
3. run a task constantly every 1s (same as above but there's no delay before the first task fires off)
4. run a task constantly every 3s, but after running it 5 times, terminate automatically
5. run a task A, pause for sometime, then execute Task B, then terminate

11. RxBus : event bus using RxJava (using RxRelay (never terminating Subjects) and debouncedBuffer)

There are accompanying blog posts that do a much better job of explaining the details on this demo:

1. Implementing an event bus with RxJava
2. DebouncedBuffer used for the fancier variant of the demo
3. share/publish/refcount

12. Persist data on Activity rotations (using Subjects and retained Fragments)

A common question that's asked when using RxJava in Android is, "how do i resume the work of an observable if a configuration change occurs (activity rotation, language locale change etc.)?".

This example shows you one strategy viz. using retained Fragments. I started using retained fragments as "worker fragments" after reading this fantastic post by Alex Lockwood quite sometime back.

Hit the start button and rotate the screen to your heart's content; you'll see the observable continue from where it left off.

There are certain quirks about the "hotness" of the source observable used in this example. Check my blog post out where I explain the specifics.

I have since rewritten this example using an alternative approach. While the ConnectedObservable approach worked it enters the lands of "multicasting" which can be tricky (thread-safety, .refcount etc.). Subjects on the other hand are far more simple. You can see it rewritten using a Subject here.

I wrote another blog post on how to think about Subjects where I go into some specifics.

13. Networking with Volley

Volley is another networking library introduced by Google at IO '13. A kind citizen of github contributed this example so we know how to integrate Volley with RxJava.

14. Pagination with Rx (using Subjects)

I leverage the simple use of a Subject here. Honestly, if you don't have your items coming down via an Observable already (like through Retrofit or a network request), there's no good reason to use Rx and complicate things.

This example basically sends the page number to a Subject, and the subject handles adding the items. Notice the use of concatMap and the return of an Observable<List> from _itemsFromNetworkCall.

For kicks, I've also included a PaginationAutoFragment example, this "auto-paginates" without us requiring to hit a button. It should be simple to follow if you got how the previous example works.

Here are some other fancy implementations (while i enjoyed reading them, i didn't land up using them for my real world app cause personally i don't think it's necessary):

• Matthias example of an Rx based pager
• Eugene's very comprehensive Pagination sample
• Recursive Paging example

15. Orchestrating Observable: make parallel network calls, then combine the result into a single data point (using flatmap & zip)

The below ascii diagram expresses the intention of our next example with panache. f1,f2,f3,f4,f5 are essentially network calls that when made, give back a result that's needed for a future calculation.

         (flatmap)
f1 ___________________ f3 _______
         (flatmap)               |    (zip)
f2 ___________________ f4 _______| ___________  final output
        \                        |
         \____________ f5 _______|

The code for this example has already been written by one Mr.skehlet in the interwebs. Head over to the gist for the code. It's written in pure Java (6) so it's pretty comprehensible if you've understood the previous examples. I'll flush it out here again when time permits or I've run out of other compelling examples.

16. Simple Timeout example (using timeout)

This is a simple example demonstrating the use of the .timeout operator. Button 1 will complete the task before the timeout constraint, while Button 2 will force a timeout error.

Notice how we can provide a custom Observable that indicates how to react under a timeout Exception.

17. Setup and teardown resources (using using)

The operator using is relatively less known and notoriously hard to Google. It's a beautiful API that helps to setup a (costly) resource, use it and then dispose off in a clean way.

The nice thing about this operator is that it provides a mechansim to use potentially costly resources in a tightly scoped manner. using -> setup, use and dispose. Think DB connections (like Realm instances), socket connections, thread locks etc.

18. Multicast Playground

Multicasting in Rx is like a dark art. Not too many folks know how to pull it off without concern. This example condiers two subscribers (in the forms of buttons) and allows you to add/remove subscribers at different points of time and see how the different operators behave under those circumstances.

The source observale is a timer (interval) observable and the reason this was chosen was to intentionally pick a non-terminating observable, so you can test/confirm if your multicast experiment will leak.

I also gave a talk about Multicasting in detail at 360|Andev. If you have the inclination and time, I highly suggest watching that talk first (specifically the Multicast operator permutation segment) and then messing around with the example here.

Rx 2.x

All the examples here have been migrated to use RxJava 2.X.

• Have a look at PR #83 to see the diff of changes between RxJava 1 and 2
• What's different in Rx 2.x

We use David Karnok's Interop library in some cases as certain libraries like RxBindings, RxRelays, RxJava-Math etc. have not been ported yet to 2.x.

Contributing:

I try to ensure the examples are not overly contrived but reflect a real-world usecase. If you have similar useful examples demonstrating the use of RxJava, feel free to send in a pull request.

I'm wrapping my head around RxJava too so if you feel there's a better way of doing one of the examples mentioned above, open up an issue explaining how. Even better, send a pull request.

Sponsorship (Memory Management/Profiling)

Rx threading is messy business. To help, this project uses YourKit tools for analysis.

YourKit supports open source projects with innovative and intelligent tools for monitoring and profiling Java applications. YourKit is the creator of YourKit Java Profiler.

License

Licensed under the Apache License, Version 2.0 (the "License"). You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

You agree that all contributions to this repository, in the form of fixes, pull-requests, new examples etc. follow the above-mentioned license.

rxjava-android-samples's Issues

feat: material design updates

While the draw to these samples are the examples and .java code per say, it would be nice to have the app not look so dated.

Not looking for anything super fancy, just a nice update to the material design defaults

RxJava2: Create an examples handing backpressure

Since we soon will have RxJava2 here (#83) it would be nice to have examples that handle backpressure. Do you have any ideas for good examples?

Question On Cache and retaining the observable

I saw your video on Rx Java on youtube. From what i understand the observable needs to be retained during orientation change. One way to do that is in a retained fragment without ui.

I attended a talk on android organized By B.A.U.G. In the sample they used dagger, retrofit and EventBus. The point was to have model, view and presenter. In the example they use a singleton to retain the Observable and they use cache operator to cache the data. There is a problem if you do not have network connection your cache is empty and you go back to previous activity and come back you get the same empty cache.

I have been going through blogs, videos and i still don't understand how to handle it in a clear way.

Sample Project https://github.com/anupcowkur/MVPSample.

So how do i deal with this.

Sorry! if this is not the right place to raise the issue. I would be happy to put this in the right place. In fact i did ask this question on Rx java group and i haven't got an answer.

Exceeds Dex Limit of 64k in its current state.

I just cloned the repo and tried to build

./gradlew clean assemble  

The build fails with the following error:

Dex: The number of method references in a .dex file cannot exceed 64K.

Is this a known issue? I can try and submit a pull request to fix this soon. I just wanted to check before a pull request.

GUI elements must not be accessed from non-main threads

As far as I understand the following code from DebounceSearchEmitterFragment.java#L76:

_disposable = RxJavaInterop.toV2Observable(RxTextView.textChangeEvents(_inputSearchText))
              .debounce(400, TimeUnit.MILLISECONDS)// default Scheduler is Computation
              .filter(changes -> isNotNullOrEmpty(_inputSearchText.getText().toString()))
              .observeOn(AndroidSchedulers.mainThread())
              .subscribeWith(_getSearchObserver());

filter() accesses the GUI element _inputSearchText, which it must not do: see StackOverflow: Is it okay to read data from UI elements in another thread?

Instead the text should be read from the change-event like this:
.filter(changes -> isNotNullOrEmpty(changes.text().toString()))

The meaning of "_subscriptions.add(tapEventEmitter.connect());"?

this is a segment of class RxBusDemo_Bottom3Fragment appears about line 77

and ,I want to know why it wrote like that,and whether that:

could

_subscriptions.add(tapEventEmitter.connect());

be replaced by only

tapEventEmitter.connect();

as I only want to emmit the items to its subscribers

Thank you very much ! :)

If there are many events, Should I define many classes?

Member variables should follow the `mVariableName` naming standard

... instead of using _variableName.

I'm sure a simple script could be written to change them all automatically.

orientation change messes it up

You should add retained fragments (see http://www.androiddesignpatterns.com/2013/04/retaining-objects-across-config-changes.html) or http://developer.android.com/reference/android/content/AsyncTaskLoader.html to retain the REST calls when the activity is restarted on device rotation. Right now, on orientation change, you get various errors in logcat besides giving a sub-optimal experience to the user. E.g.:

09-22 15:51:19.793 9639-9639/com.morihacky.android.rxjava E/ViewRootImpl﹕ sendUserActionEvent() mView == null
09-22 15:51:28.173 9639-9639/com.morihacky.android.rxjava E/RetrofitFragment﹕ woops we got an error while getting the list of contributors along with full names
retrofit.RetrofitError: 403 Forbidden

AndroidObservable

You should always use AndroidObservable to bind the source sequence to the activity/fragment. This way you ensure that if the activity is scheduled to finish no notifications will be forwarded, plus it already observes on Android UI main thread.

for example:

subscription = AndroidObservable.bindActivity(this, YourObservable).subscribe();

AndroidObservable bindActivity() or bindFragment() both return an observable product of yours, so you can treat it normally as any observable. AndroidObservable has other interesting methods such as fromBroadcast() which creates an observable that wraps a BroadcastReceiver and emit received intents.

After value change -> execute once

After getting a specfic value from an Observable, I would like to execute a method only once. -> on Completed()

How could this be possible?

Have you stopped maintaining this project?

Seems multiple PRs oustanding.

getRxBusSingleton();

Your calling

_rxBus = ((MainActivity) getActivity()).getRxBusSingleton();

but I dont see this method in RxBus.java??

edit:

I see it's in the host activity. But why isn't it in the RxBus class?

Can you explain the getsingleton call and science behind it?

thanks!

Broken Link

http://nerds.weddingpartyapp.com/tech/2014/12/24/rxjava-share-publish-refcount-and-all-that-jazz/

Can you please fix this link ?

rxbus in long running sticky service

I'm super new to rx thinking, it took me like 4-5 mos to get this far in
implementing the future in my code so bare w/ me...

is it ok to use rxbus in a service?

I have network data incoming from a background service constantly, I want to send it on its way via rxbus to fragments, quite possibly even store the data based on certain conditions, along its route.

are there any issues thats a apparent with that use case?

i'm attempting all this now.

gc overhead limit exceeded?

Hi, I just want to see your example however I am having an error when running your example. here it is.

Error:java.lang.OutOfMemoryError: GC overhead limit exceeded
Error:	at java.util.Arrays.copyOfRange(Arrays.java:3664)
Error:	at java.lang.String.<init>(String.java:207)
Error:	at java.lang.StringBuilder.toString(StringBuilder.java:407)
Error:	at com.android.dx.rop.type.Prototype.withFirstParameter(Prototype.java:370)
Error:	at com.android.dx.rop.type.Prototype.intern(Prototype.java:180)
Error:	at com.android.dx.cf.iface.StdMethod.<init>(StdMethod.java:46)
Error:	at com.android.dx.cf.direct.MethodListParser.set(MethodListParser.java:81)
Error:	at com.android.dx.cf.direct.MemberListParser.parse(MemberListParser.java:217)
Error:	at com.android.dx.cf.direct.MemberListParser.parseIfNecessary(MemberListParser.java:108)
Error:	at com.android.dx.cf.direct.MethodListParser.getList(MethodListParser.java:54)
Error:	at com.android.dx.cf.direct.DirectClassFile.parse0(DirectClassFile.java:551)
Error:	at com.android.dx.cf.direct.DirectClassFile.parse(DirectClassFile.java:406)
Error:	at com.android.dx.cf.direct.DirectClassFile.parseToInterfacesIfNecessary(DirectClassFile.java:388)
Error:	at com.android.dx.cf.direct.DirectClassFile.getMagic(DirectClassFile.java:251)
Error:	at com.android.dx.command.dexer.Main.parseClass(Main.java:772)
Error:	at com.android.dx.command.dexer.Main.access$1500(Main.java:85)
Error:	at com.android.dx.command.dexer.Main$ClassParserTask.call(Main.java:1700)
Error:	at com.android.dx.command.dexer.Main.processClass(Main.java:755)
Error:	at com.android.dx.command.dexer.Main.processFileBytes(Main.java:723)
Error:	at com.android.dx.command.dexer.Main.access$1200(Main.java:85)
Error:	at com.android.dx.command.dexer.Main$FileBytesConsumer.processFileBytes(Main.java:1653)
Error:	at com.android.dx.cf.direct.ClassPathOpener.processArchive(ClassPathOpener.java:284)
Error:	at com.android.dx.cf.direct.ClassPathOpener.processOne(ClassPathOpener.java:166)
Error:	at com.android.dx.cf.direct.ClassPathOpener.process(ClassPathOpener.java:144)
Error:	at com.android.dx.command.dexer.Main.processOne(Main.java:677)
Error:	at com.android.dx.command.dexer.Main.processAllFiles(Main.java:569)
Error:	at com.android.dx.command.dexer.Main.runMultiDex(Main.java:366)
Error:	at com.android.dx.command.dexer.Main.run(Main.java:275)
Error:	at com.android.dx.command.dexer.Main.main(Main.java:245)
Error:	at com.android.dx.command.Main.main(Main.java:106)

force close on sample bg work

I click on button several times then press back and go to first page. then this error occurs:
at com.morihacky.android.rxjava.fragments.ConcurrencyWithSchedulersDemoFragment$2.onCompleted(ConcurrencyWithSchedulersDemoFragment.java:100)
if you unsibscribe the listener why this happen? can fix this issue pls :)

Wiki Pages for examples

Do you think that having github wiki page for each example would help people new to Rx understand the examples better? I'm thinking something like Code Path. Then we could use resource like http://rxmarbles.com/ to help visualize the example.

Enhanced RxBus

Nice repo! Also really simple and easy approach to achieve Pub-Sub pattern in RxBus.

would you consider enhancing the example with smth like https://github.com/apptik/rxHub ?
I would be more than happy to provide a PR.

What does RotationPersist1WorkerFragment and RotationPersist2WorkerFragment do?

First of all, Thank you so much for these samples. I am new into android and i am working my way around advance stuff on my own. I am trying to adopt the approach you are using in this sample in my new app. I have gone through the code and i found these two fragments RotationPersist1WorkerFragment and RotationPersist2WorkerFragment. I think these two fragments are for to keep the track of the fragments on back press(correct me if i am wrong). Since i am new to this. Would you please like to explain further about it?
Why are there two worker fragments? when only one could have done the same work.
Looking forward to hear back from you.
Thanks much.

Rxjava 2 with Android new architecture components

i think using rxjava with android architecture component will be good mix especially with the new "VIEW MODEL" class and using that component for retaining that rxjava stream instead of retain fragment and reconnecting it when activity and fragment get recreated after orientation change. do u have any plan for adding this examples

I'm very interested in paging.

how to tell zip operator to execute observables one after another ?

This is not a issue on your demonstration code. More over required suggestion and request to add a example.

Based on one network call, i needed to make multiple network calls to finish a particular action taken by user and update the UI. I have used flatmap and zip to acheive this. Thanks to the demo you have put. It helped.

The problem i am facing is when making multiple network calls based on first network call response, zip operator executes every observables in parallel ( which is what expected) which at one point server is rejecting with maximum query per second error. How to execute the observables given to zip operator one after another ( serially). or should i use some other operator to execute serial. Any advise ?
Also if you could add some demonstration on this would be useful for others. As i feel this is common use case in android app development.

http://stackoverflow.com/questions/26695099/compose-multiple-network-calls-rxjava-android/26696774#26696774

Thanks.

Retrolambda is too difficult for beginners

Retrolambda is too difficult for beginners。

RxJava 1 branch

There's still plenty of RxJava users who haven't migrated to RxJava 2, please provide RxJava 1 branch. Sure you can checkout to #e200ab but I think it would be easier just to switch a dedicated branch.

Can not subscribe to observable again once the app is unsubscribed with FormValidationCombineLatestFragment

Hi,

First of all, thank you for the great samples, Kaushik. Very useful for RxJava beginners.

I have questions about the title.
I found out that in Form Validation with CombineLatest sample (FormValidationCombineLatestFragment.java), observer can not subscribe to observable again once the app goes into background by pressing "HOME" button and then returns to foreground.

I thought this is because _subscription.unsubscribe() is call at onPause() and _combineLatestEvents() is called at onCreateView().
So, I changed the code as below so that observer are subscribed again when the app comes to foreground.

// FormValidationCombineLatestFragment.java
@Override
public void onResume() {
    super.onResume();
    _combineLatestEvents();
}

@Override
public void onPause() {
    super.onPause();
    if (_subscription != null) {
        _subscription.unsubscribe();
    }
}

However, this did not work.
Then, I changed the code to call _subscription.unsubscribe() at onDestroy(), and finally it worked as I expected.

@Override
public void onDestroy() {
    super.onDestroy();
    if (_subscription != null) {
        _subscription.unsubscribe();
    }
}

So my question is,

1. Is calling _subscription.unsubscribe() at onDestroy() a right decision?
2. Why calling _combineLatestEvents() at onResume() can not re-subscribe to observable?

It would be helpful if anyone has solution for these questions.
Thank you

Rxbus + Sticky Events?

What about making the events sticky?

If I use rxbus with android's sharing mechnism how will I get the event in the activity where
the shared info is received.

The event will be sent before the activity is opened...do I have to persist the event?
would'nt that make a bus irrelevant?

takeUntil optimized cache and network method and Gson parse model

Hey
Thank you for this awesome examples :)

I am trying to implement the optimized cache method using takeUntil in my app. I am using a Gson model to parse the JSON response from the API .
I've been testing several methods but didn't succeed to get it right.am beginner in RxJava .

My code
https://gist.github.com/alouanemed/6b5f4d5d3623ae69331de25c12dd5b98

Thanks :) .

finish activity: Long running save/send data to server

Do you have any suggestions for such a usecase with rxjava?

1. user: backpress/leaves activity
2. save/send data to server -> potentially long running task

Does RotationPersist leak the Activity?

If I add LeakCanary by just putting

    LeakCanary.install(getApplication());

in the MainActivity's onCreate(), then

    debugCompile 'com.squareup.leakcanary:leakcanary-android:1.3.1' 
    releaseCompile 'com.squareup.leakcanary:leakcanary-android-no-op:1.3.1' 

to the build.gradle dependencies, I find that hitting BACK during the counter increment reports a "suspected leak" in the log and the LeakCanary "Brr..." screen comes up. In fact, you also get it if you just rotate a few times when the counter is running but exit after it has completed normally. Is this a real leak?

01-15 16:37:53.960 4308-4324/com.morihacky.android.rxjava I/art: hprof: heap dump "/storage/sdcard/Download/leakcanary/suspected_leak_heapdump.hprof" starting...

Update samples to reflect version1.0.+ (widget etc. package removed)

something strange about RxBus ？

for example: when there are 2 Observers subscribe the RxBUS, the RxBus will send event 4 times, that means ,you will receiver the event 2 times on each Observer.

even if ,I do this ,It sames does not work ,

.debounce(100, TimeUnit.MILLISECONDS)//事件去重
                //.distinct()
                .distinctUntilChanged()
                .subscribe(new Action1<Object>() {
                    @Override
                    public void call(Object event) {
                        Log.e("Rxbus", "call() called with: " + "event = [" + event + "]");
                        eventLisener.dealRxEvent(event);
                    }
                }));

can you help me ,thanks

String is van empty or null

Hi Kaushik,

I like your example of the form validation, thanks for that!

Is there a reason why you use the
com.google.common.base.Strings.isNullOrEmpty instead of the TextUtils.isEmpty() in there?

Observing a single Observable in multiple fragments/activities

To observe the same Observable (results of a Retrofit call, for example) in multiple fragments or activities, is it normal to subscribe to the Observable in the same scope where it exists (retained fragment, singleton, whatever) in order to keep the calls from terminating (e.g. chained network calls with flatMap()) and create a new Subscription to the Observable in each interested fragment/activity? Or is there a better approach?

Seems an error in PseudoCacheConcatFragment

missing "subscribeOn(Schedulers.io())" , leads a "Network on Mainthread Exception"

Double binding TextView demo also can be implemented with combineLatest

I think Double binding TextView demo also can be implemented with combineLatest operator

Configuration Changes

Device rotation in any of the examples fragments will return to the MainFragment.
just checking the savedInstanceState before replacing the mainFragment in mainActivity would be sufficient.

You are using a private api (as of support library v25 and up)

Meaning getFragments at https://github.com/kaushikgopal/RxJava-Android-Samples/blob/master/app/src/main/java/com/morihacky/android/rxjava/fragments/RotationPersist2WorkerFragment.java
Also asked about it at http://stackoverflow.com/questions/41969694/

Long operation doesn't work on orientation change

If I "start long operation" and rotates the screen, app throws this exception:
com.morihacky.android.rxjava W/System.err﹕ java.lang.InterruptedException
com.morihacky.android.rxjava W/System.err﹕ at java.lang.VMThread.sleep(Native Method)
com.morihacky.android.rxjava W/System.err﹕ at java.lang.Thread.sleep(Thread.java:1013)
com.morihacky.android.rxjava W/System.err﹕ at java.lang.Thread.sleep(Thread.java:995)
com.morihacky.android.rxjava W/System.err﹕ at com.morihacky.android.rxjava.ConcurrencyWithSchedulersDemoFragment._doSomeLongOperation_thatBlocksCurrentThread(ConcurrencyWithSchedulersDemoFragment.java:143)
com.morihacky.android.rxjava W/System.err﹕ at com.morihacky.android.rxjava.ConcurrencyWithSchedulersDemoFragment.access$100(ConcurrencyWithSchedulersDemoFragment.java:33)
com.morihacky.android.rxjava W/System.err﹕ at com.morihacky.android.rxjava.ConcurrencyWithSchedulersDemoFragment$1.call(ConcurrencyWithSchedulersDemoFragment.java:68)
com.morihacky.android.rxjava W/System.err﹕ at com.morihacky.android.rxjava.ConcurrencyWithSchedulersDemoFragment$1.call(ConcurrencyWithSchedulersDemoFragment.java:60)
com.morihacky.android.rxjava W/System.err﹕ at rx.Observable$2.call(Observable.java:270)

Port to RxJava 2.0

https://github.com/ReactiveX/RxJava/releases/tag/v2.0.0

Time and tide wait for no one. Should start porting these

Memory leak

Hi Kaushik, thanks for yours examples about RxJava. I`ve just started learning it. I investigated RxJava with Retrofit, and found an issue. When the request being done to server, an exception is thrown with 403 error. onError called, an exception logged. looks Ok.
The issue is - when I close the app with back button, and open again, and do the request again, exception is thrown, and 2 errors was logged. Do the same steps, onError was called, 3 erros was logged, etc. Memory leak occurs.

Firstly - it seems strange, the Observable finished with an error, onError was called, app was closed with back button, onDestroy was called, all objects should be garbage collected.Isn't it ?

Ok, I tried to call unsubscribe on OnDestroy, the issue still reproduce.
I googled the issue and found workaround using WeakReference. It looks awkwardly.

What is your opinion about this ? Or, I did something wrong ?

Tests

Would you be interested in me writing at least a few tests to confirm RxJava samples provide expected behaviour? Once we have tests in place, we can make other changes, such updating to latest dependencies, with much greater confidence.

This would probably require some refactoring of existing code. I can start with just one example to check if you are happy with the approach.

java.util.concurrent.ExecutionException in Volley example causes an exception in Rx java

Hello,

I've just started learning RxJava and I'm going through your examples (which are great, btw).

I've been using Volley a lot, so I wanted to try out the Volley example, however, I get an exception there from time to time:

07-07 11:03:13.327 935-992/com.morihacky.android.rxjava E/routes: com.android.volley.TimeoutError 07-07 11:03:13.331 935-935/com.morihacky.android.rxjava E/AndroidRuntime: FATAL EXCEPTION: main java.lang.IllegalStateException: Fatal Exception thrown on Scheduler.Worker thread. at rx.android.schedulers.LooperScheduler$ScheduledAction.run(LooperScheduler.java:114) at android.os.Handler.handleCallback(Handler.java:615) at android.os.Handler.dispatchMessage(Handler.java:92) at android.os.Looper.loop(Looper.java:137) at android.app.ActivityThread.main(ActivityThread.java:4745) at java.lang.reflect.Method.invokeNative(Native Method) at java.lang.reflect.Method.invoke(Method.java:511) at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:786) at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:553) at dalvik.system.NativeStart.main(Native Method) Caused by: rx.exceptions.OnErrorFailedException: Error occurred when trying to propagate error to Observer.onError at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:192) at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:120) at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.checkTerminated(OperatorObserveOn.java:264) at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.call(OperatorObserveOn.java:207) at rx.android.schedulers.LooperScheduler$ScheduledAction.run(LooperScheduler.java:107) at android.os.Handler.handleCallback(Handler.java:615)  at android.os.Handler.dispatchMessage(Handler.java:92)  at android.os.Looper.loop(Looper.java:137)  at android.app.ActivityThread.main(ActivityThread.java:4745)  at java.lang.reflect.Method.invokeNative(Native Method)  at java.lang.reflect.Method.invoke(Method.java:511)  at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:786)  at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:553)  at dalvik.system.NativeStart.main(Native Method)  Caused by: rx.exceptions.CompositeException: 2 exceptions occurred. at rx.observers.SafeSubscriber._onError(SafeSubscriber.java:192)  at rx.observers.SafeSubscriber.onError(SafeSubscriber.java:120)  at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.checkTerminated(OperatorObserveOn.java:264)  at rx.internal.operators.OperatorObserveOn$ObserveOnSubscriber.call(OperatorObserveOn.java:207)  at rx.android.schedulers.LooperScheduler$ScheduledAction.run(LooperScheduler.java:107)  at android.os.Handler.handleCallback(Handler.java:615)  at android.os.Handler.dispatchMessage(Handler.java:92)  at android.os.Looper.loop(Looper.java:137)  at android.app.ActivityThread.main(ActivityThread.java:4745)  at java.lang.reflect.Method.invokeNative(Native Method)  at java.lang.reflect.Method.invoke(Method.java:511)  at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:786)  at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:553)  at dalvik.system.NativeStart.main(Native Method)  Caused by: rx.exceptions.CompositeException$CompositeExceptionCausalChain: Chain of Causes for CompositeException In Order Received => at android.util.Log.getStackTraceString(Log.java:314) at android.util.Slog.e(Slog.java:77) at com.android.internal.os.RuntimeInit$UncaughtHandler.uncaughtException(RuntimeInit.java:72) at java.lang.ThreadGroup.uncaughtException(ThreadGroup.java:693) at java.lang.ThreadGroup.uncaughtException(ThreadGroup.java:690) at rx.android.schedulers.LooperScheduler$ScheduledAction.run(LooperScheduler.java:118) at android.os.Handler.handleCallback(Handler.java:615)  at android.os.Handler.dispatchMessage(Handler.java:92)  at android.os.Looper.loop(Looper.java:137)  at android.app.ActivityThread.main(ActivityThread.java:4745)  at java.lang.reflect.Method.invokeNative(Native Method)  at java.lang.reflect.Method.invoke(Method.java:511)  at com.android.internal.os.ZygoteInit$MethodAndArgsCaller.run(ZygoteInit.java:786)  at com.android.internal.os.ZygoteInit.main(ZygoteInit.java:553)  at dalvik.system.NativeStart.main(Native Method)  Caused by: java.util.concurrent.ExecutionException: com.android.volley.TimeoutError at com.android.volley.toolbox.RequestFuture.doGet(RequestFuture.java:117) at com.android.volley.toolbox.RequestFuture.get(RequestFuture.java:88) at com.morihacky.android.rxjava.volley.VolleyDemoFragment.getRouteData(VolleyDemoFragment.java:133) at com.morihacky.android.rxjava.volley.VolleyDemoFragment.access$000(VolleyDemoFragment.java:39) at com.morihacky.android.rxjava.volley.VolleyDemoFragment$1.call(VolleyDemoFragment.java:83) at com.morihacky.android.rxjava.volley.VolleyDemoFragment$1.call(VolleyDemoFragment.java:79) at rx.internal.operators.OnSubscribeDefer.call(OnSubscribeDefer.java:46) at rx.internal.operators.OnSubscribeDefer.call(OnSubscribeDefer.java:35) at rx.Observable.unsafeSubscribe(Observable.java:8452) at rx.internal.operators.OperatorSubscribeOn$1.call(OperatorSubscribeOn.java:94) at rx.internal.schedulers.CachedThreadScheduler$EventLoopWorker$1.call(CachedThreadScheduler.java:222) at rx.internal.schedulers.ScheduledAction.run(ScheduledAction.java:55) at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:442) at java.util.concurrent.FutureTask$Sync.innerRun(FutureTask.java:305) at java.util.concurrent.FutureTask.run(FutureTask.java:137) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.access$201(ScheduledThreadPoolExecutor.java:150) at java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:264) at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1076) at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:569) at java.lang.Thread.run(Thread.java:856) Caused by: com.android.volley.TimeoutError at com.android.volley.toolbox.BasicNetwork.performRequest(BasicNetwork.java:147) at com.android.volley.NetworkDispatcher.run(NetworkDispatcher.java:114)

The cause seems to be a timeout error in Volley; the site which is called (http://www.weather.com.cn/adat/sk/101010100.html) seems to be slow to respond on occasion. I've even had a request executing for 10+ seconds (and Volley default timeout is 5 seconds). The timeout error causes ExecutionException which should be caught and handled by the try/catch block in newGetRouteData method, however this causes something called a "CompositeException" which then kills the app.

I would also suggest avoiding TimeoutErrors even if this exception is solved; maybe use another site, something hosted by google or the like, or set a larger timeout. For instance, this should do the trick, before adding req to request queue:

req.setRetryPolicy(new DefaultRetryPolicy( (int) TimeUnit.SECONDS.toMillis(60), 1, DefaultRetryPolicy.DEFAULT_BACKOFF_MULT));

This sets timeout to 60 seconds, with 1 retry in case of a failure. Although just picking a faster site would probably be enough.

P.S. variable priority in getRoutData() is never used

[Question] Help Needed

Polling to Backend API in regular interval for certain number of times in a regular interval - Retrofit & RxJava

https://stackoverflow.com/questions/44920829/polling-to-backend-api-in-regular-interval-for-certain-number-of-times-in-a-regu

Observables Context??

Hi Kaushik,
It would be great if you can help with Observables,like
Which observable ,i can make use of when i was dealing with AsyncTask
and Which Observable I can make use of Login and SignUP and
Difference between and Disposable and Composite Disposable?

Gist link broken

The example under WIP for the cache/concat has a gist link that is broken

CacheMerge example updates entire adapter for each onNext

Hi,

Was checking out this repo (which is awesome btw)
and noticed that the CacheMerge example does some unnecessary updating (I think).

For each onNext() the entire adapter will be cleared and updated,
while there is only added one 'contributor' at a time.

This looks a bit inefficiently to me.

@Override
public void onNext(Pair<Contributor, Long> contributorAgePair) {
...
_adapter.clear();
_adapter.addAll(getListStringFromMap());
}

Only adding the 'Next' 'contributor' could be better or not?
@Override
public void onNext(Pair<Contributor, Long> contributorAgePair) {
...
String rowLog = String.format("%s [%d]", contributor.login, contributor.contributions);
_adapter.add(rowLog);
}

link is broken

1. DebouncedBuffer used for the fancier variant of the demo
   doesn't work

networkObserver get Exception so the final result is onError in your example when network fail even if localObeserver has data

networkObserver().publish(
network ->//
Observable.merge(network,//
localObserver().takeUntil(network)));

Sticky event bus

Hi.
First of all, thank you for the great samples, That's awesome!

I would like to get some advice. I really want to know how to implement sticky event with Rxbus. Any operator can reach it? I can't find any example for that, should i custom operator by .lift( )? If you feel it, what would you do?

I'm really really really confused. : (

Thanks.