Reactive programming with RxJava Support

Reactive programming is one of the programming trends of this moment. If you want to get more information about it just visit the following site:

• RxJava
• Vogella tutorial

The main scope of Rx library is simplifying asynchronous tasks. In Android platform development, Rx library is a valid replacement of AsynchTask.

Kripton library contains suppports AsyncTask and Rx too. When you want to enable Rx supports in SQLite you have to:

• add Kripton Persistence library in gradle configuration file
• add in gradle configuration Rx library references:

dependencies {
 implementation 'io.reactivex.rxjava2:rxjava:2.1.5'
 implementation 'io.reactivex.rxjava2:rxandroid:2.0.1'
}

• enable in data source generation with attribute rx=true in @BindDataSource.

@BindDataSource(daoSet = {ContactDao.class}, fileName = "contact.db", rx = true)
public interface ContactDataSource {
}

And now?

Once Rx Kripton support is enabled, when a data-source is generated, it contains:

• A subject for each table: this feature allows to register an async listener to monitor table changes. When an insert/update/delete operation is executed on the associated table, an event was fired. This event is trapped by subject.
• Executed methods to execute the transaction or simply batch operations with Observable, Flowable, Single, Maybe.

Examples

Just for example. Suppose we have an application that manages Contact. So, we have to create a Contact entity, Dao Contact interface and Data Source interface.

Contact.class

@BindTable
public class Contact {
    public long id;
    public String name;
    public String surname;
}

The dao interface DaoContact.class

public interface ContactDao {
  @BindSqlInsert
  void insert(Contact bean);

  @BindSqlUpdate(where="id=${bean.id}")
  void update(Contact bean);

  @BindSqlDelete(where="id=${bean.id}")
  void delete(Contact bean);

  @BindSqlSelect(where="id=${id}")
  Contact selectById(long id);

  @BindSqlSelect
  List<Contact> selectAll();
}

And ContactDataSource.class

@BindDataSource(daoSet = {ContactDao.class}, fileName = "contact.db", rx = true)
public interface ContactDataSource {
}

In each generated DAO implementation will generate a method:

public PublishSubject<SQLiteEvent> subject() {
  return subject;
}

That exposes a Publish Subject.

The data source class will receive some methods just to support RX. Added method with RX=true:

• execute(FlowableTransaction<T>)
• execute(MaybeTransaction<T>)
• execute(ObservableTransaction<T>)
• execute(SingleTransaction<T>)
• executeBatch(FlowableBatch<T>)
• executeBatch(FlowableBatch<T>, boolean)
• executeBatch(MaybeBatch<T>)
• executeBatch(MaybeBatch<T>, boolean)
• executeBatch(ObservableBatch<T>)
• executeBatch(ObservableBatch<T>, boolean)
• executeBatch(SingleBatch<T>)
• executeBatch(SingleBatch<T>, boolean)

An usage example:

BindPhoneDataSource ds = BindPhoneDataSource.instance();
ds.executeBatch((BindPhoneDaoFactory daoFactory, ObservableEmitter<PhoneNumber> emitter) -> {
  Logger.info("subscribe on " + Thread.currentThread().getName());
  PhoneDaoImpl dao = daoFactory.getPhoneDao();
 
  PhoneNumber bean = new PhoneNumber();
  bean.action = random.nextBoolean() ? ActionType.ADD_PREFIX : ActionType.DO_NOTHING;
  bean.name = "Tonj Maner #" + random.nextInt();
  bean.number = "040" + random.nextInt(9) + random.nextInt(9) + random.nextInt(9) + random.nextInt(9) + random.nextInt(9);
 
  dao.insert(bean);
  emitter.onNext(bean);
 
  }, true)
.subscribeOn(Schedulers.io())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(result -> {
  Logger.info("observe on " + Thread.currentThread().getName());
});