Getting started with RxJava with sqlite

Question

I'm in the process of learning RxJava and have went thru several articles and videos. I totally felt awesome of what RxJava can offer, so I think currently getting the sense of what Reactive Programming is all about. This tutorial by Dan Lew helped me somewhat to understand the basics of using RxJava. But I found out that the more I thought I've fully understood RxJava, the more questions come popping out of my mind.

Since I'm a guy who learns most when writing code, here's what I'm trying to do. I already have an app that utilized SQLite. My app has 7 pie chart, which get its data from a Database cursor. Each of the pie chart gets a different row of data (float values) from the cursor, and generates the chart according to its data.

What I'm trying to do now is, I want to use RxJava to retrieve the data from the database and populate the chart. But I'm at loss on how RxJava should interact with database. Do I need to use any additional library to do this? Because I found rxjava-jdbc and SqlBrite which I think should help me with this, but do I really need them? I also found a question in SO which doesn't seem to use any additional library (except for RxJava of course) and he seems to use ContentObservable but I still don't fully get it. Besides the guy who answered his question doesn't even use ContentObservable. And I just knew that Android has its own Observable class.. is it fully compatible to be used together with RxJava? If not, why the same name?.

Someone please help me...

========================================================

In response to GreyBeardedGeek's solution below:

This is my Callable that retrieves data from sqlite

public class GetIncome implements Callable<Map<String,String>> {
    public static final String FILE_NAME = "CGetIncome";

    Context myContext;
    int year, month;

    public GetIncome(Context context, int getYear, int getMonth){
        myContext = context;
        year = getYear;
        month = getMonth;
    }

    @Override
    public Map<String,String> call() throws Exception {
        Map<String,String> output = new HashMap<>();
        JarControl jc = new JarControl(myContext);
        JSONObject getIncome = jc.getIncome(year,month,0,0);

        output.put("necessities",getIncome.getString("necessities"));
        output.put("savings",getIncome.getString("savings"));

        return output;
    }
}

And here's how I'm trying to use the Callable from my main activity:

Callable<Map<String,String>> getIncome = new GetIncome(getContext(),2015,9);

Now I don't know how to put the getIncome callable into Observables. You mentioned fromCallables in rx should do that.. but actually how to do it? According to this tutorial that shows how to use Observable.from, so I expect to be able to use Callables by doing Observable.fromCallable but apparently the function doesn't exist. Quick googled showed that .fromCallable should be used by doing something like Async.fromCallable(), but since I'm very new to RxJava, I can only understand how to create callable according to how the tutorial shows...

And sorry for my extreme noobness on RxJava

Answer 1

I'm sure that there are going to be a bunch of other opinions, but so far, I've taken a couple of different approaches:

• If you want a relatively small amount of data (and you can guarantee that it will always have a known upper bound on size), then:

  • create a synchronous method that accesses the database, gets a cursor, iterates the cursor to create a data structure (e.g. a List) containing the data, and then returns it.
  • Create a Callable that invokes the method
  • Use rx.fromCallable to create an an Observable

• If you need an unbounded number of rows, do something similar to the above, but return the Cursor.

EDIT: sample code:

private Map<String, String> getDataFromDatabase() { 
   Map<String, String> result = new HashMap<>(); 
   // do whatever you need to (i.e. db query, cursor) to fill it in 
   return result; 
} 

private Callable<Map<String, String>> getData() { 
   return new Callable() { 
      public Map<String, String> call() { 
        return getDataFromDatabase(); 
    } 
}

// in utility class  RxUtil.java 
public static <T> Observable<T> makeObservable(final Callable<T> func) {
        return Observable.create(
                new Observable.OnSubscribe<T>() {
                    @Override
                    public void call(Subscriber<? super T> subscriber) {
                        try {
                            T observed = func.call();
                            if (observed != null) { // to make defaultIfEmpty work
                                subscriber.onNext(observed);
                            }
                            subscriber.onCompleted();
                        } catch (Exception ex) {
                            subscriber.onError(ex);
                        }
                    }
                });
    }


// and finally, 
public Observable<Map<String, String>> getDataObservable() {
   return RxUtil.makeObservable(getData());
}