Android DbFlow Update/Delete/Save models in batch with RxJava - ReactiveX with DbFlow

DeleteModelInBatchOnSubscribe.java

import android.database.Cursor;
import android.support.annotation.NonNull;

import com.raizlabs.android.dbflow.runtime.DBTransactionInfo;
import com.raizlabs.android.dbflow.runtime.TransactionManager;
import com.raizlabs.android.dbflow.runtime.transaction.BaseTransaction;
import com.raizlabs.android.dbflow.runtime.transaction.QueryTransaction;
import com.raizlabs.android.dbflow.runtime.transaction.TransactionListener;
import com.raizlabs.android.dbflow.sql.builder.ConditionQueryBuilder;
import com.raizlabs.android.dbflow.sql.language.Delete;
import com.raizlabs.android.dbflow.sql.language.Where;
import com.raizlabs.android.dbflow.structure.Model;

import rx.Observable;
import rx.Subscriber;
import rx.schedulers.Schedulers;

public class DeleteModelInBatchOnSubscribe<ModelClass extends Model> implements Observable.OnSubscribe<Void>, TransactionListener<Cursor> {

    private final Where<ModelClass> where;
    private final Class<ModelClass> modelType;
    private Subscriber<? super Void> subscriber;

    public DeleteModelInBatchOnSubscribe(@NonNull ConditionQueryBuilder<ModelClass> whereConditionBuilder, Class<ModelClass> modelType) {
        this.where = new Delete().from(whereConditionBuilder.getTableClass()).where(whereConditionBuilder);
        this.modelType = modelType;
    }

    public static <ModelClass extends Model> Observable<Void> toObservable(@NonNull ConditionQueryBuilder<ModelClass> whereConditionBuilder, Class<ModelClass> modelType) {
        return createObservable(new DeleteModelInBatchOnSubscribe<>(whereConditionBuilder, modelType));
    }

    private static <ModelClass extends Model> Observable<Void> createObservable(DeleteModelInBatchOnSubscribe<ModelClass> instance) {
        return Observable.create(instance).subscribeOn(Schedulers.io());
    }

    public Observable<Void> toObservable() {
        return createObservable(this);
    }

    @Override
    public void call(Subscriber<? super Void> subscriber) {
        this.subscriber = subscriber;

        if (!this.subscriber.isUnsubscribed()) {
//            Logger.d("Starting BatchDelete Transaction for: %s\nQuery: %s", where.getTable(), where.toString());
            TransactionManager.getInstance().addTransaction(new QueryTransaction<ModelClass>(DBTransactionInfo.create(BaseTransaction.PRIORITY_NORMAL), this.where, this));
        }
    }

    @Override
    public void onResultReceived(Cursor cursor) {
//        Logger.d("BatchDelete TRANSACTION for %s completed", where.getTable());
        if (!this.subscriber.isUnsubscribed()) {
            this.subscriber.onNext(null);
            this.subscriber.onCompleted();
        }
    }

    @Override
    public boolean onReady(BaseTransaction<Cursor> baseTransaction) {
        return true;
    }

    @Override
    public boolean hasResult(BaseTransaction<Cursor> baseTransaction, Cursor cursor) {
        return true;
    }
}

SaveModelInBatchOnSubscribe.java

import android.support.annotation.NonNull;

import com.orhanobut.logger.Logger;
import com.raizlabs.android.dbflow.runtime.DBTransactionInfo;
import com.raizlabs.android.dbflow.runtime.TransactionManager;
import com.raizlabs.android.dbflow.runtime.transaction.BaseTransaction;
import com.raizlabs.android.dbflow.runtime.transaction.TransactionListener;
import com.raizlabs.android.dbflow.runtime.transaction.process.InsertModelTransaction;
import com.raizlabs.android.dbflow.runtime.transaction.process.ProcessModelInfo;
import com.raizlabs.android.dbflow.structure.Model;

import java.util.List;

import rx.Observable;
import rx.Subscriber;
import rx.schedulers.Schedulers;

public class SaveModelInBatchOnSubscribe<ModelClass extends Model> implements Observable.OnSubscribe<Void>, TransactionListener<List<ModelClass>> {

    private final List<ModelClass> models;
    private final Class<ModelClass> modelType;
    private Subscriber<? super Void> subscriber;

    public SaveModelInBatchOnSubscribe(@NonNull List<ModelClass> models, Class<ModelClass> modelType) {
        this.models = models;
        this.modelType = modelType;
    }

    public static <ModelClass extends Model> Observable<Void> toObservable(@NonNull List<ModelClass> models, Class<ModelClass> modelType) {
        return Observable.create(new SaveModelInBatchOnSubscribe<>(models, modelType)).subscribeOn(Schedulers.io());
    }

    @Override
    public void call(Subscriber<? super Void> subscriber) {
        this.subscriber = subscriber;

        Logger.d("Starting BatchSave TRANSACTION for: %s", modelType.getName());

        if (models.isEmpty()) {
            Logger.d("BatchSave TRANSACTION for: %s, skipping - no data", modelType.getName());

            if (!subscriber.isUnsubscribed()) {
                subscriber.onNext(null);
                subscriber.onCompleted();
            }
        } else {
            Logger.d("BatchSave TRANSACTION for: %s, total records for save: %d", modelType.getName(), models.size());
            ProcessModelInfo<ModelClass> processModelInfo =
                    ProcessModelInfo
                            .withModels(models)
                            .result(this)
                            .info(DBTransactionInfo.create(BaseTransaction.PRIORITY_NORMAL));

            TransactionManager.getInstance().addTransaction(new InsertModelTransaction<>(processModelInfo));
        }
    }

    @Override
    public void onResultReceived(List<ModelClass> modelClasses) {
        Logger.d("BatchSave TRANSACTION for: %s DONE, saved %d record(s)", modelType.getName(), modelClasses.size());
        if (subscriber != null && !subscriber.isUnsubscribed()) {
            subscriber.onNext(null);
            subscriber.onCompleted();
        }
    }

    @Override
    public boolean onReady(BaseTransaction<List<ModelClass>> baseTransaction) {
        return true;
    }

    @Override
    public boolean hasResult(BaseTransaction<List<ModelClass>> baseTransaction, List<ModelClass> modelClasses) {
        return true;
    }
}

UpdateModelInBatchOnSubscribe.java

import android.database.Cursor;
import android.support.annotation.NonNull;

import com.raizlabs.android.dbflow.runtime.DBTransactionInfo;
import com.raizlabs.android.dbflow.runtime.TransactionManager;
import com.raizlabs.android.dbflow.runtime.transaction.BaseTransaction;
import com.raizlabs.android.dbflow.runtime.transaction.QueryTransaction;
import com.raizlabs.android.dbflow.runtime.transaction.TransactionListener;
import com.raizlabs.android.dbflow.sql.builder.Condition;
import com.raizlabs.android.dbflow.sql.builder.ConditionQueryBuilder;
import com.raizlabs.android.dbflow.sql.language.Update;
import com.raizlabs.android.dbflow.sql.language.Where;
import com.raizlabs.android.dbflow.structure.Model;

import rx.Observable;
import rx.Subscriber;
import rx.schedulers.Schedulers;

public class UpdateModelInBatchOnSubscribe<ModelClass extends Model> implements Observable.OnSubscribe<Void>, TransactionListener<Cursor> {

    private final Where<ModelClass> where;
    private final Class<ModelClass> modelType;
    private Subscriber<? super Void> subscriber;

    /**
     * From DBFlow Update in batch to Observable
     *
     * @param whereConditionBuilder example: new ConditionQueryBuilder&lt;&gt;(Conversation.class, Condition.column(Conversation$Table.CONVERSATIONID).in(headId, tailIdArray))
     * @param setConditions         example: Condition.column(Conversation$Table.READ).eq(Boolean.TRUE)
     */
    public UpdateModelInBatchOnSubscribe(@NonNull ConditionQueryBuilder<ModelClass> whereConditionBuilder, Class<ModelClass> modelType, Condition... setConditions) {
        this.modelType = modelType;
        this.where = (new Update<>(whereConditionBuilder.getTableClass())).set(setConditions).where(whereConditionBuilder);
    }

    private static <ModelClass extends Model> Observable<Void> createObservable(UpdateModelInBatchOnSubscribe<ModelClass> instance) {
        return Observable.create(instance).subscribeOn(Schedulers.io());
    }

    public Observable<Void> toObservable() {
        return createObservable(this);
    }

    @Override
    public void call(Subscriber<? super Void> subscriber) {
        this.subscriber = subscriber;

        if (!this.subscriber.isUnsubscribed()) {
//            Logger.d("Starting BatchUpdate TRANSACTION for: %s\nQuery: %s", where.getTable(), where.toString());
            TransactionManager.getInstance().addTransaction(new QueryTransaction<ModelClass>(DBTransactionInfo.create(BaseTransaction.PRIORITY_NORMAL), this.where, this));
        }
    }

    @Override
    public void onResultReceived(Cursor cursor) {
//        Logger.d("BatchUpdate TRANSACTION for %s completed", where.getTable());
        if (!this.subscriber.isUnsubscribed()) {
            this.subscriber.onNext(null);
            this.subscriber.onCompleted();
        }
    }

    @Override
    public boolean onReady(BaseTransaction<Cursor> baseTransaction) {
        return true;
    }

    @Override
    public boolean hasResult(BaseTransaction<Cursor> baseTransaction, Cursor cursor) {
        return true;
    }
}