+package io.realm.kotlin.test.mongodb.shared++import io.realm.kotlin.Realm+import io.realm.kotlin.entities.sync.flx.FlexChildObject+import io.realm.kotlin.entities.sync.flx.FlexEmbeddedObject+import io.realm.kotlin.entities.sync.flx.FlexParentObject+import io.realm.kotlin.ext.query+import io.realm.kotlin.internal.platform.runBlocking+import io.realm.kotlin.mongodb.ext.subscribe+import io.realm.kotlin.mongodb.subscriptions+import io.realm.kotlin.mongodb.sync.Subscription+import io.realm.kotlin.mongodb.sync.SubscriptionSetState+import io.realm.kotlin.mongodb.sync.SyncConfiguration+import io.realm.kotlin.mongodb.sync.WaitForSync+import io.realm.kotlin.mongodb.syncSession+import io.realm.kotlin.query.RealmResults+import io.realm.kotlin.test.mongodb.TEST_APP_FLEX+import io.realm.kotlin.test.mongodb.TestApp+import io.realm.kotlin.test.mongodb.createUserAndLogIn+import io.realm.kotlin.test.util.TestHelper+import io.realm.kotlin.test.util.use+import kotlinx.coroutines.TimeoutCancellationException+import kotlin.random.Random+import kotlin.test.AfterTest+import kotlin.test.BeforeTest+import kotlin.test.Test+import kotlin.test.assertEquals+import kotlin.test.assertFailsWith+import kotlin.test.assertNotEquals+import kotlin.test.assertNull+import kotlin.time.Duration.Companion.nanoseconds+import kotlin.time.Duration.Companion.seconds++/**+ * Class for testing the various extension methods we have for bridging the gap between Subscriptions+ * and RealmQuery/RealmResults.+ */+class SubscriptionExtensionsTests {++    private lateinit var app: TestApp+    private lateinit var realm: Realm++    @BeforeTest+    fun setup() {+        app = TestApp(appName = TEST_APP_FLEX)+        val (email, password) = TestHelper.randomEmail() to "password1234"+        val user = runBlocking {+            app.createUserAndLogIn(email, password)+        }+        val config = SyncConfiguration.Builder(+            user,+            schema = setOf(FlexParentObject::class, FlexChildObject::class, FlexEmbeddedObject::class)+        )+            .build()+        realm = Realm.open(config)+    }++    @AfterTest+    fun tearDown() {+        if (this::realm.isInitialized && !realm.isClosed()) {+            realm.close()+        }+        if (this::app.isInitialized) {+            app.close()+        }+    }++    @Test+    fun realmQuery_subscribe_anonymous() = runBlocking {+        val subs = realm.subscriptions+        assertEquals(0, subs.size)+        val results: RealmResults<FlexParentObject> = realm.query<FlexParentObject>().subscribe()+        assertEquals(SubscriptionSetState.COMPLETE, subs.state)+        assertEquals(1, subs.size)+        val sub: Subscription = subs.first()+        assertNull(sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals(FlexParentObject::class.simpleName, sub.objectType)+    }++    // Check that subscribing twice to a query will result in the same subscription+    @Test+    fun realmQuery_subscribe_anonymousTwice() = runBlocking {+        val subs = realm.subscriptions+        assertEquals(0, subs.size)+        realm.query<FlexParentObject>().subscribe()+        realm.query<FlexParentObject>().subscribe()+        assertEquals(SubscriptionSetState.COMPLETE, subs.state)+        assertEquals(1, subs.size)+        val sub: Subscription = subs.first()+        assertNull(sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals(FlexParentObject::class.simpleName, sub.objectType)+    }++    // Check that anonymous RealmQuery and RealmResults .subscribe calls result in the same sub.+    @Test+    fun anonymousSubscriptionsOverlap() = runBlocking {+        val subs = realm.subscriptions+        assertEquals(0, subs.size)+        realm.query<FlexParentObject>().subscribe()+        realm.query<FlexParentObject>().find().subscribe()+        assertEquals(SubscriptionSetState.COMPLETE, subs.state)+        assertEquals(1, subs.size)+        val sub: Subscription = subs.first()+        assertNull(sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals(FlexParentObject::class.simpleName, sub.objectType)+    }++    // Verify that the realm query doesn't run against a frozen version previous to the Realm

+@file:Suppress("invisible_reference", "invisible_member")+package io.realm.kotlin.mongodb.ext++import io.realm.kotlin.mongodb.annotations.ExperimentalFlexibleSyncApi+import io.realm.kotlin.mongodb.sync.Subscription+import io.realm.kotlin.mongodb.sync.SubscriptionSet+import io.realm.kotlin.mongodb.sync.WaitForSync+import io.realm.kotlin.query.RealmQuery+import io.realm.kotlin.query.RealmResults+import io.realm.kotlin.types.RealmObject+import kotlin.time.Duration++/**+ * Automatically create an anonymous [Subscription] from a local query result in the background and+ * return the result of re-running the same query against the Realm file.+ *+ * This is a more streamlined alternative to doing something like this:+ *+ * ```+ * fun suspend getData(realm: Realm): RealmResults<Person> {+ *     val results = realm.query<Person>().find()+ *     realm.subscriptions.update { bgRealm ->+ *         add("myquery", results.query(""))+ *     }+ *     realm.subscriptions.waitForSynchronization()+ *     return realm.query<Person>().find()+ * }+ * ```+ *+ * It is possible to define whether or not to wait for the server to send all data before+ * running the local query. This is relevant as there might be delay from creating a subscription+ * to the data being available on the device due to either latency or because a large dataset needs+ * be downloaded.+ *+ * The default behaviour is that the first time `subscribe` is called, the query result will not+ * be returned until data has been downloaded from the server. On subsequent calls to `subscribe`+ * for the same query, the query will run immediately on the local database while any updates+ * are downloaded in the background.+ *+ * @param name name of the subscription. This can be used to identify it later in the [SubscriptionSet].+ * @param mode type of mode used to resolve the subscription. See [WaitForSync] for more details.+ * @param timeout How long to wait for the server to return the objects defined by the subscription.+ * This is only relevant for [WaitForSync.ALWAYS] and [WaitForSync.FIRST_TIME].+ * @return The result of running the query against the local Realm file. The results returned will+ * depend on which [mode] was used.+ * @throws kotlinx.coroutines.TimeoutCancellationException if the specified timeout was hit before+ * a query result could be returned.+ * @Throws IllegalStateException if this method is called on a Realm that isn't using Flexible Sync.+ */@ExperimentalFlexibleSyncApi+public suspend fun <T : RealmObject> RealmResults<T>.subscribe(+    mode: WaitForSync = WaitForSync.FIRST_TIME,+    timeout: Duration = Duration.INFINITE+): RealmResults<T> {+    val query: RealmQuery<T> = this.query("")+    return query.subscribe(mode, timeout)+}++/**+ * Automatically create a named [Subscription] from a local query result in the background and+ * return the result of re-running the same query against the Realm file.+ *+ * This is a more streamlined alternative to doing something like this:+ *+ * ```+ * fun suspend getData(realm: Realm): RealmResults<Person> {+ *     val results = realm.query<Person>().find()+ *     realm.subscriptions.update { bgRealm ->+ *         add("myquery", results.query(""))+ *     }+ *     realm.subscriptions.waitForSynchronization()+ *     return realm.query<Person>().find()+ * }+ * ```+ *+ * It is possible to define whether or not to wait for the server to send all data before+ * running the local query. This is relevant as there might be delay from creating a subscription+ * to the data being available on the device due to either latency or because a large dataset needs+ * be downloaded.+ *+ * The default behaviour is that the first time `subscribe` is called, the query result will not+ * be returned until data has been downloaded from the server. On subsequent calls to `subscribe`+ * for the same query, the query will run immediately on the local database while any updates+ * are downloaded in the background.+ *+ * @param name name of the subscription. This can be used to identify it later in the [SubscriptionSet].+ * @param mode type of mode used to resolve the subscription. See [WaitForSync] for more details.+ * @param timeout How long to wait for the server to return the objects defined by the subscription.+ * This is only relevant for [WaitForSync.ALWAYS] and [WaitForSync.FIRST_TIME].+ * @return The result of running the query against the local Realm file. The results returned will+ * depend on which [mode] was used.+ * @throws kotlinx.coroutines.TimeoutCancellationException if the specified timeout was hit before+ * a query result could be returned.+ * @Throws IllegalStateException if this method is called on a Realm that isn't using Flexible Sync.+ */+@ExperimentalFlexibleSyncApi+public suspend fun <T : RealmObject> RealmResults<T>.subscribe(+    name: String,

+@file:Suppress("invisible_reference", "invisible_member")+package io.realm.kotlin.mongodb.ext++import io.realm.kotlin.mongodb.annotations.ExperimentalFlexibleSyncApi+import io.realm.kotlin.mongodb.sync.Subscription+import io.realm.kotlin.mongodb.sync.SubscriptionSet+import io.realm.kotlin.mongodb.sync.WaitForSync+import io.realm.kotlin.query.RealmQuery+import io.realm.kotlin.query.RealmResults+import io.realm.kotlin.types.RealmObject+import kotlin.time.Duration++/**+ * Automatically create an anonymous [Subscription] from a local query result in the background and+ * return the result of re-running the same query against the Realm file.+ *+ * This is a more streamlined alternative to doing something like this:+ *+ * ```+ * fun suspend getData(realm: Realm): RealmResults<Person> {+ *     val results = realm.query<Person>().find()+ *     realm.subscriptions.update { bgRealm ->+ *         add("myquery", results.query(""))+ *     }+ *     realm.subscriptions.waitForSynchronization()+ *     return realm.query<Person>().find()+ * }+ * ```+ *+ * It is possible to define whether or not to wait for the server to send all data before+ * running the local query. This is relevant as there might be delay from creating a subscription+ * to the data being available on the device due to either latency or because a large dataset needs+ * be downloaded.+ *+ * The default behaviour is that the first time `subscribe` is called, the query result will not+ * be returned until data has been downloaded from the server. On subsequent calls to `subscribe`+ * for the same query, the query will run immediately on the local database while any updates+ * are downloaded in the background.+ *+ * @param name name of the subscription. This can be used to identify it later in the [SubscriptionSet].+ * @param mode type of mode used to resolve the subscription. See [WaitForSync] for more details.

+package io.realm.kotlin.test.mongodb.shared++import io.realm.kotlin.Realm+import io.realm.kotlin.entities.sync.flx.FlexChildObject+import io.realm.kotlin.entities.sync.flx.FlexEmbeddedObject+import io.realm.kotlin.entities.sync.flx.FlexParentObject+import io.realm.kotlin.ext.query+import io.realm.kotlin.internal.platform.runBlocking+import io.realm.kotlin.mongodb.ext.subscribe+import io.realm.kotlin.mongodb.subscriptions+import io.realm.kotlin.mongodb.sync.Subscription+import io.realm.kotlin.mongodb.sync.SubscriptionSetState+import io.realm.kotlin.mongodb.sync.SyncConfiguration+import io.realm.kotlin.mongodb.sync.WaitForSync+import io.realm.kotlin.mongodb.syncSession+import io.realm.kotlin.query.RealmResults+import io.realm.kotlin.test.mongodb.TEST_APP_FLEX+import io.realm.kotlin.test.mongodb.TestApp+import io.realm.kotlin.test.mongodb.createUserAndLogIn+import io.realm.kotlin.test.util.TestHelper+import io.realm.kotlin.test.util.use+import kotlinx.coroutines.TimeoutCancellationException+import kotlin.random.Random+import kotlin.test.AfterTest+import kotlin.test.BeforeTest+import kotlin.test.Test+import kotlin.test.assertEquals+import kotlin.test.assertFailsWith+import kotlin.test.assertNotEquals+import kotlin.test.assertNull+import kotlin.time.Duration.Companion.nanoseconds+import kotlin.time.Duration.Companion.seconds++/**+ * Class for testing the various extension methods we have for bridging the gap between Subscriptions+ * and RealmQuery/RealmResults.+ */+class SubscriptionExtensionsTests {++    private lateinit var app: TestApp+    private lateinit var realm: Realm++    @BeforeTest+    fun setup() {+        app = TestApp(appName = TEST_APP_FLEX)+        val (email, password) = TestHelper.randomEmail() to "password1234"+        val user = runBlocking {+            app.createUserAndLogIn(email, password)+        }+        val config = SyncConfiguration.Builder(+            user,+            schema = setOf(FlexParentObject::class, FlexChildObject::class, FlexEmbeddedObject::class)+        )+            .build()+        realm = Realm.open(config)+    }++    @AfterTest+    fun tearDown() {+        if (this::realm.isInitialized && !realm.isClosed()) {+            realm.close()+        }+        if (this::app.isInitialized) {+            app.close()+        }+    }++    @Test+    fun realmQuery_subscribe_anonymous() = runBlocking {+        val subs = realm.subscriptions+        assertEquals(0, subs.size)+        val results: RealmResults<FlexParentObject> = realm.query<FlexParentObject>().subscribe()+        assertEquals(SubscriptionSetState.COMPLETE, subs.state)+        assertEquals(1, subs.size)+        val sub: Subscription = subs.first()+        assertNull(sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals(FlexParentObject::class.simpleName, sub.objectType)+    }++    // Check that subscribing twice to a query will result in the same subscription+    @Test+    fun realmQuery_subscribe_anonymousTwice() = runBlocking {+        val subs = realm.subscriptions+        assertEquals(0, subs.size)+        realm.query<FlexParentObject>().subscribe()+        realm.query<FlexParentObject>().subscribe()+        assertEquals(SubscriptionSetState.COMPLETE, subs.state)+        assertEquals(1, subs.size)+        val sub: Subscription = subs.first()+        assertNull(sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals(FlexParentObject::class.simpleName, sub.objectType)+    }++    // Check that anonymous RealmQuery and RealmResults .subscribe calls result in the same sub.

+package io.realm.kotlin.test.mongodb.shared++import io.realm.kotlin.Realm+import io.realm.kotlin.entities.sync.flx.FlexChildObject+import io.realm.kotlin.entities.sync.flx.FlexEmbeddedObject+import io.realm.kotlin.entities.sync.flx.FlexParentObject+import io.realm.kotlin.ext.query+import io.realm.kotlin.internal.platform.runBlocking+import io.realm.kotlin.mongodb.ext.subscribe+import io.realm.kotlin.mongodb.subscriptions+import io.realm.kotlin.mongodb.sync.Subscription+import io.realm.kotlin.mongodb.sync.SubscriptionSetState+import io.realm.kotlin.mongodb.sync.SyncConfiguration+import io.realm.kotlin.mongodb.sync.WaitForSync+import io.realm.kotlin.mongodb.syncSession+import io.realm.kotlin.query.RealmResults+import io.realm.kotlin.test.mongodb.TEST_APP_FLEX+import io.realm.kotlin.test.mongodb.TestApp+import io.realm.kotlin.test.mongodb.createUserAndLogIn+import io.realm.kotlin.test.util.TestHelper+import io.realm.kotlin.test.util.use+import kotlinx.coroutines.TimeoutCancellationException+import kotlin.random.Random+import kotlin.test.AfterTest+import kotlin.test.BeforeTest+import kotlin.test.Test+import kotlin.test.assertEquals+import kotlin.test.assertFailsWith+import kotlin.test.assertNotEquals+import kotlin.test.assertNull+import kotlin.time.Duration.Companion.nanoseconds+import kotlin.time.Duration.Companion.seconds++/**+ * Class for testing the various extension methods we have for bridging the gap between Subscriptions+ * and RealmQuery/RealmResults.+ */+class SubscriptionExtensionsTests {++    private lateinit var app: TestApp+    private lateinit var realm: Realm++    @BeforeTest+    fun setup() {+        app = TestApp(appName = TEST_APP_FLEX)+        val (email, password) = TestHelper.randomEmail() to "password1234"+        val user = runBlocking {+            app.createUserAndLogIn(email, password)+        }+        val config = SyncConfiguration.Builder(+            user,+            schema = setOf(FlexParentObject::class, FlexChildObject::class, FlexEmbeddedObject::class)+        )+            .build()+        realm = Realm.open(config)+    }++    @AfterTest+    fun tearDown() {+        if (this::realm.isInitialized && !realm.isClosed()) {+            realm.close()+        }+        if (this::app.isInitialized) {+            app.close()+        }+    }++    @Test+    fun realmQuery_subscribe_anonymous() = runBlocking {+        val subs = realm.subscriptions+        assertEquals(0, subs.size)+        val results: RealmResults<FlexParentObject> = realm.query<FlexParentObject>().subscribe()+        assertEquals(SubscriptionSetState.COMPLETE, subs.state)+        assertEquals(1, subs.size)+        val sub: Subscription = subs.first()+        assertNull(sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals(FlexParentObject::class.simpleName, sub.objectType)+    }++    // Check that subscribing twice to a query will result in the same subscription+    @Test+    fun realmQuery_subscribe_anonymousTwice() = runBlocking {+        val subs = realm.subscriptions+        assertEquals(0, subs.size)+        realm.query<FlexParentObject>().subscribe()+        realm.query<FlexParentObject>().subscribe()+        assertEquals(SubscriptionSetState.COMPLETE, subs.state)+        assertEquals(1, subs.size)+        val sub: Subscription = subs.first()+        assertNull(sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals(FlexParentObject::class.simpleName, sub.objectType)+    }++    // Check that anonymous RealmQuery and RealmResults .subscribe calls result in the same sub.+    @Test+    fun anonymousSubscriptionsOverlap() = runBlocking {+        val subs = realm.subscriptions+        assertEquals(0, subs.size)+        realm.query<FlexParentObject>().subscribe()+        realm.query<FlexParentObject>().find().subscribe()+        assertEquals(SubscriptionSetState.COMPLETE, subs.state)+        assertEquals(1, subs.size)+        val sub: Subscription = subs.first()+        assertNull(sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals(FlexParentObject::class.simpleName, sub.objectType)+    }++    // Verify that the realm query doesn't run against a frozen version previous to the Realm+    // being updated from `waitForSynchronization`.+    @Test+    fun realmQuery_subscribe_queryResultIsLatestVersion() = runBlocking {+        // Write data to a server Realm+        val section = Random.nextInt()+        val (email, password) = TestHelper.randomEmail() to "password1234"+        val user1 = app.createUserAndLogIn(email, password)+        val config = SyncConfiguration.Builder(+            user1,+            schema = setOf(FlexParentObject::class, FlexChildObject::class, FlexEmbeddedObject::class)+        ).initialSubscriptions { realm: Realm ->+            realm.query<FlexParentObject>("section = $0", section).subscribe()+        }.build()++        Realm.open(config).use { realmFromAnotherDevice ->+            realmFromAnotherDevice.writeBlocking {+                copyToRealm(FlexParentObject(section))+            }+            realmFromAnotherDevice.syncSession.uploadAllLocalChanges(30.seconds)+        }++        // Data still hasn't reached this device+        assertEquals(0, realm.query<FlexParentObject>().count().find())+        // Check that subscribing to a query, will run the query on the data downloaded from+        // the server and not just local data, due to WaitForSync.FIRST_TIME being the default.+        val result = realm.query<FlexParentObject>("section = $0", section).subscribe()+        assertEquals(1, result.size)+        assertEquals(section, result.first().section)+    }++    @Test+    fun realmQuery_subscribe_waitFirstTime() = runBlocking<Unit> {+        // Unnamed+        realm.query<FlexParentObject>().subscribe() // Default value is WaitForSync.FIRST_TIME+        var updatedSubs = realm.subscriptions+        assertEquals(1, updatedSubs.size)+        assertEquals(SubscriptionSetState.COMPLETE, updatedSubs.state)+        var sub: Subscription = updatedSubs.first()+        assertNull(sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals("FlexParentObject", sub.objectType)++        // Named+        realm.query<FlexParentObject>().subscribe("my-name") // Default value is WaitForSync.FIRST_TIME+        updatedSubs = realm.subscriptions+        assertEquals(2, updatedSubs.size)+        assertEquals(SubscriptionSetState.COMPLETE, updatedSubs.state)+        sub = updatedSubs.last()+        assertEquals("my-name", sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals("FlexParentObject", sub.objectType)+    }++    @Test+    fun realmQuery_subscribe_waitNever() = runBlocking {+        // Un-named+        realm.query<FlexParentObject>().subscribe(mode = WaitForSync.NEVER)+        var updatedSubs = realm.subscriptions+        assertEquals(1, updatedSubs.size)+        // Updating the subscription will happen in the background, but+        // hopefully hasn't reached COMPLETE yet.+        assertNotEquals(SubscriptionSetState.COMPLETE, updatedSubs.state)++        // Named+        realm.query<FlexParentObject>().subscribe(name = "my-name", mode = WaitForSync.NEVER)+        updatedSubs = realm.subscriptions+        assertEquals(2, updatedSubs.size)+        // Updating the subscription will happen in the background, but+        // hopefully hasn't reached COMPLETE yet.+        assertNotEquals(SubscriptionSetState.COMPLETE, updatedSubs.state)+    }++    @Test+    fun realmQuery_subscribe_waitAlways() = runBlocking {

+@file:Suppress("invisible_reference", "invisible_member")+package io.realm.kotlin.mongodb.ext++import io.realm.kotlin.Realm+import io.realm.kotlin.internal.RealmImpl+import io.realm.kotlin.internal.getRealm+import io.realm.kotlin.mongodb.annotations.ExperimentalFlexibleSyncApi+import io.realm.kotlin.mongodb.internal.AppImpl+import io.realm.kotlin.mongodb.subscriptions+import io.realm.kotlin.mongodb.sync.Subscription+import io.realm.kotlin.mongodb.sync.SubscriptionSet+import io.realm.kotlin.mongodb.sync.SyncConfiguration+import io.realm.kotlin.mongodb.sync.WaitForSync+import io.realm.kotlin.mongodb.syncSession+import io.realm.kotlin.query.RealmQuery+import io.realm.kotlin.query.RealmResults+import io.realm.kotlin.types.RealmObject+import kotlinx.coroutines.CoroutineDispatcher+import kotlinx.coroutines.withContext+import kotlinx.coroutines.withTimeout+import kotlin.time.Duration++/**+ * Automatically create an anonymous [Subscription] from a query in the background and return the+ * result of running the same query against the local Realm file.+ *+ * This is a more streamlined alternative to doing something like this:+ *+ * ```+ * fun suspend getData(realm: Realm): RealmResults<Person> {+ *     realm.subscriptions.update { bgRealm ->+ *         add(bgRealm.query<Person>())+ *     }+ *     realm.subscriptions.waitForSynchronization()+ *     return realm.query<Person>().find()+ * }+ * ```+ *+ * It is possible to define whether or not to wait for the server to send all data before+ * running the local query. This is relevant as there might be delay from creating a subscription+ * to the data being available on the device due to either latency or because a large dataset needs+ * be downloaded.+ *+ * The default behaviour is that the first time `subscribe` is called, the query result will not+ * be returned until data has been downloaded from the server. On subsequent calls to `subscribe`+ * for the same query, the query will run immediately on the local database while any updates+ * are downloaded in the background.+ *+ * @param mode type of mode used to resolve the subscription. See [WaitForSync] for more details.+ * @param timeout How long to wait for the server to return the objects defined by the subscription.+ * This is only relevant for [WaitForSync.ALWAYS] and [WaitForSync.FIRST_TIME].+ * @return The result of running the query against the local Realm file. The results returned will+ * depend on which [mode] was used.+ * @throws kotlinx.coroutines.TimeoutCancellationException if the specified timeout was hit before+ * a query result could be returned.+ * @Throws IllegalStateException if this method is called on a Realm that isn't using Flexible Sync.+ */+@ExperimentalFlexibleSyncApi+public suspend fun <T : RealmObject> RealmQuery<T>.subscribe(+    mode: WaitForSync = WaitForSync.FIRST_TIME,+    timeout: Duration = Duration.INFINITE+): RealmResults<T> {+    return createSubscriptionFromQuery(this, null, false, mode, timeout)+}++/**+ * Automatically create a named [Subscription] from a query in the background and return the+ * result of running the same query against the local Realm file.+ *+ * This is a more streamlined alternative to doing something like this:+ *+ * ```+ * fun suspend getData(realm: Realm): RealmResults<Person> {+ *     realm.subscriptions.update { bgRealm ->+ *         add("myquery", bgRealm.query<Person>())+ *     }+ *     realm.subscriptions.waitForSynchronization()+ *     return realm.query<Person>().find()+ * }+ * ```+ *+ * It is possible to define whether or not to wait for the server to send all data before+ * running the local query. This is relevant as there might be delay from creating a subscription+ * to the data being available on the device due to either latency or because a large dataset needs+ * be downloaded.+ *+ * The default behaviour is that the first time `subscribe` is called, the query result will not+ * be returned until data has been downloaded from the server. On subsequent calls to `subscribe`+ * for the same query, the query will run immediately on the local database while any updates+ * are downloaded in the background.+ *+ * @param name name of the subscription. This can be used to identify it later in the [SubscriptionSet].+ * @param mode type of mode used to resolve the subscription. See [WaitForSync] for more details.+ * @param timeout How long to wait for the server to return the objects defined by the subscription.+ * This is only relevant for [WaitForSync.ALWAYS] and [WaitForSync.FIRST_TIME].+ * @return The result of running the query against the local Realm file. The results returned will+ * depend on which [mode] was used.+ * @throws kotlinx.coroutines.TimeoutCancellationException if the specified timeout was hit before+ * a query result could be returned.+ * @Throws IllegalStateException if this method is called on a Realm that isn't using Flexible Sync.+ */+@ExperimentalFlexibleSyncApi+public suspend fun <T : RealmObject> RealmQuery<T>.subscribe(+    name: String,+    updateExisting: Boolean = false,+    mode: WaitForSync = WaitForSync.FIRST_TIME,+    timeout: Duration = Duration.INFINITE+): RealmResults<T> {+    return createSubscriptionFromQuery(this, name, updateExisting, mode, timeout)+}++private suspend fun <T : RealmObject> createSubscriptionFromQuery(+    query: RealmQuery<T>,+    name: String?,+    updateExisting: Boolean = false,+    mode: WaitForSync,+    timeout: Duration+): RealmResults<T> {++    if (query !is io.realm.kotlin.internal.query.ObjectQuery<T>) {+        throw IllegalStateException("Only queries on objects are supported. This was: ${query::class}")+    }+    if (query.realmReference.owner !is RealmImpl) {+        throw IllegalStateException("Calling `subscribe()` inside a write transaction is not allowed.")+    }+    val realm: Realm = query.getRealm()+    val subscriptions = realm.subscriptions+    val appDispatcher: CoroutineDispatcher = ((realm.configuration as SyncConfiguration).user.app as AppImpl).appNetworkDispatcher.dispatcher

+package io.realm.kotlin.test.mongodb.shared++import io.realm.kotlin.Realm+import io.realm.kotlin.entities.sync.flx.FlexChildObject+import io.realm.kotlin.entities.sync.flx.FlexEmbeddedObject+import io.realm.kotlin.entities.sync.flx.FlexParentObject+import io.realm.kotlin.ext.query+import io.realm.kotlin.internal.platform.runBlocking+import io.realm.kotlin.mongodb.ext.subscribe+import io.realm.kotlin.mongodb.subscriptions+import io.realm.kotlin.mongodb.sync.Subscription+import io.realm.kotlin.mongodb.sync.SubscriptionSetState+import io.realm.kotlin.mongodb.sync.SyncConfiguration+import io.realm.kotlin.mongodb.sync.WaitForSync+import io.realm.kotlin.mongodb.syncSession+import io.realm.kotlin.query.RealmResults+import io.realm.kotlin.test.mongodb.TEST_APP_FLEX+import io.realm.kotlin.test.mongodb.TestApp+import io.realm.kotlin.test.mongodb.createUserAndLogIn+import io.realm.kotlin.test.util.TestHelper+import io.realm.kotlin.test.util.use+import kotlinx.coroutines.TimeoutCancellationException+import kotlin.random.Random+import kotlin.test.AfterTest+import kotlin.test.BeforeTest+import kotlin.test.Test+import kotlin.test.assertEquals+import kotlin.test.assertFailsWith+import kotlin.test.assertNotEquals+import kotlin.test.assertNull+import kotlin.time.Duration.Companion.nanoseconds+import kotlin.time.Duration.Companion.seconds++/**+ * Class for testing the various extension methods we have for bridging the gap between Subscriptions+ * and RealmQuery/RealmResults.+ */+class SubscriptionExtensionsTests {++    private lateinit var app: TestApp+    private lateinit var realm: Realm++    @BeforeTest+    fun setup() {+        app = TestApp(appName = TEST_APP_FLEX)+        val (email, password) = TestHelper.randomEmail() to "password1234"+        val user = runBlocking {+            app.createUserAndLogIn(email, password)+        }+        val config = SyncConfiguration.Builder(+            user,+            schema = setOf(FlexParentObject::class, FlexChildObject::class, FlexEmbeddedObject::class)+        )+            .build()+        realm = Realm.open(config)+    }++    @AfterTest+    fun tearDown() {+        if (this::realm.isInitialized && !realm.isClosed()) {+            realm.close()+        }+        if (this::app.isInitialized) {+            app.close()+        }+    }++    @Test+    fun realmQuery_subscribe_anonymous() = runBlocking {+        val subs = realm.subscriptions+        assertEquals(0, subs.size)+        val results: RealmResults<FlexParentObject> = realm.query<FlexParentObject>().subscribe()+        assertEquals(SubscriptionSetState.COMPLETE, subs.state)+        assertEquals(1, subs.size)+        val sub: Subscription = subs.first()+        assertNull(sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals(FlexParentObject::class.simpleName, sub.objectType)+    }++    // Check that subscribing twice to a query will result in the same subscription+    @Test+    fun realmQuery_subscribe_anonymousTwice() = runBlocking {+        val subs = realm.subscriptions+        assertEquals(0, subs.size)+        realm.query<FlexParentObject>().subscribe()+        realm.query<FlexParentObject>().subscribe()+        assertEquals(SubscriptionSetState.COMPLETE, subs.state)+        assertEquals(1, subs.size)+        val sub: Subscription = subs.first()+        assertNull(sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals(FlexParentObject::class.simpleName, sub.objectType)+    }++    // Check that anonymous RealmQuery and RealmResults .subscribe calls result in the same sub.+    @Test+    fun anonymousSubscriptionsOverlap() = runBlocking {+        val subs = realm.subscriptions+        assertEquals(0, subs.size)+        realm.query<FlexParentObject>().subscribe()+        realm.query<FlexParentObject>().find().subscribe()+        assertEquals(SubscriptionSetState.COMPLETE, subs.state)+        assertEquals(1, subs.size)+        val sub: Subscription = subs.first()+        assertNull(sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals(FlexParentObject::class.simpleName, sub.objectType)+    }++    // Verify that the realm query doesn't run against a frozen version previous to the Realm+    // being updated from `waitForSynchronization`.+    @Test+    fun realmQuery_subscribe_queryResultIsLatestVersion() = runBlocking {+        // Write data to a server Realm+        val section = Random.nextInt()+        val (email, password) = TestHelper.randomEmail() to "password1234"+        val user1 = app.createUserAndLogIn(email, password)+        val config = SyncConfiguration.Builder(+            user1,+            schema = setOf(FlexParentObject::class, FlexChildObject::class, FlexEmbeddedObject::class)+        ).initialSubscriptions { realm: Realm ->+            realm.query<FlexParentObject>("section = $0", section).subscribe()+        }.build()++        Realm.open(config).use { realmFromAnotherDevice ->+            realmFromAnotherDevice.writeBlocking {+                copyToRealm(FlexParentObject(section))+            }+            realmFromAnotherDevice.syncSession.uploadAllLocalChanges(30.seconds)+        }++        // Data still hasn't reached this device+        assertEquals(0, realm.query<FlexParentObject>().count().find())+        // Check that subscribing to a query, will run the query on the data downloaded from+        // the server and not just local data, due to WaitForSync.FIRST_TIME being the default.+        val result = realm.query<FlexParentObject>("section = $0", section).subscribe()+        assertEquals(1, result.size)+        assertEquals(section, result.first().section)+    }++    @Test+    fun realmQuery_subscribe_waitFirstTime() = runBlocking<Unit> {+        // Unnamed+        realm.query<FlexParentObject>().subscribe() // Default value is WaitForSync.FIRST_TIME+        var updatedSubs = realm.subscriptions+        assertEquals(1, updatedSubs.size)+        assertEquals(SubscriptionSetState.COMPLETE, updatedSubs.state)+        var sub: Subscription = updatedSubs.first()+        assertNull(sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals("FlexParentObject", sub.objectType)++        // Named+        realm.query<FlexParentObject>().subscribe("my-name") // Default value is WaitForSync.FIRST_TIME+        updatedSubs = realm.subscriptions+        assertEquals(2, updatedSubs.size)+        assertEquals(SubscriptionSetState.COMPLETE, updatedSubs.state)+        sub = updatedSubs.last()+        assertEquals("my-name", sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals("FlexParentObject", sub.objectType)+    }++    @Test+    fun realmQuery_subscribe_waitNever() = runBlocking {+        // Un-named+        realm.query<FlexParentObject>().subscribe(mode = WaitForSync.NEVER)+        var updatedSubs = realm.subscriptions+        assertEquals(1, updatedSubs.size)+        // Updating the subscription will happen in the background, but+        // hopefully hasn't reached COMPLETE yet.+        assertNotEquals(SubscriptionSetState.COMPLETE, updatedSubs.state)++        // Named+        realm.query<FlexParentObject>().subscribe(name = "my-name", mode = WaitForSync.NEVER)+        updatedSubs = realm.subscriptions+        assertEquals(2, updatedSubs.size)+        // Updating the subscription will happen in the background, but+        // hopefully hasn't reached COMPLETE yet.+        assertNotEquals(SubscriptionSetState.COMPLETE, updatedSubs.state)+    }++    @Test+    fun realmQuery_subscribe_waitAlways() = runBlocking {+        val sectionId = Random.nextInt()+        val results1 = realm.query<FlexParentObject>("section = $0", sectionId).subscribe() // Default value is WaitForSync.FIRST_TIME+        assertEquals(0, results1.size)+        uploadServerData(sectionId, 5)+        // Since the subscription is already present, we cannot control if the data is downloaded+        // before creating the next subscription. Instead we pause the syncSession and verify+        // that WaitForSync.ALWAYS timeout during network failures and resuming the session should+        // then work+        realm.syncSession.pause()+        assertFailsWith<TimeoutCancellationException> {+            realm.query<FlexParentObject>("section = $0", sectionId).subscribe(timeout = 3.seconds, mode = WaitForSync.ALWAYS)+        }+        realm.syncSession.resume()+        val results2 = realm.query<FlexParentObject>("section = $0", sectionId).subscribe(mode = WaitForSync.ALWAYS)+        assertEquals(5, results2.size)+    }++    @Test+    fun realmQuery_subscribe_timeOut_fails() = runBlocking<Unit> {+        assertFailsWith<TimeoutCancellationException> {+            realm.query<FlexParentObject>().subscribe(timeout = 1.nanoseconds)

+package io.realm.kotlin.test.mongodb.shared++import io.realm.kotlin.Realm+import io.realm.kotlin.entities.sync.flx.FlexChildObject+import io.realm.kotlin.entities.sync.flx.FlexEmbeddedObject+import io.realm.kotlin.entities.sync.flx.FlexParentObject+import io.realm.kotlin.ext.query+import io.realm.kotlin.internal.platform.runBlocking+import io.realm.kotlin.mongodb.ext.subscribe+import io.realm.kotlin.mongodb.subscriptions+import io.realm.kotlin.mongodb.sync.Subscription+import io.realm.kotlin.mongodb.sync.SubscriptionSetState+import io.realm.kotlin.mongodb.sync.SyncConfiguration+import io.realm.kotlin.mongodb.sync.WaitForSync+import io.realm.kotlin.mongodb.syncSession+import io.realm.kotlin.query.RealmResults+import io.realm.kotlin.test.mongodb.TEST_APP_FLEX+import io.realm.kotlin.test.mongodb.TestApp+import io.realm.kotlin.test.mongodb.createUserAndLogIn+import io.realm.kotlin.test.util.TestHelper+import io.realm.kotlin.test.util.use+import kotlinx.coroutines.TimeoutCancellationException+import kotlin.random.Random+import kotlin.test.AfterTest+import kotlin.test.BeforeTest+import kotlin.test.Test+import kotlin.test.assertEquals+import kotlin.test.assertFailsWith+import kotlin.test.assertNotEquals+import kotlin.test.assertNull+import kotlin.time.Duration.Companion.nanoseconds+import kotlin.time.Duration.Companion.seconds++/**+ * Class for testing the various extension methods we have for bridging the gap between Subscriptions+ * and RealmQuery/RealmResults.+ */+class SubscriptionExtensionsTests {++    private lateinit var app: TestApp+    private lateinit var realm: Realm++    @BeforeTest+    fun setup() {+        app = TestApp(appName = TEST_APP_FLEX)+        val (email, password) = TestHelper.randomEmail() to "password1234"+        val user = runBlocking {+            app.createUserAndLogIn(email, password)+        }+        val config = SyncConfiguration.Builder(+            user,+            schema = setOf(FlexParentObject::class, FlexChildObject::class, FlexEmbeddedObject::class)+        )+            .build()+        realm = Realm.open(config)+    }++    @AfterTest+    fun tearDown() {+        if (this::realm.isInitialized && !realm.isClosed()) {+            realm.close()+        }+        if (this::app.isInitialized) {+            app.close()+        }+    }++    @Test+    fun realmQuery_subscribe_anonymous() = runBlocking {+        val subs = realm.subscriptions+        assertEquals(0, subs.size)+        val results: RealmResults<FlexParentObject> = realm.query<FlexParentObject>().subscribe()+        assertEquals(SubscriptionSetState.COMPLETE, subs.state)+        assertEquals(1, subs.size)+        val sub: Subscription = subs.first()+        assertNull(sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals(FlexParentObject::class.simpleName, sub.objectType)+    }++    // Check that subscribing twice to a query will result in the same subscription+    @Test+    fun realmQuery_subscribe_anonymousTwice() = runBlocking {+        val subs = realm.subscriptions+        assertEquals(0, subs.size)+        realm.query<FlexParentObject>().subscribe()+        realm.query<FlexParentObject>().subscribe()+        assertEquals(SubscriptionSetState.COMPLETE, subs.state)+        assertEquals(1, subs.size)+        val sub: Subscription = subs.first()+        assertNull(sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals(FlexParentObject::class.simpleName, sub.objectType)+    }++    // Check that anonymous RealmQuery and RealmResults .subscribe calls result in the same sub.+    @Test+    fun anonymousSubscriptionsOverlap() = runBlocking {+        val subs = realm.subscriptions+        assertEquals(0, subs.size)+        realm.query<FlexParentObject>().subscribe()+        realm.query<FlexParentObject>().find().subscribe()+        assertEquals(SubscriptionSetState.COMPLETE, subs.state)+        assertEquals(1, subs.size)+        val sub: Subscription = subs.first()+        assertNull(sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals(FlexParentObject::class.simpleName, sub.objectType)+    }++    // Verify that the realm query doesn't run against a frozen version previous to the Realm+    // being updated from `waitForSynchronization`.+    @Test+    fun realmQuery_subscribe_queryResultIsLatestVersion() = runBlocking {+        // Write data to a server Realm+        val section = Random.nextInt()+        val (email, password) = TestHelper.randomEmail() to "password1234"+        val user1 = app.createUserAndLogIn(email, password)+        val config = SyncConfiguration.Builder(+            user1,+            schema = setOf(FlexParentObject::class, FlexChildObject::class, FlexEmbeddedObject::class)+        ).initialSubscriptions { realm: Realm ->+            realm.query<FlexParentObject>("section = $0", section).subscribe()+        }.build()++        Realm.open(config).use { realmFromAnotherDevice ->+            realmFromAnotherDevice.writeBlocking {+                copyToRealm(FlexParentObject(section))+            }+            realmFromAnotherDevice.syncSession.uploadAllLocalChanges(30.seconds)+        }++        // Data still hasn't reached this device+        assertEquals(0, realm.query<FlexParentObject>().count().find())+        // Check that subscribing to a query, will run the query on the data downloaded from+        // the server and not just local data, due to WaitForSync.FIRST_TIME being the default.+        val result = realm.query<FlexParentObject>("section = $0", section).subscribe()+        assertEquals(1, result.size)+        assertEquals(section, result.first().section)+    }++    @Test+    fun realmQuery_subscribe_waitFirstTime() = runBlocking<Unit> {

+package io.realm.kotlin.test.mongodb.shared++import io.realm.kotlin.Realm+import io.realm.kotlin.entities.sync.flx.FlexChildObject+import io.realm.kotlin.entities.sync.flx.FlexEmbeddedObject+import io.realm.kotlin.entities.sync.flx.FlexParentObject+import io.realm.kotlin.ext.query+import io.realm.kotlin.internal.platform.runBlocking+import io.realm.kotlin.mongodb.ext.subscribe+import io.realm.kotlin.mongodb.subscriptions+import io.realm.kotlin.mongodb.sync.Subscription+import io.realm.kotlin.mongodb.sync.SubscriptionSetState+import io.realm.kotlin.mongodb.sync.SyncConfiguration+import io.realm.kotlin.mongodb.sync.WaitForSync+import io.realm.kotlin.mongodb.syncSession+import io.realm.kotlin.query.RealmResults+import io.realm.kotlin.test.mongodb.TEST_APP_FLEX+import io.realm.kotlin.test.mongodb.TestApp+import io.realm.kotlin.test.mongodb.createUserAndLogIn+import io.realm.kotlin.test.util.TestHelper+import io.realm.kotlin.test.util.use+import kotlinx.coroutines.TimeoutCancellationException+import kotlin.random.Random+import kotlin.test.AfterTest+import kotlin.test.BeforeTest+import kotlin.test.Test+import kotlin.test.assertEquals+import kotlin.test.assertFailsWith+import kotlin.test.assertNotEquals+import kotlin.test.assertNull+import kotlin.time.Duration.Companion.nanoseconds+import kotlin.time.Duration.Companion.seconds++/**+ * Class for testing the various extension methods we have for bridging the gap between Subscriptions+ * and RealmQuery/RealmResults.+ */+class SubscriptionExtensionsTests {++    private lateinit var app: TestApp+    private lateinit var realm: Realm++    @BeforeTest+    fun setup() {+        app = TestApp(appName = TEST_APP_FLEX)+        val (email, password) = TestHelper.randomEmail() to "password1234"+        val user = runBlocking {+            app.createUserAndLogIn(email, password)+        }+        val config = SyncConfiguration.Builder(+            user,+            schema = setOf(FlexParentObject::class, FlexChildObject::class, FlexEmbeddedObject::class)+        )+            .build()+        realm = Realm.open(config)+    }++    @AfterTest+    fun tearDown() {+        if (this::realm.isInitialized && !realm.isClosed()) {+            realm.close()+        }+        if (this::app.isInitialized) {+            app.close()+        }+    }++    @Test+    fun realmQuery_subscribe_anonymous() = runBlocking {+        val subs = realm.subscriptions+        assertEquals(0, subs.size)+        val results: RealmResults<FlexParentObject> = realm.query<FlexParentObject>().subscribe()+        assertEquals(SubscriptionSetState.COMPLETE, subs.state)+        assertEquals(1, subs.size)+        val sub: Subscription = subs.first()+        assertNull(sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals(FlexParentObject::class.simpleName, sub.objectType)+    }++    // Check that subscribing twice to a query will result in the same subscription+    @Test+    fun realmQuery_subscribe_anonymousTwice() = runBlocking {+        val subs = realm.subscriptions+        assertEquals(0, subs.size)+        realm.query<FlexParentObject>().subscribe()+        realm.query<FlexParentObject>().subscribe()+        assertEquals(SubscriptionSetState.COMPLETE, subs.state)+        assertEquals(1, subs.size)+        val sub: Subscription = subs.first()+        assertNull(sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals(FlexParentObject::class.simpleName, sub.objectType)+    }++    // Check that anonymous RealmQuery and RealmResults .subscribe calls result in the same sub.+    @Test+    fun anonymousSubscriptionsOverlap() = runBlocking {+        val subs = realm.subscriptions+        assertEquals(0, subs.size)+        realm.query<FlexParentObject>().subscribe()+        realm.query<FlexParentObject>().find().subscribe()+        assertEquals(SubscriptionSetState.COMPLETE, subs.state)+        assertEquals(1, subs.size)+        val sub: Subscription = subs.first()+        assertNull(sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals(FlexParentObject::class.simpleName, sub.objectType)+    }++    // Verify that the realm query doesn't run against a frozen version previous to the Realm+    // being updated from `waitForSynchronization`.+    @Test+    fun realmQuery_subscribe_queryResultIsLatestVersion() = runBlocking {+        // Write data to a server Realm+        val section = Random.nextInt()+        val (email, password) = TestHelper.randomEmail() to "password1234"+        val user1 = app.createUserAndLogIn(email, password)+        val config = SyncConfiguration.Builder(+            user1,+            schema = setOf(FlexParentObject::class, FlexChildObject::class, FlexEmbeddedObject::class)+        ).initialSubscriptions { realm: Realm ->+            realm.query<FlexParentObject>("section = $0", section).subscribe()+        }.build()++        Realm.open(config).use { realmFromAnotherDevice ->+            realmFromAnotherDevice.writeBlocking {+                copyToRealm(FlexParentObject(section))+            }+            realmFromAnotherDevice.syncSession.uploadAllLocalChanges(30.seconds)+        }++        // Data still hasn't reached this device+        assertEquals(0, realm.query<FlexParentObject>().count().find())+        // Check that subscribing to a query, will run the query on the data downloaded from+        // the server and not just local data, due to WaitForSync.FIRST_TIME being the default.+        val result = realm.query<FlexParentObject>("section = $0", section).subscribe()+        assertEquals(1, result.size)+        assertEquals(section, result.first().section)+    }++    @Test+    fun realmQuery_subscribe_waitFirstTime() = runBlocking<Unit> {+        // Unnamed+        realm.query<FlexParentObject>().subscribe() // Default value is WaitForSync.FIRST_TIME+        var updatedSubs = realm.subscriptions+        assertEquals(1, updatedSubs.size)+        assertEquals(SubscriptionSetState.COMPLETE, updatedSubs.state)+        var sub: Subscription = updatedSubs.first()+        assertNull(sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals("FlexParentObject", sub.objectType)++        // Named+        realm.query<FlexParentObject>().subscribe("my-name") // Default value is WaitForSync.FIRST_TIME+        updatedSubs = realm.subscriptions+        assertEquals(2, updatedSubs.size)+        assertEquals(SubscriptionSetState.COMPLETE, updatedSubs.state)+        sub = updatedSubs.last()+        assertEquals("my-name", sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals("FlexParentObject", sub.objectType)+    }++    @Test+    fun realmQuery_subscribe_waitNever() = runBlocking {+        // Un-named+        realm.query<FlexParentObject>().subscribe(mode = WaitForSync.NEVER)+        var updatedSubs = realm.subscriptions+        assertEquals(1, updatedSubs.size)+        // Updating the subscription will happen in the background, but+        // hopefully hasn't reached COMPLETE yet.+        assertNotEquals(SubscriptionSetState.COMPLETE, updatedSubs.state)++        // Named+        realm.query<FlexParentObject>().subscribe(name = "my-name", mode = WaitForSync.NEVER)+        updatedSubs = realm.subscriptions+        assertEquals(2, updatedSubs.size)+        // Updating the subscription will happen in the background, but+        // hopefully hasn't reached COMPLETE yet.+        assertNotEquals(SubscriptionSetState.COMPLETE, updatedSubs.state)+    }++    @Test+    fun realmQuery_subscribe_waitAlways() = runBlocking {+        val sectionId = Random.nextInt()+        val results1 = realm.query<FlexParentObject>("section = $0", sectionId).subscribe() // Default value is WaitForSync.FIRST_TIME+        assertEquals(0, results1.size)+        uploadServerData(sectionId, 5)+        // Since the subscription is already present, we cannot control if the data is downloaded+        // before creating the next subscription. Instead we pause the syncSession and verify+        // that WaitForSync.ALWAYS timeout during network failures and resuming the session should+        // then work+        realm.syncSession.pause()+        assertFailsWith<TimeoutCancellationException> {+            realm.query<FlexParentObject>("section = $0", sectionId).subscribe(timeout = 3.seconds, mode = WaitForSync.ALWAYS)+        }+        realm.syncSession.resume()+        val results2 = realm.query<FlexParentObject>("section = $0", sectionId).subscribe(mode = WaitForSync.ALWAYS)+        assertEquals(5, results2.size)+    }++    @Test+    fun realmQuery_subscribe_timeOut_fails() = runBlocking<Unit> {+        assertFailsWith<TimeoutCancellationException> {+            realm.query<FlexParentObject>().subscribe(timeout = 1.nanoseconds)+        }+    }++    @Test+    fun realmQuery_subscribe_throwsInsideWrite() {+        realm.writeBlocking {+            // `subscribe()` being a suspend function make in hard to call+            // subscribe inside a write, but we should still detect it.+            runBlocking {+                assertFailsWith<IllegalStateException> {+                    query<FlexParentObject>().subscribe()+                }+                assertFailsWith<IllegalStateException> {+                    query<FlexParentObject>().subscribe(name = "my-name")+                }+            }+        }+    }++    @Test+    fun realmResults_subscribe_waitFirstTime() = runBlocking {+        // Unnamed+        realm.query<FlexParentObject>().find().subscribe() // Default value is WaitForSync.FIRST_TIME+        var updatedSubs = realm.subscriptions+        assertEquals(1, updatedSubs.size)+        assertEquals(SubscriptionSetState.COMPLETE, updatedSubs.state)+        var sub: Subscription = updatedSubs.first()+        assertNull(sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals("FlexParentObject", sub.objectType)++        // Named+        realm.query<FlexParentObject>().find().subscribe("my-name") // Default value is WaitForSync.FIRST_TIME+        updatedSubs = realm.subscriptions+        assertEquals(2, updatedSubs.size)+        assertEquals(SubscriptionSetState.COMPLETE, updatedSubs.state)+        sub = updatedSubs.last()+        assertEquals("my-name", sub.name)+        assertEquals("TRUEPREDICATE ", sub.queryDescription)+        assertEquals("FlexParentObject", sub.objectType)+    }++    @Test+    fun realmResults_subscribe_waitOnNever() = runBlocking {+        // Un-named+        realm.query<FlexParentObject>().find().subscribe(mode = WaitForSync.NEVER)+        var updatedSubs = realm.subscriptions+        assertEquals(1, updatedSubs.size)+        // Updating the subscription will happen in the background, but+        // hopefully hasn't reached COMPLETE yet.+        assertNotEquals(SubscriptionSetState.COMPLETE, updatedSubs.state)++        // Named+        realm.query<FlexParentObject>().find().subscribe(name = "my-name", mode = WaitForSync.NEVER)+        updatedSubs = realm.subscriptions+        assertEquals(2, updatedSubs.size)+        // Updating the subscription will happen in the background, but+        // hopefully hasn't reached COMPLETE yet.+        assertNotEquals(SubscriptionSetState.COMPLETE, updatedSubs.state)+    }++    @Test+    fun realmResults_subscribe_waitAlways() = runBlocking {+        val sectionId = Random.nextInt()+        val results1 = realm.query<FlexParentObject>("section = $0", sectionId).find().subscribe() // Default value is WaitForSync.FIRST_TIME+        assertEquals(0, results1.size)+        uploadServerData(sectionId, 5)+        // Since the subscription is already present, we cannot control if the data is downloaded+        // before creating the next subscription. Instead we pause the syncSession and verify+        // that WaitForSync.ALWAYS timeout during network failures and resuming the session should+        // then work+        realm.syncSession.pause()+        assertFailsWith<TimeoutCancellationException> {+            realm.query<FlexParentObject>("section = $0", sectionId).find().subscribe(timeout = 3.seconds, mode = WaitForSync.ALWAYS)+        }+        realm.syncSession.resume()+        val results2 = realm.query<FlexParentObject>("section = $0", sectionId).find().subscribe(mode = WaitForSync.ALWAYS)+        assertEquals(5, results2.size)+    }++    @Test+    fun realmResults_subscribe_subquery() = runBlocking {+        val topQueryResult: RealmResults<FlexParentObject> = realm.query<FlexParentObject>("section = 42").find()+        val subQueryResult: RealmResults<FlexParentObject> = topQueryResult.query("name == $0", "Jane").find()+        subQueryResult.subscribe()+        val subs = realm.subscriptions+        assertEquals(1, subs.size)+        assertEquals("section == 42 and name == \"Jane\" ", subs.first().queryDescription)+        subQueryResult.subscribe("my-name")+        assertEquals(2, subs.size)+        val lastSub = subs.last()+        assertEquals("my-name", lastSub.name)+        assertEquals("section == 42 and name == \"Jane\" ", lastSub.queryDescription)+    }++    @Test+    fun realmResults_subscribe_timeOut_fails() = runBlocking<Unit> {+        assertFailsWith<TimeoutCancellationException> {+            realm.query<FlexParentObject>().find().subscribe(timeout = 1.nanoseconds)

+@file:Suppress("invisible_reference", "invisible_member")+package io.realm.kotlin.mongodb.ext++import io.realm.kotlin.mongodb.annotations.ExperimentalFlexibleSyncApi+import io.realm.kotlin.mongodb.sync.Subscription+import io.realm.kotlin.mongodb.sync.SubscriptionSet+import io.realm.kotlin.mongodb.sync.WaitForSync+import io.realm.kotlin.query.RealmQuery+import io.realm.kotlin.query.RealmResults+import io.realm.kotlin.types.RealmObject+import kotlin.time.Duration++/**+ * Automatically create an anonymous [Subscription] from a local query result in the background and+ * return the result of re-running the same query against the Realm file.+ *+ * This is a more streamlined alternative to doing something like this:+ *+ * ```+ * fun suspend getData(realm: Realm): RealmResults<Person> {+ *     val results = realm.query<Person>().find()+ *     realm.subscriptions.update { bgRealm ->+ *         add("myquery", results.query(""))+ *     }+ *     realm.subscriptions.waitForSynchronization()+ *     return realm.query<Person>().find()+ * }+ * ```+ *+ * It is possible to define whether or not to wait for the server to send all data before+ * running the local query. This is relevant as there might be delay from creating a subscription+ * to the data being available on the device due to either latency or because a large dataset needs+ * be downloaded.+ *+ * The default behaviour is that the first time `subscribe` is called, the query result will not+ * be returned until data has been downloaded from the server. On subsequent calls to `subscribe`+ * for the same query, the query will run immediately on the local database while any updates+ * are downloaded in the background.+ *+ * @param name name of the subscription. This can be used to identify it later in the [SubscriptionSet].+ * @param mode type of mode used to resolve the subscription. See [WaitForSync] for more details.+ * @param timeout How long to wait for the server to return the objects defined by the subscription.+ * This is only relevant for [WaitForSync.ALWAYS] and [WaitForSync.FIRST_TIME].+ * @return The result of running the query against the local Realm file. The results returned will+ * depend on which [mode] was used.+ * @throws kotlinx.coroutines.TimeoutCancellationException if the specified timeout was hit before+ * a query result could be returned.+ * @Throws IllegalStateException if this method is called on a Realm that isn't using Flexible Sync.+ */@ExperimentalFlexibleSyncApi+public suspend fun <T : RealmObject> RealmResults<T>.subscribe(+    mode: WaitForSync = WaitForSync.FIRST_TIME,+    timeout: Duration = Duration.INFINITE+): RealmResults<T> {+    val query: RealmQuery<T> = this.query("")+    return query.subscribe(mode, timeout)+}++/**+ * Automatically create a named [Subscription] from a local query result in the background and+ * return the result of re-running the same query against the Realm file.+ *+ * This is a more streamlined alternative to doing something like this:+ *+ * ```+ * fun suspend getData(realm: Realm): RealmResults<Person> {+ *     val results = realm.query<Person>().find()+ *     realm.subscriptions.update { bgRealm ->+ *         add("myquery", results.query(""))+ *     }+ *     realm.subscriptions.waitForSynchronization()+ *     return realm.query<Person>().find()+ * }+ * ```+ *+ * It is possible to define whether or not to wait for the server to send all data before+ * running the local query. This is relevant as there might be delay from creating a subscription+ * to the data being available on the device due to either latency or because a large dataset needs+ * be downloaded.+ *+ * The default behaviour is that the first time `subscribe` is called, the query result will not+ * be returned until data has been downloaded from the server. On subsequent calls to `subscribe`+ * for the same query, the query will run immediately on the local database while any updates+ * are downloaded in the background.+ *+ * @param name name of the subscription. This can be used to identify it later in the [SubscriptionSet].+ * @param mode type of mode used to resolve the subscription. See [WaitForSync] for more details.