trajakovic · May 2, 2016 12:51 · May 2, 2016
diff --git a/DeleteModelInBatchOnSubscribe.java b/DeleteModelInBatchOnSubscribe.java
@@ -0,0 +1,69 @@
+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;
+    }
+}
diff --git a/SaveModelInBatchOnSubscribe.java b/SaveModelInBatchOnSubscribe.java
@@ -0,0 +1,76 @@
+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;
+    }
+}
diff --git a/UpdateModelInBatchOnSubscribe.java b/UpdateModelInBatchOnSubscribe.java
@@ -0,0 +1,72 @@
+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;
+    }
+}