Serialize Transactions and allow configuring transaction executor.
Add an API for setting a different executor to execute database
transactions. Different from the Query Executor, Room will not use more
than one thread from the transaction executor since transactions are
exclusive by default there is no point in using multiple threads to have
them just wait for other transactions to finish.
Bug: 120226549
Test: TransactionExecutorTest, SuspendingQueryTest and others.
Change-Id: Ie34be23619867d8ccb9534d2386f5be5724cf235
diff --git a/room/compiler/src/main/kotlin/androidx/room/processor/MethodProcessorDelegate.kt b/room/compiler/src/main/kotlin/androidx/room/processor/MethodProcessorDelegate.kt
index 8dc5742..4d8cfa7 100644
--- a/room/compiler/src/main/kotlin/androidx/room/processor/MethodProcessorDelegate.kt
+++ b/room/compiler/src/main/kotlin/androidx/room/processor/MethodProcessorDelegate.kt
@@ -201,9 +201,10 @@
adapter = context.typeAdapterStore.findPreparedQueryResultAdapter(returnType, query)
) { callableImpl, dbField ->
addStatement(
- "return $T.execute($N, $L, $N)",
+ "return $T.execute($N, $L, $L, $N)",
RoomCoroutinesTypeNames.COROUTINES_ROOM,
dbField,
+ "true", // inTransaction
callableImpl,
continuationParam.simpleName.toString()
)
@@ -217,9 +218,10 @@
adapter = context.typeAdapterStore.findInsertAdapter(returnType, params)
) { callableImpl, dbField ->
addStatement(
- "return $T.execute($N, $L, $N)",
+ "return $T.execute($N, $L, $L, $N)",
RoomCoroutinesTypeNames.COROUTINES_ROOM,
dbField,
+ "true", // inTransaction
callableImpl,
continuationParam.simpleName.toString()
)
@@ -231,9 +233,10 @@
adapter = context.typeAdapterStore.findDeleteOrUpdateAdapter(returnType)
) { callableImpl, dbField ->
addStatement(
- "return $T.execute($N, $L, $N)",
+ "return $T.execute($N, $L, $L, $N)",
RoomCoroutinesTypeNames.COROUTINES_ROOM,
dbField,
+ "true", // inTransaction
callableImpl,
continuationParam.simpleName.toString()
)
diff --git a/room/compiler/src/main/kotlin/androidx/room/solver/prepared/binderprovider/GuavaListenableFuturePreparedQueryResultBinderProvider.kt b/room/compiler/src/main/kotlin/androidx/room/solver/prepared/binderprovider/GuavaListenableFuturePreparedQueryResultBinderProvider.kt
index 95e4e71..54b3d23 100644
--- a/room/compiler/src/main/kotlin/androidx/room/solver/prepared/binderprovider/GuavaListenableFuturePreparedQueryResultBinderProvider.kt
+++ b/room/compiler/src/main/kotlin/androidx/room/solver/prepared/binderprovider/GuavaListenableFuturePreparedQueryResultBinderProvider.kt
@@ -52,9 +52,10 @@
adapter = context.typeAdapterStore.findPreparedQueryResultAdapter(typeArg, query)
) { callableImpl, dbField ->
addStatement(
- "return $T.createListenableFuture($N, $L)",
+ "return $T.createListenableFuture($N, $L, $L)",
RoomGuavaTypeNames.GUAVA_ROOM,
dbField,
+ "true", // inTransaction
callableImpl
)
}
diff --git a/room/compiler/src/main/kotlin/androidx/room/solver/query/result/CoroutineResultBinder.kt b/room/compiler/src/main/kotlin/androidx/room/solver/query/result/CoroutineResultBinder.kt
index b59ecc0..d97ba61 100644
--- a/room/compiler/src/main/kotlin/androidx/room/solver/query/result/CoroutineResultBinder.kt
+++ b/room/compiler/src/main/kotlin/androidx/room/solver/query/result/CoroutineResultBinder.kt
@@ -57,9 +57,10 @@
scope.builder().apply {
addStatement(
- "return $T.execute($N, $L, $N)",
+ "return $T.execute($N, $L, $L, $N)",
RoomCoroutinesTypeNames.COROUTINES_ROOM,
dbField,
+ if (inTransaction) "true" else "false",
callableImpl,
continuationParamName)
}
diff --git a/room/compiler/src/main/kotlin/androidx/room/solver/query/result/GuavaListenableFutureQueryResultBinder.kt b/room/compiler/src/main/kotlin/androidx/room/solver/query/result/GuavaListenableFutureQueryResultBinder.kt
index 7141786..722242e 100644
--- a/room/compiler/src/main/kotlin/androidx/room/solver/query/result/GuavaListenableFutureQueryResultBinder.kt
+++ b/room/compiler/src/main/kotlin/androidx/room/solver/query/result/GuavaListenableFutureQueryResultBinder.kt
@@ -56,9 +56,10 @@
scope.builder().apply {
addStatement(
- "return $T.createListenableFuture($N, $L, $L, $L)",
+ "return $T.createListenableFuture($N, $L, $L, $L, $L)",
RoomGuavaTypeNames.GUAVA_ROOM,
dbField,
+ if (inTransaction) "true" else "false",
callableImpl,
roomSQLiteQueryVar,
canReleaseQuery
diff --git a/room/compiler/src/main/kotlin/androidx/room/solver/query/result/LiveDataQueryResultBinder.kt b/room/compiler/src/main/kotlin/androidx/room/solver/query/result/LiveDataQueryResultBinder.kt
index 2eebcda5..cd84015 100644
--- a/room/compiler/src/main/kotlin/androidx/room/solver/query/result/LiveDataQueryResultBinder.kt
+++ b/room/compiler/src/main/kotlin/androidx/room/solver/query/result/LiveDataQueryResultBinder.kt
@@ -67,8 +67,12 @@
scope.builder().apply {
val tableNamesList = tableNames.joinToString(",") { "\"$it\"" }
addStatement(
- "return $N.getInvalidationTracker().createLiveData(new $T{$L}, $L)",
- dbField, String::class.arrayTypeName(), tableNamesList, callableImpl
+ "return $N.getInvalidationTracker().createLiveData(new $T{$L}, $L, $L)",
+ dbField,
+ String::class.arrayTypeName(),
+ tableNamesList,
+ if (inTransaction) "true" else "false",
+ callableImpl
)
}
}
diff --git a/room/compiler/src/main/kotlin/androidx/room/solver/query/result/RxQueryResultBinder.kt b/room/compiler/src/main/kotlin/androidx/room/solver/query/result/RxQueryResultBinder.kt
index 7b2e54b..d52e804 100644
--- a/room/compiler/src/main/kotlin/androidx/room/solver/query/result/RxQueryResultBinder.kt
+++ b/room/compiler/src/main/kotlin/androidx/room/solver/query/result/RxQueryResultBinder.kt
@@ -58,9 +58,14 @@
}.build()
scope.builder().apply {
val tableNamesList = queryTableNames.joinToString(",") { "\"$it\"" }
- addStatement("return $T.$N($N, new $T{$L}, $L)",
- RoomRxJava2TypeNames.RX_ROOM, rxType.methodName, dbField,
- String::class.arrayTypeName(), tableNamesList, callableImpl)
+ addStatement("return $T.$N($N, $L, new $T{$L}, $L)",
+ RoomRxJava2TypeNames.RX_ROOM,
+ rxType.methodName,
+ dbField,
+ if (inTransaction) "true" else "false",
+ String::class.arrayTypeName(),
+ tableNamesList,
+ callableImpl)
}
}
diff --git a/room/compiler/src/main/kotlin/androidx/room/solver/shortcut/binderprovider/GuavaListenableFutureDeleteOrUpdateMethodBinderProvider.kt b/room/compiler/src/main/kotlin/androidx/room/solver/shortcut/binderprovider/GuavaListenableFutureDeleteOrUpdateMethodBinderProvider.kt
index b43e23c..106bc14 100644
--- a/room/compiler/src/main/kotlin/androidx/room/solver/shortcut/binderprovider/GuavaListenableFutureDeleteOrUpdateMethodBinderProvider.kt
+++ b/room/compiler/src/main/kotlin/androidx/room/solver/shortcut/binderprovider/GuavaListenableFutureDeleteOrUpdateMethodBinderProvider.kt
@@ -54,9 +54,10 @@
val adapter = context.typeAdapterStore.findDeleteOrUpdateAdapter(typeArg)
return createDeleteOrUpdateBinder(typeArg, adapter) { callableImpl, dbField ->
addStatement(
- "return $T.createListenableFuture($N, $L)",
+ "return $T.createListenableFuture($N, $L, $L)",
RoomGuavaTypeNames.GUAVA_ROOM,
dbField,
+ "true", // inTransaction
callableImpl
)
}
diff --git a/room/compiler/src/main/kotlin/androidx/room/solver/shortcut/binderprovider/GuavaListenableFutureInsertMethodBinderProvider.kt b/room/compiler/src/main/kotlin/androidx/room/solver/shortcut/binderprovider/GuavaListenableFutureInsertMethodBinderProvider.kt
index 7b3cb47..f603510 100644
--- a/room/compiler/src/main/kotlin/androidx/room/solver/shortcut/binderprovider/GuavaListenableFutureInsertMethodBinderProvider.kt
+++ b/room/compiler/src/main/kotlin/androidx/room/solver/shortcut/binderprovider/GuavaListenableFutureInsertMethodBinderProvider.kt
@@ -58,9 +58,10 @@
val adapter = context.typeAdapterStore.findInsertAdapter(typeArg, params)
return createInsertBinder(typeArg, adapter) { callableImpl, dbField ->
addStatement(
- "return $T.createListenableFuture($N, $L)",
+ "return $T.createListenableFuture($N, $L, $L)",
RoomGuavaTypeNames.GUAVA_ROOM,
dbField,
+ "true", // inTransaction
callableImpl
)
}
diff --git a/room/compiler/src/test/data/daoWriter/output/ComplexDao.java b/room/compiler/src/test/data/daoWriter/output/ComplexDao.java
index bb945a3..f047ecd4 100644
--- a/room/compiler/src/test/data/daoWriter/output/ComplexDao.java
+++ b/room/compiler/src/test/data/daoWriter/output/ComplexDao.java
@@ -281,7 +281,7 @@
final RoomSQLiteQuery _statement = RoomSQLiteQuery.acquire(_sql, 1);
int _argIndex = 1;
_statement.bindLong(_argIndex, id);
- return __db.getInvalidationTracker().createLiveData(new String[]{"user"}, new Callable<User>() {
+ return __db.getInvalidationTracker().createLiveData(new String[]{"user"}, false, new Callable<User>() {
@Override
public User call() throws Exception {
final Cursor _cursor = DBUtil.query(__db, _statement, false);
@@ -330,7 +330,7 @@
_statement.bindLong(_argIndex, _item);
_argIndex ++;
}
- return __db.getInvalidationTracker().createLiveData(new String[]{"user"}, new Callable<List<User>>() {
+ return __db.getInvalidationTracker().createLiveData(new String[]{"user"}, false, new Callable<List<User>>() {
@Override
public List<User> call() throws Exception {
final Cursor _cursor = DBUtil.query(__db, _statement, false);
diff --git a/room/guava/src/main/java/androidx/room/guava/GuavaRoom.java b/room/guava/src/main/java/androidx/room/guava/GuavaRoom.java
index d4f6c2f..4e2f500 100644
--- a/room/guava/src/main/java/androidx/room/guava/GuavaRoom.java
+++ b/room/guava/src/main/java/androidx/room/guava/GuavaRoom.java
@@ -48,8 +48,8 @@
* Returns a {@link ListenableFuture<T>} created by submitting the input {@code callable} to
* {@link ArchTaskExecutor}'s background-threaded Executor.
*
- * @deprecated
- * Use {@link #createListenableFuture(RoomDatabase, Callable, RoomSQLiteQuery, boolean)}
+ * @deprecated Use {@link #createListenableFuture(RoomDatabase, boolean, Callable,
+ * RoomSQLiteQuery, boolean)}
*/
@Deprecated
public static <T> ListenableFuture<T> createListenableFuture(
@@ -63,7 +63,11 @@
/**
* Returns a {@link ListenableFuture<T>} created by submitting the input {@code callable} to
* {@link RoomDatabase}'s {@link java.util.concurrent.Executor}.
+ *
+ * @deprecated Use {@link #createListenableFuture(RoomDatabase, boolean, Callable,
+ * RoomSQLiteQuery, boolean)}
*/
+ @Deprecated
public static <T> ListenableFuture<T> createListenableFuture(
final RoomDatabase roomDatabase,
final Callable<T> callable,
@@ -73,6 +77,20 @@
roomDatabase.getQueryExecutor(), callable, query, releaseQuery);
}
+ /**
+ * Returns a {@link ListenableFuture<T>} created by submitting the input {@code callable} to
+ * {@link RoomDatabase}'s {@link java.util.concurrent.Executor}.
+ */
+ public static <T> ListenableFuture<T> createListenableFuture(
+ final RoomDatabase roomDatabase,
+ final boolean inTransaction,
+ final Callable<T> callable,
+ final RoomSQLiteQuery query,
+ final boolean releaseQuery) {
+ return createListenableFuture(
+ getExecutor(roomDatabase, inTransaction), callable, query, releaseQuery);
+ }
+
private static <T> ListenableFuture<T> createListenableFuture(
final Executor executor,
final Callable<T> callable,
@@ -104,12 +122,34 @@
/**
* Returns a {@link ListenableFuture<T>} created by submitting the input {@code callable} to
* {@link RoomDatabase}'s {@link java.util.concurrent.Executor}.
+ *
+ * @deprecated Use {@link #createListenableFuture(RoomDatabase, boolean, Callable)}
*/
+ @Deprecated
public static <T> ListenableFuture<T> createListenableFuture(
final RoomDatabase roomDatabase,
final Callable<T> callable) {
+ return createListenableFuture(roomDatabase, false, callable);
+ }
+
+ /**
+ * Returns a {@link ListenableFuture<T>} created by submitting the input {@code callable} to
+ * {@link RoomDatabase}'s {@link java.util.concurrent.Executor}.
+ */
+ public static <T> ListenableFuture<T> createListenableFuture(
+ final RoomDatabase roomDatabase,
+ final boolean inTransaction,
+ final Callable<T> callable) {
ListenableFutureTask<T> listenableFutureTask = ListenableFutureTask.create(callable);
- roomDatabase.getQueryExecutor().execute(listenableFutureTask);
+ getExecutor(roomDatabase, inTransaction).execute(listenableFutureTask);
return listenableFutureTask;
}
+
+ private static Executor getExecutor(RoomDatabase database, boolean inTransaction) {
+ if (inTransaction) {
+ return database.getTransactionExecutor();
+ } else {
+ return database.getQueryExecutor();
+ }
+ }
}
diff --git a/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/test/SuspendingQueryTest.kt b/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/test/SuspendingQueryTest.kt
index 901f410..b8e81fa 100644
--- a/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/test/SuspendingQueryTest.kt
+++ b/room/integration-tests/kotlintestapp/src/androidTest/java/androidx/room/integration/kotlintestapp/test/SuspendingQueryTest.kt
@@ -17,6 +17,7 @@
package androidx.room.integration.kotlintestapp.test
import android.os.Build
+import androidx.arch.core.executor.ArchTaskExecutor
import androidx.room.Room
import androidx.room.RoomDatabase
import androidx.room.integration.kotlintestapp.NewThreadDispatcher
@@ -45,8 +46,10 @@
import org.junit.runner.RunWith
import java.io.IOException
import java.util.concurrent.CountDownLatch
+import java.util.concurrent.ExecutorService
import java.util.concurrent.Executors
import java.util.concurrent.TimeUnit
+import java.util.concurrent.atomic.AtomicInteger
@LargeTest
@RunWith(AndroidJUnit4::class)
@@ -633,6 +636,52 @@
@Test
@Suppress("DeferredResultUnused")
+ fun withTransaction_multipleTransactions_verifyThreadUsage() {
+ val busyThreadsCount = AtomicInteger()
+ // Executor wrapper that counts threads that are busy executing commands.
+ class WrappedService(val delegate: ExecutorService) : ExecutorService by delegate {
+ override fun execute(command: Runnable) {
+ delegate.execute {
+ busyThreadsCount.incrementAndGet()
+ try {
+ command.run()
+ } finally {
+ busyThreadsCount.decrementAndGet()
+ }
+ }
+ }
+ }
+ val wrappedExecutor = WrappedService(Executors.newCachedThreadPool())
+ val localDatabase = Room.inMemoryDatabaseBuilder(
+ ApplicationProvider.getApplicationContext(), TestDatabase::class.java)
+ .setQueryExecutor(ArchTaskExecutor.getIOThreadExecutor())
+ .setTransactionExecutor(wrappedExecutor)
+ .build()
+
+ // Run two parallel transactions but verify that only 1 thread is busy when the transactions
+ // execute, indicating that threads are not busy waiting on sql connections but are instead
+ // suspended.
+ runBlocking(Dispatchers.IO) {
+ async {
+ localDatabase.withTransaction {
+ delay(200) // delay a bit to let the other transaction proceed
+ assertThat(busyThreadsCount.get()).isEqualTo(1)
+ }
+ }
+
+ async {
+ localDatabase.withTransaction {
+ delay(200) // delay a bit to let the other transaction proceed
+ assertThat(busyThreadsCount.get()).isEqualTo(1)
+ }
+ }
+ }
+
+ wrappedExecutor.awaitTermination(1, TimeUnit.SECONDS)
+ }
+
+ @Test
+ @Suppress("DeferredResultUnused")
fun withTransaction_leakTransactionContext_async() {
runBlocking {
val leakedContext = database.withTransaction {
@@ -700,7 +749,7 @@
val executorService = Executors.newSingleThreadExecutor()
val localDatabase = Room.inMemoryDatabaseBuilder(
ApplicationProvider.getApplicationContext(), TestDatabase::class.java)
- .setQueryExecutor(executorService)
+ .setTransactionExecutor(executorService)
.build()
// Simulate a busy executor, no thread to acquire for transaction.
@@ -735,7 +784,7 @@
val executorService = Executors.newCachedThreadPool()
val localDatabase = Room.inMemoryDatabaseBuilder(
ApplicationProvider.getApplicationContext(), TestDatabase::class.java)
- .setQueryExecutor(executorService)
+ .setTransactionExecutor(executorService)
.build()
executorService.shutdownNow()
diff --git a/room/integration-tests/testapp/src/androidTest/java/androidx/room/integration/testapp/test/InvalidationTrackerTest.java b/room/integration-tests/testapp/src/androidTest/java/androidx/room/integration/testapp/test/InvalidationTrackerTest.java
index d49779a..6b88e33 100644
--- a/room/integration-tests/testapp/src/androidTest/java/androidx/room/integration/testapp/test/InvalidationTrackerTest.java
+++ b/room/integration-tests/testapp/src/androidTest/java/androidx/room/integration/testapp/test/InvalidationTrackerTest.java
@@ -125,7 +125,7 @@
public void createLiveData() throws ExecutionException, InterruptedException, TimeoutException {
final LiveData<Item> liveData = mDb
.getInvalidationTracker()
- .createLiveData(new String[]{"Item"}, () -> mDb.getItemDao().itemById(1));
+ .createLiveData(new String[]{"Item"}, false, () -> mDb.getItemDao().itemById(1));
mDb.getItemDao().insert(new Item(1, "v1"));
@@ -147,7 +147,7 @@
throws ExecutionException, InterruptedException, TimeoutException {
LiveData<Item> liveData = mDb
.getInvalidationTracker()
- .createLiveData(new String[]{"Item"}, () -> mDb.getItemDao().itemById(1));
+ .createLiveData(new String[]{"Item"}, false, () -> mDb.getItemDao().itemById(1));
mDb.getItemDao().insert(new Item(1, "v1"));
diff --git a/room/ktx/api/2.1.0-alpha06.txt b/room/ktx/api/2.1.0-alpha06.txt
index f5dcd16..851eb86 100644
--- a/room/ktx/api/2.1.0-alpha06.txt
+++ b/room/ktx/api/2.1.0-alpha06.txt
@@ -1,6 +1,10 @@
// Signature format: 3.0
package androidx.room {
+ public final class CoroutinesRoomKt {
+ ctor public CoroutinesRoomKt();
+ }
+
public final class RoomDatabaseKt {
ctor public RoomDatabaseKt();
method public static suspend Object? acquireTransactionThread(java.util.concurrent.Executor, kotlinx.coroutines.Job controlJob, kotlin.coroutines.experimental.Continuation<? super kotlin.coroutines.ContinuationInterceptor> p);
diff --git a/room/ktx/api/current.txt b/room/ktx/api/current.txt
index f5dcd16..851eb86 100644
--- a/room/ktx/api/current.txt
+++ b/room/ktx/api/current.txt
@@ -1,6 +1,10 @@
// Signature format: 3.0
package androidx.room {
+ public final class CoroutinesRoomKt {
+ ctor public CoroutinesRoomKt();
+ }
+
public final class RoomDatabaseKt {
ctor public RoomDatabaseKt();
method public static suspend Object? acquireTransactionThread(java.util.concurrent.Executor, kotlinx.coroutines.Job controlJob, kotlin.coroutines.experimental.Continuation<? super kotlin.coroutines.ContinuationInterceptor> p);
diff --git a/room/ktx/api/restricted_2.1.0-alpha06.txt b/room/ktx/api/restricted_2.1.0-alpha06.txt
index f31bf4d..866bb4d 100644
--- a/room/ktx/api/restricted_2.1.0-alpha06.txt
+++ b/room/ktx/api/restricted_2.1.0-alpha06.txt
@@ -2,12 +2,12 @@
package androidx.room {
@RestrictTo({RestrictTo.Scope.LIBRARY_GROUP_PREFIX}) public final class CoroutinesRoom {
- method public static suspend <R> Object? execute(androidx.room.RoomDatabase p, java.util.concurrent.Callable<R> db, kotlin.coroutines.experimental.Continuation<? super R> callable);
+ method public static suspend <R> Object? execute(androidx.room.RoomDatabase p, boolean db, java.util.concurrent.Callable<R> inTransaction, kotlin.coroutines.experimental.Continuation<? super R> callable);
field public static final androidx.room.CoroutinesRoom.Companion! Companion;
}
public static final class CoroutinesRoom.Companion {
- method public suspend <R> Object? execute(androidx.room.RoomDatabase db, java.util.concurrent.Callable<R> callable, kotlin.coroutines.experimental.Continuation<? super R> p);
+ method public suspend <R> Object? execute(androidx.room.RoomDatabase db, boolean inTransaction, java.util.concurrent.Callable<R> callable, kotlin.coroutines.experimental.Continuation<? super R> p);
}
}
diff --git a/room/ktx/api/restricted_current.txt b/room/ktx/api/restricted_current.txt
index f31bf4d..866bb4d 100644
--- a/room/ktx/api/restricted_current.txt
+++ b/room/ktx/api/restricted_current.txt
@@ -2,12 +2,12 @@
package androidx.room {
@RestrictTo({RestrictTo.Scope.LIBRARY_GROUP_PREFIX}) public final class CoroutinesRoom {
- method public static suspend <R> Object? execute(androidx.room.RoomDatabase p, java.util.concurrent.Callable<R> db, kotlin.coroutines.experimental.Continuation<? super R> callable);
+ method public static suspend <R> Object? execute(androidx.room.RoomDatabase p, boolean db, java.util.concurrent.Callable<R> inTransaction, kotlin.coroutines.experimental.Continuation<? super R> callable);
field public static final androidx.room.CoroutinesRoom.Companion! Companion;
}
public static final class CoroutinesRoom.Companion {
- method public suspend <R> Object? execute(androidx.room.RoomDatabase db, java.util.concurrent.Callable<R> callable, kotlin.coroutines.experimental.Continuation<? super R> p);
+ method public suspend <R> Object? execute(androidx.room.RoomDatabase db, boolean inTransaction, java.util.concurrent.Callable<R> callable, kotlin.coroutines.experimental.Continuation<? super R> p);
}
}
diff --git a/room/ktx/src/main/java/androidx/room/CoroutinesRoom.kt b/room/ktx/src/main/java/androidx/room/CoroutinesRoom.kt
index f4e3cc9..d38ed58 100644
--- a/room/ktx/src/main/java/androidx/room/CoroutinesRoom.kt
+++ b/room/ktx/src/main/java/androidx/room/CoroutinesRoom.kt
@@ -17,6 +17,7 @@
package androidx.room
import androidx.annotation.RestrictTo
+import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.asCoroutineDispatcher
import kotlinx.coroutines.withContext
import java.util.concurrent.Callable
@@ -33,18 +34,42 @@
companion object {
@JvmStatic
- suspend fun <R> execute(db: RoomDatabase, callable: Callable<R>): R {
+ suspend fun <R> execute(
+ db: RoomDatabase,
+ inTransaction: Boolean,
+ callable: Callable<R>
+ ): R {
if (db.isOpen && db.inTransaction()) {
return callable.call()
}
// Use the transaction dispatcher if we are on a transaction coroutine, otherwise
- // use the query executor as dispatcher.
+ // use the database dispatchers.
val context = coroutineContext[TransactionElement]?.transactionDispatcher
- ?: db.queryExecutor.asCoroutineDispatcher()
+ ?: if (inTransaction) db.transactionDispatcher else db.queryDispatcher
return withContext(context) {
callable.call()
}
}
}
-}
\ No newline at end of file
+}
+
+/**
+ * Gets the query coroutine dispatcher.
+ *
+ * @hide
+ */
+internal val RoomDatabase.queryDispatcher: CoroutineDispatcher
+ get() = backingFieldMap.getOrPut("QueryDispatcher") {
+ queryExecutor.asCoroutineDispatcher()
+ } as CoroutineDispatcher
+
+/**
+ * Gets the transaction coroutine dispatcher.
+ *
+ * @hide
+ */
+internal val RoomDatabase.transactionDispatcher: CoroutineDispatcher
+ get() = backingFieldMap.getOrPut("TransactionDispatcher") {
+ queryExecutor.asCoroutineDispatcher()
+ } as CoroutineDispatcher
diff --git a/room/ktx/src/main/java/androidx/room/RoomDatabase.kt b/room/ktx/src/main/java/androidx/room/RoomDatabase.kt
index 86f1fde..49a972a 100644
--- a/room/ktx/src/main/java/androidx/room/RoomDatabase.kt
+++ b/room/ktx/src/main/java/androidx/room/RoomDatabase.kt
@@ -91,7 +91,7 @@
*/
private suspend fun RoomDatabase.createTransactionContext(): CoroutineContext {
val controlJob = Job()
- val dispatcher = queryExecutor.acquireTransactionThread(controlJob)
+ val dispatcher = transactionExecutor.acquireTransactionThread(controlJob)
val transactionElement = TransactionElement(controlJob, dispatcher)
val threadLocalElement =
suspendingTransactionId.asContextElement(System.identityHashCode(controlJob))
diff --git a/room/runtime/api/2.1.0-alpha06.txt b/room/runtime/api/2.1.0-alpha06.txt
index 00d448a..882b112 100644
--- a/room/runtime/api/2.1.0-alpha06.txt
+++ b/room/runtime/api/2.1.0-alpha06.txt
@@ -15,6 +15,7 @@
field public final java.util.concurrent.Executor queryExecutor;
field public final boolean requireMigration;
field public final androidx.sqlite.db.SupportSQLiteOpenHelper.Factory sqliteOpenHelperFactory;
+ field public final java.util.concurrent.Executor transactionExecutor;
}
public class InvalidationTracker {
@@ -48,6 +49,7 @@
method public androidx.room.InvalidationTracker getInvalidationTracker();
method public androidx.sqlite.db.SupportSQLiteOpenHelper getOpenHelper();
method public java.util.concurrent.Executor getQueryExecutor();
+ method public java.util.concurrent.Executor getTransactionExecutor();
method public boolean inTransaction();
method @CallSuper public void init(androidx.room.DatabaseConfiguration);
method protected void internalInitInvalidationTracker(androidx.sqlite.db.SupportSQLiteDatabase);
@@ -73,6 +75,7 @@
method public androidx.room.RoomDatabase.Builder<T> openHelperFactory(androidx.sqlite.db.SupportSQLiteOpenHelper.Factory?);
method public androidx.room.RoomDatabase.Builder<T> setJournalMode(androidx.room.RoomDatabase.JournalMode);
method public androidx.room.RoomDatabase.Builder<T> setQueryExecutor(java.util.concurrent.Executor);
+ method public androidx.room.RoomDatabase.Builder<T> setTransactionExecutor(java.util.concurrent.Executor);
}
public abstract static class RoomDatabase.Callback {
diff --git a/room/runtime/api/current.txt b/room/runtime/api/current.txt
index 00d448a..882b112 100644
--- a/room/runtime/api/current.txt
+++ b/room/runtime/api/current.txt
@@ -15,6 +15,7 @@
field public final java.util.concurrent.Executor queryExecutor;
field public final boolean requireMigration;
field public final androidx.sqlite.db.SupportSQLiteOpenHelper.Factory sqliteOpenHelperFactory;
+ field public final java.util.concurrent.Executor transactionExecutor;
}
public class InvalidationTracker {
@@ -48,6 +49,7 @@
method public androidx.room.InvalidationTracker getInvalidationTracker();
method public androidx.sqlite.db.SupportSQLiteOpenHelper getOpenHelper();
method public java.util.concurrent.Executor getQueryExecutor();
+ method public java.util.concurrent.Executor getTransactionExecutor();
method public boolean inTransaction();
method @CallSuper public void init(androidx.room.DatabaseConfiguration);
method protected void internalInitInvalidationTracker(androidx.sqlite.db.SupportSQLiteDatabase);
@@ -73,6 +75,7 @@
method public androidx.room.RoomDatabase.Builder<T> openHelperFactory(androidx.sqlite.db.SupportSQLiteOpenHelper.Factory?);
method public androidx.room.RoomDatabase.Builder<T> setJournalMode(androidx.room.RoomDatabase.JournalMode);
method public androidx.room.RoomDatabase.Builder<T> setQueryExecutor(java.util.concurrent.Executor);
+ method public androidx.room.RoomDatabase.Builder<T> setTransactionExecutor(java.util.concurrent.Executor);
}
public abstract static class RoomDatabase.Callback {
diff --git a/room/runtime/api/restricted_2.1.0-alpha06.txt b/room/runtime/api/restricted_2.1.0-alpha06.txt
index 10b8084..a792dc1 100644
--- a/room/runtime/api/restricted_2.1.0-alpha06.txt
+++ b/room/runtime/api/restricted_2.1.0-alpha06.txt
@@ -2,7 +2,7 @@
package androidx.room {
public class DatabaseConfiguration {
- ctor @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public DatabaseConfiguration(android.content.Context, String?, androidx.sqlite.db.SupportSQLiteOpenHelper.Factory, androidx.room.RoomDatabase.MigrationContainer, java.util.List<androidx.room.RoomDatabase.Callback>?, boolean, androidx.room.RoomDatabase.JournalMode!, java.util.concurrent.Executor, boolean, boolean, boolean, java.util.Set<java.lang.Integer>?);
+ ctor @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public DatabaseConfiguration(android.content.Context, String?, androidx.sqlite.db.SupportSQLiteOpenHelper.Factory, androidx.room.RoomDatabase.MigrationContainer, java.util.List<androidx.room.RoomDatabase.Callback>?, boolean, androidx.room.RoomDatabase.JournalMode!, java.util.concurrent.Executor, java.util.concurrent.Executor, boolean, boolean, boolean, java.util.Set<java.lang.Integer>?);
}
@RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public abstract class EntityDeletionOrUpdateAdapter<T> extends androidx.room.SharedSQLiteStatement {
@@ -32,7 +32,8 @@
ctor @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public InvalidationTracker(androidx.room.RoomDatabase!, java.lang.String...!);
ctor @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public InvalidationTracker(androidx.room.RoomDatabase!, java.util.Map<java.lang.String,java.lang.String>!, java.util.Map<java.lang.String,java.util.Set<java.lang.String>>!, java.lang.String...!);
method @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public void addWeakObserver(androidx.room.InvalidationTracker.Observer!);
- method @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public <T> androidx.lifecycle.LiveData<T>! createLiveData(String[]!, java.util.concurrent.Callable<T>!);
+ method @Deprecated @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public <T> androidx.lifecycle.LiveData<T>! createLiveData(String[]!, java.util.concurrent.Callable<T>!);
+ method @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public <T> androidx.lifecycle.LiveData<T>! createLiveData(String[]!, boolean, java.util.concurrent.Callable<T>!);
method @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) @WorkerThread public void refreshVersionsSync();
}
diff --git a/room/runtime/api/restricted_current.txt b/room/runtime/api/restricted_current.txt
index 10b8084..a792dc1 100644
--- a/room/runtime/api/restricted_current.txt
+++ b/room/runtime/api/restricted_current.txt
@@ -2,7 +2,7 @@
package androidx.room {
public class DatabaseConfiguration {
- ctor @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public DatabaseConfiguration(android.content.Context, String?, androidx.sqlite.db.SupportSQLiteOpenHelper.Factory, androidx.room.RoomDatabase.MigrationContainer, java.util.List<androidx.room.RoomDatabase.Callback>?, boolean, androidx.room.RoomDatabase.JournalMode!, java.util.concurrent.Executor, boolean, boolean, boolean, java.util.Set<java.lang.Integer>?);
+ ctor @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public DatabaseConfiguration(android.content.Context, String?, androidx.sqlite.db.SupportSQLiteOpenHelper.Factory, androidx.room.RoomDatabase.MigrationContainer, java.util.List<androidx.room.RoomDatabase.Callback>?, boolean, androidx.room.RoomDatabase.JournalMode!, java.util.concurrent.Executor, java.util.concurrent.Executor, boolean, boolean, boolean, java.util.Set<java.lang.Integer>?);
}
@RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public abstract class EntityDeletionOrUpdateAdapter<T> extends androidx.room.SharedSQLiteStatement {
@@ -32,7 +32,8 @@
ctor @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public InvalidationTracker(androidx.room.RoomDatabase!, java.lang.String...!);
ctor @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public InvalidationTracker(androidx.room.RoomDatabase!, java.util.Map<java.lang.String,java.lang.String>!, java.util.Map<java.lang.String,java.util.Set<java.lang.String>>!, java.lang.String...!);
method @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public void addWeakObserver(androidx.room.InvalidationTracker.Observer!);
- method @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public <T> androidx.lifecycle.LiveData<T>! createLiveData(String[]!, java.util.concurrent.Callable<T>!);
+ method @Deprecated @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public <T> androidx.lifecycle.LiveData<T>! createLiveData(String[]!, java.util.concurrent.Callable<T>!);
+ method @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public <T> androidx.lifecycle.LiveData<T>! createLiveData(String[]!, boolean, java.util.concurrent.Callable<T>!);
method @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) @WorkerThread public void refreshVersionsSync();
}
diff --git a/room/runtime/build.gradle b/room/runtime/build.gradle
index c7e7958..620d499 100644
--- a/room/runtime/build.gradle
+++ b/room/runtime/build.gradle
@@ -46,6 +46,7 @@
testImplementation(MOCKITO_CORE)
testImplementation(ARCH_LIFECYCLE_EXTENSIONS)
testImplementation(KOTLIN_STDLIB)
+ testImplementation(TRUTH)
androidTestImplementation(JUNIT)
androidTestImplementation(TEST_EXT_JUNIT)
diff --git a/room/runtime/src/main/java/androidx/room/DatabaseConfiguration.java b/room/runtime/src/main/java/androidx/room/DatabaseConfiguration.java
index bb893fd..dae3a09 100644
--- a/room/runtime/src/main/java/androidx/room/DatabaseConfiguration.java
+++ b/room/runtime/src/main/java/androidx/room/DatabaseConfiguration.java
@@ -74,6 +74,12 @@
public final Executor queryExecutor;
/**
+ * The Executor used to execute asynchronous transactions.
+ */
+ @NonNull
+ public final Executor transactionExecutor;
+
+ /**
* If true, table invalidation in an instance of {@link RoomDatabase} is broadcast and
* synchronized with other instances of the same {@link RoomDatabase} file, including those
* in a separate process.
@@ -124,6 +130,7 @@
boolean allowMainThreadQueries,
RoomDatabase.JournalMode journalMode,
@NonNull Executor queryExecutor,
+ @NonNull Executor transactionExecutor,
boolean multiInstanceInvalidation,
boolean requireMigration,
boolean allowDestructiveMigrationOnDowngrade,
@@ -136,6 +143,7 @@
this.allowMainThreadQueries = allowMainThreadQueries;
this.journalMode = journalMode;
this.queryExecutor = queryExecutor;
+ this.transactionExecutor = transactionExecutor;
this.multiInstanceInvalidation = multiInstanceInvalidation;
this.requireMigration = requireMigration;
this.allowDestructiveMigrationOnDowngrade = allowDestructiveMigrationOnDowngrade;
diff --git a/room/runtime/src/main/java/androidx/room/InvalidationLiveDataContainer.java b/room/runtime/src/main/java/androidx/room/InvalidationLiveDataContainer.java
index 4168f47..e1e8155 100644
--- a/room/runtime/src/main/java/androidx/room/InvalidationLiveDataContainer.java
+++ b/room/runtime/src/main/java/androidx/room/InvalidationLiveDataContainer.java
@@ -43,8 +43,10 @@
mDatabase = database;
}
- <T> LiveData<T> create(String[] tableNames, Callable<T> computeFunction) {
- return new RoomTrackingLiveData<>(mDatabase, this, computeFunction, tableNames);
+ <T> LiveData<T> create(String[] tableNames, boolean inTransaction,
+ Callable<T> computeFunction) {
+ return new RoomTrackingLiveData<>(mDatabase, this, inTransaction, computeFunction,
+ tableNames);
}
void onActive(LiveData liveData) {
diff --git a/room/runtime/src/main/java/androidx/room/InvalidationTracker.java b/room/runtime/src/main/java/androidx/room/InvalidationTracker.java
index fe33a55..28ba653 100644
--- a/room/runtime/src/main/java/androidx/room/InvalidationTracker.java
+++ b/room/runtime/src/main/java/androidx/room/InvalidationTracker.java
@@ -554,6 +554,8 @@
* <p>
* Holds a strong reference to the created LiveData as long as it is active.
*
+ * @deprecated Use {@link #createLiveData(String[], boolean, Callable)}
+ *
* @param computeFunction The function that calculates the value
* @param tableNames The list of tables to observe
* @param <T> The return type
@@ -561,10 +563,32 @@
* invalidates.
* @hide
*/
+ @Deprecated
@RestrictTo(RestrictTo.Scope.LIBRARY_GROUP_PREFIX)
public <T> LiveData<T> createLiveData(String[] tableNames, Callable<T> computeFunction) {
+ return createLiveData(tableNames, false, computeFunction);
+ }
+
+ /**
+ * Creates a LiveData that computes the given function once and for every other invalidation
+ * of the database.
+ * <p>
+ * Holds a strong reference to the created LiveData as long as it is active.
+ *
+ * @param tableNames The list of tables to observe
+ * @param inTransaction True if the computeFunction will be done in a transaction, false
+ * otherwise.
+ * @param computeFunction The function that calculates the value
+ * @param <T> The return type
+ * @return A new LiveData that computes the given function when the given list of tables
+ * invalidates.
+ * @hide
+ */
+ @RestrictTo(RestrictTo.Scope.LIBRARY_GROUP_PREFIX)
+ public <T> LiveData<T> createLiveData(String[] tableNames, boolean inTransaction,
+ Callable<T> computeFunction) {
return mInvalidationLiveDataContainer.create(
- validateAndResolveTableNames(tableNames), computeFunction);
+ validateAndResolveTableNames(tableNames), inTransaction, computeFunction);
}
/**
diff --git a/room/runtime/src/main/java/androidx/room/RoomDatabase.java b/room/runtime/src/main/java/androidx/room/RoomDatabase.java
index a33383a..7cd0536 100644
--- a/room/runtime/src/main/java/androidx/room/RoomDatabase.java
+++ b/room/runtime/src/main/java/androidx/room/RoomDatabase.java
@@ -45,8 +45,10 @@
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.Callable;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executor;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -77,6 +79,7 @@
@Deprecated
protected volatile SupportSQLiteDatabase mDatabase;
private Executor mQueryExecutor;
+ private Executor mTransactionExecutor;
private SupportSQLiteOpenHelper mOpenHelper;
private final InvalidationTracker mInvalidationTracker;
private boolean mAllowMainThreadQueries;
@@ -121,6 +124,19 @@
return mSuspendingTransactionId;
}
+
+ private final Map<String, Object> mBackingFieldMap = new ConcurrentHashMap<>();
+
+ /**
+ * Gets the map for storing extension properties of Kotlin type.
+ *
+ * @hide
+ */
+ @RestrictTo(RestrictTo.Scope.LIBRARY_GROUP)
+ Map<String, Object> getBackingFieldMap() {
+ return mBackingFieldMap;
+ }
+
/**
* Creates a RoomDatabase.
* <p>
@@ -147,6 +163,7 @@
}
mCallbacks = configuration.callbacks;
mQueryExecutor = configuration.queryExecutor;
+ mTransactionExecutor = new TransactionExecutor(configuration.transactionExecutor);
mAllowMainThreadQueries = configuration.allowMainThreadQueries;
mWriteAheadLoggingEnabled = wal;
if (configuration.multiInstanceInvalidation) {
@@ -336,6 +353,14 @@
}
/**
+ * @return The Executor in use by this database for async transactions.
+ */
+ @NonNull
+ public Executor getTransactionExecutor() {
+ return mTransactionExecutor;
+ }
+
+ /**
* Wrapper for {@link SupportSQLiteDatabase#setTransactionSuccessful()}.
*
* @deprecated Use {@link #runInTransaction(Runnable)}
@@ -481,6 +506,8 @@
/** The Executor used to run database queries. This should be background-threaded. */
private Executor mQueryExecutor;
+ /** The Executor used to run database transactions. This should be background-threaded. */
+ private Executor mTransactionExecutor;
private SupportSQLiteOpenHelper.Factory mFactory;
private boolean mAllowMainThreadQueries;
private JournalMode mJournalMode;
@@ -598,12 +625,19 @@
* queries and tasks, including {@code LiveData} invalidation, {@code Flowable} scheduling
* and {@code ListenableFuture} tasks.
* <p>
- * When unset, a default {@code Executor} will be used. The default {@code Executor}
- * allocates and shares threads amongst Architecture Components libraries.
+ * When both the query executor and transaction executor are unset, then a default
+ * {@code Executor} will be used. The default {@code Executor} allocates and shares threads
+ * amongst Architecture Components libraries. If the query executor is unset but a
+ * transaction executor was set, then the same {@code Executor} will be used for queries.
+ * <p>
+ * For best performance the given {@code Executor} should be bounded (max number of threads
+ * is limited).
* <p>
* The input {@code Executor} cannot run tasks on the UI thread.
- *
+ **
* @return this
+ *
+ * @see #setTransactionExecutor(Executor)
*/
@NonNull
public Builder<T> setQueryExecutor(@NonNull Executor executor) {
@@ -612,6 +646,32 @@
}
/**
+ * Sets the {@link Executor} that will be used to execute all non-blocking asynchronous
+ * transaction queries and tasks, including {@code LiveData} invalidation, {@code Flowable}
+ * scheduling and {@code ListenableFuture} tasks.
+ * <p>
+ * When both the transaction executor and query executor are unset, then a default
+ * {@code Executor} will be used. The default {@code Executor} allocates and shares threads
+ * amongst Architecture Components libraries. If the transaction executor is unset but a
+ * query executor was set, then the same {@code Executor} will be used for transactions.
+ * <p>
+ * If the given {@code Executor} is shared then it should be unbounded to avoid the
+ * possibility of a deadlock. Room will not use more than one thread at a time from this
+ * executor.
+ * <p>
+ * The input {@code Executor} cannot run tasks on the UI thread.
+ *
+ * @return this
+ *
+ * @see #setQueryExecutor(Executor)
+ */
+ @NonNull
+ public Builder<T> setTransactionExecutor(@NonNull Executor executor) {
+ mTransactionExecutor = executor;
+ return this;
+ }
+
+ /**
* Sets whether table invalidation in this instance of {@link RoomDatabase} should be
* broadcast and synchronized with other instances of the same {@link RoomDatabase},
* including those in a separate process. In order to enable multi-instance invalidation,
@@ -741,8 +801,12 @@
throw new IllegalArgumentException("Must provide an abstract class that"
+ " extends RoomDatabase");
}
- if (mQueryExecutor == null) {
- mQueryExecutor = ArchTaskExecutor.getIOThreadExecutor();
+ if (mQueryExecutor == null && mTransactionExecutor == null) {
+ mQueryExecutor = mTransactionExecutor = ArchTaskExecutor.getIOThreadExecutor();
+ } else if (mQueryExecutor != null && mTransactionExecutor == null) {
+ mTransactionExecutor = mQueryExecutor;
+ } else if (mQueryExecutor == null && mTransactionExecutor != null) {
+ mQueryExecutor = mTransactionExecutor;
}
if (mMigrationStartAndEndVersions != null && mMigrationsNotRequiredFrom != null) {
@@ -763,12 +827,20 @@
mFactory = new FrameworkSQLiteOpenHelperFactory();
}
DatabaseConfiguration configuration =
- new DatabaseConfiguration(mContext, mName, mFactory, mMigrationContainer,
- mCallbacks, mAllowMainThreadQueries, mJournalMode.resolve(mContext),
+ new DatabaseConfiguration(
+ mContext,
+ mName,
+ mFactory,
+ mMigrationContainer,
+ mCallbacks,
+ mAllowMainThreadQueries,
+ mJournalMode.resolve(mContext),
mQueryExecutor,
+ mTransactionExecutor,
mMultiInstanceInvalidation,
mRequireMigration,
- mAllowDestructiveMigrationOnDowngrade, mMigrationsNotRequiredFrom);
+ mAllowDestructiveMigrationOnDowngrade,
+ mMigrationsNotRequiredFrom);
T db = Room.getGeneratedImplementation(mDatabaseClass, DB_IMPL_SUFFIX);
db.init(configuration);
return db;
diff --git a/room/runtime/src/main/java/androidx/room/RoomTrackingLiveData.java b/room/runtime/src/main/java/androidx/room/RoomTrackingLiveData.java
index 61ef38e..8df1014 100644
--- a/room/runtime/src/main/java/androidx/room/RoomTrackingLiveData.java
+++ b/room/runtime/src/main/java/androidx/room/RoomTrackingLiveData.java
@@ -27,6 +27,7 @@
import java.util.Set;
import java.util.concurrent.Callable;
+import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicBoolean;
/**
@@ -48,6 +49,9 @@
final RoomDatabase mDatabase;
@SuppressWarnings("WeakerAccess")
+ final boolean mInTransaction;
+
+ @SuppressWarnings("WeakerAccess")
final Callable<T> mComputeFunction;
private final InvalidationLiveDataContainer mContainer;
@@ -116,7 +120,7 @@
boolean isActive = hasActiveObservers();
if (mInvalid.compareAndSet(false, true)) {
if (isActive) {
- mDatabase.getQueryExecutor().execute(mRefreshRunnable);
+ getQueryExecutor().execute(mRefreshRunnable);
}
}
}
@@ -125,9 +129,11 @@
RoomTrackingLiveData(
RoomDatabase database,
InvalidationLiveDataContainer container,
+ boolean inTransaction,
Callable<T> computeFunction,
String[] tableNames) {
mDatabase = database;
+ mInTransaction = inTransaction;
mComputeFunction = computeFunction;
mContainer = container;
mObserver = new InvalidationTracker.Observer(tableNames) {
@@ -142,7 +148,7 @@
protected void onActive() {
super.onActive();
mContainer.onActive(this);
- mDatabase.getQueryExecutor().execute(mRefreshRunnable);
+ getQueryExecutor().execute(mRefreshRunnable);
}
@Override
@@ -150,4 +156,12 @@
super.onInactive();
mContainer.onInactive(this);
}
+
+ Executor getQueryExecutor() {
+ if (mInTransaction) {
+ return mDatabase.getTransactionExecutor();
+ } else {
+ return mDatabase.getQueryExecutor();
+ }
+ }
}
diff --git a/room/runtime/src/main/java/androidx/room/TransactionExecutor.java b/room/runtime/src/main/java/androidx/room/TransactionExecutor.java
new file mode 100644
index 0000000..6a6bc12
--- /dev/null
+++ b/room/runtime/src/main/java/androidx/room/TransactionExecutor.java
@@ -0,0 +1,62 @@
+/*
+ * Copyright 2019 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package androidx.room;
+
+import androidx.annotation.NonNull;
+
+import java.util.ArrayDeque;
+import java.util.concurrent.Executor;
+
+/**
+ * Executor wrapper for performing database transactions serially.
+ * <p>
+ * Since database transactions are exclusive, this executor ensures that transactions are performed
+ * in-order and one at a time, preventing threads from blocking each other when multiple concurrent
+ * transactions are attempted.
+ */
+class TransactionExecutor implements Executor {
+
+ private final Executor mExecutor;
+ private final ArrayDeque<Runnable> mTasks = new ArrayDeque<>();
+ private Runnable mActive;
+
+ TransactionExecutor(@NonNull Executor executor) {
+ mExecutor = executor;
+ }
+
+ public synchronized void execute(final Runnable command) {
+ mTasks.offer(new Runnable() {
+ public void run() {
+ try {
+ command.run();
+ } finally {
+ scheduleNext();
+ }
+ }
+ });
+ if (mActive == null) {
+ scheduleNext();
+ }
+ }
+
+ @SuppressWarnings("WeakerAccess")
+ synchronized void scheduleNext() {
+ if ((mActive = mTasks.poll()) != null) {
+ mExecutor.execute(mActive);
+ }
+ }
+}
diff --git a/room/runtime/src/test/java/androidx/room/BuilderTest.java b/room/runtime/src/test/java/androidx/room/BuilderTest.java
index 0c29dd7..eabefdb 100644
--- a/room/runtime/src/test/java/androidx/room/BuilderTest.java
+++ b/room/runtime/src/test/java/androidx/room/BuilderTest.java
@@ -39,6 +39,7 @@
import org.junit.runners.JUnit4;
import java.util.List;
+import java.util.concurrent.Executor;
@SuppressWarnings("ArraysAsListWithZeroOrOneArgument")
@RunWith(JUnit4.class)
@@ -66,6 +67,41 @@
Room.databaseBuilder(mock(Context.class), RoomDatabase.class, " ").build();
}
+ public void executors_setQueryExecutor() {
+ Executor executor = mock(Executor.class);
+
+ TestDatabase db = Room.databaseBuilder(mock(Context.class), TestDatabase.class, "foo")
+ .setQueryExecutor(executor)
+ .build();
+
+ assertThat(db.mDatabaseConfiguration.queryExecutor, is(executor));
+ assertThat(db.mDatabaseConfiguration.transactionExecutor, is(executor));
+ }
+
+ public void executors_setTransactionExecutor() {
+ Executor executor = mock(Executor.class);
+
+ TestDatabase db = Room.databaseBuilder(mock(Context.class), TestDatabase.class, "foo")
+ .setTransactionExecutor(executor)
+ .build();
+
+ assertThat(db.mDatabaseConfiguration.queryExecutor, is(executor));
+ assertThat(db.mDatabaseConfiguration.transactionExecutor, is(executor));
+ }
+
+ public void executors_setBothExecutors() {
+ Executor executor1 = mock(Executor.class);
+ Executor executor2 = mock(Executor.class);
+
+ TestDatabase db = Room.databaseBuilder(mock(Context.class), TestDatabase.class, "foo")
+ .setQueryExecutor(executor1)
+ .setTransactionExecutor(executor2)
+ .build();
+
+ assertThat(db.mDatabaseConfiguration.queryExecutor, is(executor1));
+ assertThat(db.mDatabaseConfiguration.transactionExecutor, is(executor2));
+ }
+
@Test
public void migration() {
Migration m1 = new EmptyMigration(0, 1);
@@ -387,6 +423,14 @@
}
abstract static class TestDatabase extends RoomDatabase {
+
+ DatabaseConfiguration mDatabaseConfiguration;
+
+ @Override
+ public void init(@NonNull DatabaseConfiguration configuration) {
+ super.init(configuration);
+ mDatabaseConfiguration = configuration;
+ }
}
static class EmptyMigration extends Migration {
diff --git a/room/runtime/src/test/java/androidx/room/InvalidationLiveDataContainerTest.kt b/room/runtime/src/test/java/androidx/room/InvalidationLiveDataContainerTest.kt
index 97c22d1..6034b67 100644
--- a/room/runtime/src/test/java/androidx/room/InvalidationLiveDataContainerTest.kt
+++ b/room/runtime/src/test/java/androidx/room/InvalidationLiveDataContainerTest.kt
@@ -99,6 +99,7 @@
private fun createLiveData(): LiveData<Any> {
return container.create(
arrayOf("a", "b"),
+ false,
createComputeFunction<Any>()
) as LiveData
}
diff --git a/room/runtime/src/test/java/androidx/room/TransactionExecutorTest.kt b/room/runtime/src/test/java/androidx/room/TransactionExecutorTest.kt
new file mode 100644
index 0000000..edda059
--- /dev/null
+++ b/room/runtime/src/test/java/androidx/room/TransactionExecutorTest.kt
@@ -0,0 +1,87 @@
+/*
+ * Copyright 2019 The Android Open Source Project
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package androidx.room
+
+import com.google.common.truth.Truth.assertThat
+import org.junit.After
+import org.junit.Test
+import org.junit.runner.RunWith
+import org.junit.runners.JUnit4
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.Executors
+import java.util.concurrent.TimeUnit
+
+@RunWith(JUnit4::class)
+class TransactionExecutorTest {
+
+ private val testExecutor = Executors.newCachedThreadPool()
+ private val transactionExecutor = TransactionExecutor(testExecutor)
+
+ @After
+ fun teardown() {
+ testExecutor.shutdownNow()
+ }
+
+ @Test
+ @Throws(InterruptedException::class)
+ fun testSerialExecution() {
+
+ val latch = CountDownLatch(3)
+ val runnableA = TimingRunnable(latch)
+ val runnableB = TimingRunnable(latch)
+ val runnableC = TimingRunnable(latch)
+
+ transactionExecutor.execute(runnableA)
+ transactionExecutor.execute(runnableB)
+ transactionExecutor.execute(runnableC)
+
+ // Await for the runnables to finish.
+ latch.await(1, TimeUnit.SECONDS)
+
+ // Assert that everything ran.
+ assertThat(runnableA.run).isTrue()
+ assertThat(runnableB.run).isTrue()
+ assertThat(runnableC.run).isTrue()
+
+ // Assert that runnables were run in order of submission.
+ assertThat(runnableA.start).isLessThan(runnableB.start)
+ assertThat(runnableB.start).isLessThan(runnableC.start)
+
+ // Assert that a runnable finishes before the runnable after it starts.
+ assertThat(runnableA.finish).isLessThan(runnableB.start)
+ assertThat(runnableB.finish).isLessThan(runnableC.start)
+ }
+
+ private class TimingRunnable(val latch: CountDownLatch) : Runnable {
+ var start: Long = 0
+ var finish: Long = 0
+ var run: Boolean = false
+
+ override fun run() {
+ run = true
+ start = System.nanoTime()
+ try {
+ // Sleep for a bit as if we were doing real work.
+ Thread.sleep(100)
+ } catch (e: InterruptedException) {
+ throw RuntimeException(e)
+ }
+ finish = System.nanoTime()
+ latch.countDown()
+ }
+ }
+}
\ No newline at end of file
diff --git a/room/rxjava2/api/restricted_2.1.0-alpha06.txt b/room/rxjava2/api/restricted_2.1.0-alpha06.txt
index 36290b8..6731dfa 100644
--- a/room/rxjava2/api/restricted_2.1.0-alpha06.txt
+++ b/room/rxjava2/api/restricted_2.1.0-alpha06.txt
@@ -2,8 +2,10 @@
package androidx.room {
public class RxRoom {
- method @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public static <T> io.reactivex.Flowable<T>! createFlowable(androidx.room.RoomDatabase!, String[]!, java.util.concurrent.Callable<T>!);
- method @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public static <T> io.reactivex.Observable<T>! createObservable(androidx.room.RoomDatabase!, String[]!, java.util.concurrent.Callable<T>!);
+ method @Deprecated @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public static <T> io.reactivex.Flowable<T>! createFlowable(androidx.room.RoomDatabase!, String[]!, java.util.concurrent.Callable<T>!);
+ method @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public static <T> io.reactivex.Flowable<T>! createFlowable(androidx.room.RoomDatabase!, boolean, String[]!, java.util.concurrent.Callable<T>!);
+ method @Deprecated @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public static <T> io.reactivex.Observable<T>! createObservable(androidx.room.RoomDatabase!, String[]!, java.util.concurrent.Callable<T>!);
+ method @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public static <T> io.reactivex.Observable<T>! createObservable(androidx.room.RoomDatabase!, boolean, String[]!, java.util.concurrent.Callable<T>!);
}
}
diff --git a/room/rxjava2/api/restricted_current.txt b/room/rxjava2/api/restricted_current.txt
index 36290b8..6731dfa 100644
--- a/room/rxjava2/api/restricted_current.txt
+++ b/room/rxjava2/api/restricted_current.txt
@@ -2,8 +2,10 @@
package androidx.room {
public class RxRoom {
- method @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public static <T> io.reactivex.Flowable<T>! createFlowable(androidx.room.RoomDatabase!, String[]!, java.util.concurrent.Callable<T>!);
- method @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public static <T> io.reactivex.Observable<T>! createObservable(androidx.room.RoomDatabase!, String[]!, java.util.concurrent.Callable<T>!);
+ method @Deprecated @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public static <T> io.reactivex.Flowable<T>! createFlowable(androidx.room.RoomDatabase!, String[]!, java.util.concurrent.Callable<T>!);
+ method @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public static <T> io.reactivex.Flowable<T>! createFlowable(androidx.room.RoomDatabase!, boolean, String[]!, java.util.concurrent.Callable<T>!);
+ method @Deprecated @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public static <T> io.reactivex.Observable<T>! createObservable(androidx.room.RoomDatabase!, String[]!, java.util.concurrent.Callable<T>!);
+ method @RestrictTo(androidx.annotation.RestrictTo.Scope.LIBRARY_GROUP_PREFIX) public static <T> io.reactivex.Observable<T>! createObservable(androidx.room.RoomDatabase!, boolean, String[]!, java.util.concurrent.Callable<T>!);
}
}
diff --git a/room/rxjava2/src/main/java/androidx/room/RxRoom.java b/room/rxjava2/src/main/java/androidx/room/RxRoom.java
index 1e78572..3badde1 100644
--- a/room/rxjava2/src/main/java/androidx/room/RxRoom.java
+++ b/room/rxjava2/src/main/java/androidx/room/RxRoom.java
@@ -20,6 +20,7 @@
import java.util.Set;
import java.util.concurrent.Callable;
+import java.util.concurrent.Executor;
import io.reactivex.BackpressureStrategy;
import io.reactivex.Flowable;
@@ -97,12 +98,27 @@
* Helper method used by generated code to bind a Callable such that it will be run in
* our disk io thread and will automatically block null values since RxJava2 does not like null.
*
+ * @deprecated Use {@link #createFlowable(RoomDatabase, boolean, String[], Callable)}
+ *
+ * @hide
+ */
+ @Deprecated
+ @RestrictTo(RestrictTo.Scope.LIBRARY_GROUP_PREFIX)
+ public static <T> Flowable<T> createFlowable(final RoomDatabase database,
+ final String[] tableNames, final Callable<T> callable) {
+ return createFlowable(database, false, tableNames, callable);
+ }
+
+ /**
+ * Helper method used by generated code to bind a Callable such that it will be run in
+ * our disk io thread and will automatically block null values since RxJava2 does not like null.
+ *
* @hide
*/
@RestrictTo(RestrictTo.Scope.LIBRARY_GROUP_PREFIX)
public static <T> Flowable<T> createFlowable(final RoomDatabase database,
- final String[] tableNames, final Callable<T> callable) {
- Scheduler scheduler = Schedulers.from(database.getQueryExecutor());
+ final boolean inTransaction, final String[] tableNames, final Callable<T> callable) {
+ Scheduler scheduler = Schedulers.from(getExecutor(database, inTransaction));
final Maybe<T> maybe = Maybe.fromCallable(callable);
return createFlowable(database, tableNames)
.subscribeOn(scheduler)
@@ -161,12 +177,27 @@
* Helper method used by generated code to bind a Callable such that it will be run in
* our disk io thread and will automatically block null values since RxJava2 does not like null.
*
+ * @deprecated Use {@link #createObservable(RoomDatabase, boolean, String[], Callable)}
+ *
+ * @hide
+ */
+ @Deprecated
+ @RestrictTo(RestrictTo.Scope.LIBRARY_GROUP_PREFIX)
+ public static <T> Observable<T> createObservable(final RoomDatabase database,
+ final String[] tableNames, final Callable<T> callable) {
+ return createObservable(database, false, tableNames, callable);
+ }
+
+ /**
+ * Helper method used by generated code to bind a Callable such that it will be run in
+ * our disk io thread and will automatically block null values since RxJava2 does not like null.
+ *
* @hide
*/
@RestrictTo(RestrictTo.Scope.LIBRARY_GROUP_PREFIX)
public static <T> Observable<T> createObservable(final RoomDatabase database,
- final String[] tableNames, final Callable<T> callable) {
- Scheduler scheduler = Schedulers.from(database.getQueryExecutor());
+ final boolean inTransaction, final String[] tableNames, final Callable<T> callable) {
+ Scheduler scheduler = Schedulers.from(getExecutor(database, inTransaction));
final Maybe<T> maybe = Maybe.fromCallable(callable);
return createObservable(database, tableNames)
.subscribeOn(scheduler)
@@ -180,6 +211,14 @@
});
}
+ private static Executor getExecutor(RoomDatabase database, boolean inTransaction) {
+ if (inTransaction) {
+ return database.getTransactionExecutor();
+ } else {
+ return database.getQueryExecutor();
+ }
+ }
+
/** @deprecated This type should not be instantiated as it contains only static methods. */
@Deprecated
@SuppressWarnings("PrivateConstructorForUtilityClass")
diff --git a/room/rxjava2/src/test/java/androidx/room/RxRoomTest.java b/room/rxjava2/src/test/java/androidx/room/RxRoomTest.java
index cc96b50..dd5bebc 100644
--- a/room/rxjava2/src/test/java/androidx/room/RxRoomTest.java
+++ b/room/rxjava2/src/test/java/androidx/room/RxRoomTest.java
@@ -167,7 +167,7 @@
final AtomicReference<String> value = new AtomicReference<>(null);
String[] tables = {"a", "b"};
Set<String> tableSet = new HashSet<>(Arrays.asList(tables));
- final Flowable<String> flowable = RxRoom.createFlowable(mDatabase, tables,
+ final Flowable<String> flowable = RxRoom.createFlowable(mDatabase, false, tables,
new Callable<String>() {
@Override
public String call() throws Exception {
@@ -201,7 +201,7 @@
final AtomicReference<String> value = new AtomicReference<>(null);
String[] tables = {"a", "b"};
Set<String> tableSet = new HashSet<>(Arrays.asList(tables));
- final Observable<String> flowable = RxRoom.createObservable(mDatabase, tables,
+ final Observable<String> flowable = RxRoom.createObservable(mDatabase, false, tables,
new Callable<String>() {
@Override
public String call() throws Exception {
@@ -232,7 +232,7 @@
@Test
public void exception_Flowable() throws Exception {
- final Flowable<String> flowable = RxRoom.createFlowable(mDatabase, new String[]{"a"},
+ final Flowable<String> flowable = RxRoom.createFlowable(mDatabase, false, new String[]{"a"},
new Callable<String>() {
@Override
public String call() throws Exception {
@@ -248,7 +248,8 @@
@Test
public void exception_Observable() throws Exception {
- final Observable<String> flowable = RxRoom.createObservable(mDatabase, new String[]{"a"},
+ final Observable<String> flowable = RxRoom.createObservable(mDatabase, false,
+ new String[]{"a"},
new Callable<String>() {
@Override
public String call() throws Exception {
diff --git a/room/testing/src/main/java/androidx/room/testing/MigrationTestHelper.java b/room/testing/src/main/java/androidx/room/testing/MigrationTestHelper.java
index c0e1c4b..06b5a6b 100644
--- a/room/testing/src/main/java/androidx/room/testing/MigrationTestHelper.java
+++ b/room/testing/src/main/java/androidx/room/testing/MigrationTestHelper.java
@@ -158,6 +158,7 @@
true,
RoomDatabase.JournalMode.TRUNCATE,
ArchTaskExecutor.getIOThreadExecutor(),
+ ArchTaskExecutor.getIOThreadExecutor(),
false,
true,
false,
@@ -215,6 +216,7 @@
true,
RoomDatabase.JournalMode.TRUNCATE,
ArchTaskExecutor.getIOThreadExecutor(),
+ ArchTaskExecutor.getIOThreadExecutor(),
false,
true,
false,
