Transactions no longer work after upgrading to 0.5.0

[mattkranzler5]

Transactions are broken in 0.5.0. I believe the commit that broke it might be 7cbd8d3.

[sockeqwe]

oh, I'm sorry!
Can you just share a little example of how do you typically implmenet a transaction?

[mattkranzler5]

Sure! Below is a method I use which no longer uses the transaction after the update. Maybe I'm using transactions wrong? It worked previously.

    public void saveAll(long showId, Collection<Episode> episodes) {
        BriteDatabase.Transaction transaction = newTransaction();
        deleteAll(showId);
        Observable.from(episodes)
                .subscribe(
                        episode -> insert(Episode.TABLE, episode.toContentValues(), SQLiteDatabase.CONFLICT_REPLACE)
                                .first()
                                .subscribeOn(Schedulers.io())
                                .observeOn(Schedulers.io())
                                .subscribe(
                                        __ -> Timber.d("Saved episode " + episode.getId()),
                                        throwable -> Timber.e(throwable, "An error occurred while saving episode " + episode.getId())
                                ),
                        throwable -> Timber.e(throwable, "An error occurred while syncing episodes."),
                        () -> {
                            transaction.markSuccessful();
                            transaction.end();
                        }
                );
    }

[sockeqwe]

The problem is that things run in different threads:

Observable.from(episodes) executes the transaction in onComplete()
BUT

insert(Episode.TABLE, episode.toContentValues(), SQLiteDatabase.CONFLICT_REPLACE)
                                .first()
                                .subscribeOn(Schedulers.io())
                                .observeOn(Schedulers.io())

runs in its own thread (Schedulers.io()).

So the transaction transaction.markSuccessful(); will be called before all insert(Episode.TABLE, episode.toContentValues(), SQLiteDatabase.CONFLICT_REPLACE) are completed.

So why was this working before SqlBrite-DAO0.5.0?
It has worked because in version 0.4.x the insert() call was blocking. That means that insert() was exceuted directly and .subscribeOn(Schedulers.io()) was useless because the result of insert() was already there since insert() has already been executed.

So there is a simple way to get back to the behaviour you already have implemented like this:

public void saveAll(long showId, Collection<Episode> episodes) {
        BriteDatabase.Transaction transaction = newTransaction();
        deleteAll(showId);
        Observable.from(episodes)
                .subscribe(
                        episode -> insert(Episode.TABLE, episode.toContentValues(), SQLiteDatabase.CONFLICT_REPLACE).toBlocking()
                        throwable -> Timber.e(throwable, "An error occurred while syncing episodes."),
                        () -> {
                            transaction.markSuccessful();
                            transaction.end();
                        }
                );
    }

However, your code is a little bit ugly and breaks the observable chain.
What about something like this (pseudo code; assumes deleteAll(showId) returns an Observable):

 BriteDatabase.Transaction transaction = newTransaction();
List<Observable> operationObservables = new ArrayList();
operationObservables.add(deleteAll(showId));

for (Episode e : episodes){
     operationObservables.add(insert(Episode.TABLE, episode.toContentValues(), SQLiteDatabase.CONFLICT_REPLACE));
}

 Observable producer = Observable.from(operationObservables);

Observable.concat(producer)
    .subscribeOn( ...)
    .observeOn( ... )
    .subscribe(
                       somethingNotReallyNeeded -> ... ,
                        throwable -> Timber.e(throwable, "An error occurred while syncing episodes."),
                        () -> {
                            transaction.markSuccessful();
                            transaction.end();
                        }

The idea is to concat all operations (deleteAll() and insert()) into one observable by using concat() operator. Than all this operations (each operation is a own observable) will be executed when subscribing the concat observable and the transaction gets commited only if all of this operations has been executed without any error.

[mattkranzler5]

Awesome, thank you for the suggestions! I realized I was breaking the chain in this case which I was ok with, but I wasn't aware of the behavior you mentioned. I'll do some refactoring after upgrading to 0.5.0.

[mattkranzler5]

Hey @sockeqwe, I tried out your suggestions and it looks as though the subscription never completes and therefore the transaction is never marked successful. Here is the current code I'm using:

BriteDatabase.Transaction transaction = newTransaction();
        List<Observable> operations = new ArrayList<>();
        operations.add(deleteAll(showId));
        for (Episode episode : episodes) {
            operations.add(insert(Episode.TABLE, episode.toContentValues(), SQLiteDatabase.CONFLICT_REPLACE));
        }
        Observable producer = Observable.from(operations);
        Observable.concat(producer)
                .subscribeOn(Schedulers.io())
                .observeOn(Schedulers.io())
                .subscribe(
                        __ -> {},
                        throwable -> {},
                        () -> {
                            transaction.markSuccessful();
                            transaction.end();
                            Timber.d("Saved episodes for show " + showId);
                        }
                );

Any ideas?

[mattkranzler5]

Also, here is the deleteAll() method:

public Observable<Integer> deleteAll(long seriesId) {
        return delete(Episode.TABLE, Episode.SERIES_ID + " = ?", String.valueOf(seriesId))
                .doOnNext(count -> Timber.d("Deleted " + count + " episodes for series " + seriesId + "."))
                .doOnError(throwable -> Timber.e(throwable, "An error occurred while deleting episodes for " + seriesId + "."));
    }

[mattkranzler5]

Interesting, if I remove the schedulers when subscribing to the Observable.concat() it completes successfully, but with them it never completes.

[sockeqwe]

I think this is an internal issue of sqlbrite and not of this library. I think square has fixed that in sqlbrite-0.6.0 (released yesterday).
https://twitter.com/JakeWharton/status/700022994798845952

I will update this library to 0.6.0 and then you can check if the issue has been fixed.

btw. a transaction.end(); should also be in the onError handling to rollback the changes.

[sockeqwe]

Ok, I have release 0.6.0 which uses sqlbrite 0.6.0 (shouldn't require any code migration from 0.5.0 to 0.6.0). Please check if the problem has been solved.