|
| 1 | +package com.powersync.sync |
| 2 | + |
| 3 | +import app.cash.turbine.turbineScope |
| 4 | +import com.powersync.ExperimentalPowerSyncAPI |
| 5 | +import com.powersync.bucket.Checkpoint |
| 6 | +import com.powersync.bucket.StreamPriority |
| 7 | +import com.powersync.testutils.bucket |
| 8 | +import com.powersync.testutils.databaseTest |
| 9 | +import com.powersync.testutils.waitFor |
| 10 | +import com.powersync.utils.JsonParam |
| 11 | +import com.powersync.utils.JsonUtil |
| 12 | +import io.kotest.matchers.collections.shouldHaveSingleElement |
| 13 | +import io.kotest.matchers.collections.shouldHaveSize |
| 14 | +import io.kotest.matchers.nulls.shouldNotBeNull |
| 15 | +import io.kotest.matchers.shouldBe |
| 16 | +import kotlinx.coroutines.DelicateCoroutinesApi |
| 17 | +import kotlinx.coroutines.delay |
| 18 | +import kotlinx.serialization.json.JsonArray |
| 19 | +import kotlinx.serialization.json.JsonObject |
| 20 | +import kotlinx.serialization.json.buildJsonArray |
| 21 | +import kotlinx.serialization.json.buildJsonObject |
| 22 | +import kotlinx.serialization.json.jsonArray |
| 23 | +import kotlinx.serialization.json.jsonObject |
| 24 | +import kotlinx.serialization.json.jsonPrimitive |
| 25 | +import kotlinx.serialization.json.put |
| 26 | +import kotlin.test.Test |
| 27 | + |
| 28 | +@OptIn(ExperimentalPowerSyncAPI::class, LegacySyncImplementation::class) |
| 29 | +class SyncStreamTest : AbstractSyncTest(true) { |
| 30 | + @Test |
| 31 | + fun `can disable default streams`() = |
| 32 | + databaseTest { |
| 33 | + database.connect( |
| 34 | + connector, |
| 35 | + options = |
| 36 | + SyncOptions( |
| 37 | + newClientImplementation = true, |
| 38 | + includeDefaultStreams = false, |
| 39 | + clientConfiguration = SyncClientConfiguration.ExistingClient(createSyncClient()), |
| 40 | + ), |
| 41 | + ) |
| 42 | + |
| 43 | + turbineScope { |
| 44 | + val turbine = database.currentStatus.asFlow().testIn(this) |
| 45 | + turbine.waitFor { it.connected && !it.downloading } |
| 46 | + |
| 47 | + requestedSyncStreams shouldHaveSingleElement { |
| 48 | + val streams = it.jsonObject["streams"]!!.jsonObject |
| 49 | + streams["include_defaults"]!!.jsonPrimitive.content shouldBe "false" |
| 50 | + |
| 51 | + true |
| 52 | + } |
| 53 | + |
| 54 | + turbine.cancelAndIgnoreRemainingEvents() |
| 55 | + } |
| 56 | + } |
| 57 | + |
| 58 | + @Test |
| 59 | + fun `subscribes with streams`() = |
| 60 | + databaseTest { |
| 61 | + val a = database.syncStream("stream", mapOf("foo" to JsonParam.String("a"))).subscribe() |
| 62 | + val b = database.syncStream("stream", mapOf("foo" to JsonParam.String("b"))).subscribe(priority = StreamPriority(1)) |
| 63 | + |
| 64 | + database.connect(connector, options = getOptions()) |
| 65 | + turbineScope { |
| 66 | + val turbine = database.currentStatus.asFlow().testIn(this) |
| 67 | + turbine.waitFor { it.connected && !it.downloading } |
| 68 | + |
| 69 | + // Should request subscriptions |
| 70 | + requestedSyncStreams shouldHaveSingleElement { |
| 71 | + val streams = it.jsonObject["streams"]!!.jsonObject |
| 72 | + val subscriptions = streams["subscriptions"]!!.jsonArray |
| 73 | + |
| 74 | + subscriptions shouldHaveSize 2 |
| 75 | + JsonUtil.json.encodeToString(subscriptions[0]) shouldBe |
| 76 | + """{"stream":"stream","parameters":{"foo":"a"},"override_priority":null}""" |
| 77 | + JsonUtil.json.encodeToString(subscriptions[1]) shouldBe |
| 78 | + """{"stream":"stream","parameters":{"foo":"b"},"override_priority":1}""" |
| 79 | + true |
| 80 | + } |
| 81 | + |
| 82 | + syncLines.send( |
| 83 | + SyncLine.FullCheckpoint( |
| 84 | + Checkpoint( |
| 85 | + lastOpId = "1", |
| 86 | + checksums = |
| 87 | + listOf( |
| 88 | + bucket( |
| 89 | + "a", |
| 90 | + 0, |
| 91 | + subscriptions = |
| 92 | + buildJsonArray { |
| 93 | + add(defaultSubscription(0)) |
| 94 | + }, |
| 95 | + ), |
| 96 | + bucket( |
| 97 | + "b", |
| 98 | + 0, |
| 99 | + priority = StreamPriority(1), |
| 100 | + subscriptions = |
| 101 | + buildJsonArray { |
| 102 | + add(defaultSubscription(1)) |
| 103 | + }, |
| 104 | + ), |
| 105 | + ), |
| 106 | + streams = listOf(stream("stream", false)), |
| 107 | + ), |
| 108 | + ), |
| 109 | + ) |
| 110 | + |
| 111 | + // Subscriptions should be active now, but not marked as synced. |
| 112 | + var status = turbine.awaitItem() |
| 113 | + for (subscription in listOf(a, b)) { |
| 114 | + val subscriptionStatus = status.forStream(subscription)!! |
| 115 | + subscriptionStatus.subscription.active shouldBe true |
| 116 | + subscriptionStatus.subscription.lastSyncedAt shouldBe null |
| 117 | + subscriptionStatus.subscription.hasExplicitSubscription shouldBe true |
| 118 | + } |
| 119 | + |
| 120 | + syncLines.send( |
| 121 | + SyncLine.CheckpointPartiallyComplete( |
| 122 | + lastOpId = "0", |
| 123 | + priority = StreamPriority(1), |
| 124 | + ), |
| 125 | + ) |
| 126 | + status = turbine.awaitItem() |
| 127 | + status.forStream(a)!!.subscription.lastSyncedAt shouldBe null |
| 128 | + status.forStream(b)!!.subscription.lastSyncedAt shouldNotBeNull {} |
| 129 | + b.waitForFirstSync() |
| 130 | + |
| 131 | + syncLines.send(SyncLine.CheckpointComplete(lastOpId = "0")) |
| 132 | + a.waitForFirstSync() |
| 133 | + |
| 134 | + turbine.cancelAndIgnoreRemainingEvents() |
| 135 | + } |
| 136 | + } |
| 137 | + |
| 138 | + @Test |
| 139 | + fun `reports default streams`() = |
| 140 | + databaseTest { |
| 141 | + database.connect(connector, options = getOptions()) |
| 142 | + turbineScope { |
| 143 | + val turbine = database.currentStatus.asFlow().testIn(this) |
| 144 | + turbine.waitFor { it.connected && !it.downloading } |
| 145 | + |
| 146 | + syncLines.send( |
| 147 | + SyncLine.FullCheckpoint( |
| 148 | + Checkpoint( |
| 149 | + lastOpId = "1", |
| 150 | + checksums = listOf(), |
| 151 | + streams = listOf(stream("default_stream", true)), |
| 152 | + ), |
| 153 | + ), |
| 154 | + ) |
| 155 | + |
| 156 | + val status = turbine.awaitItem() |
| 157 | + status.syncStreams!! shouldHaveSingleElement { |
| 158 | + it.subscription.name shouldBe "default_stream" |
| 159 | + it.subscription.parameters shouldBe null |
| 160 | + it.subscription.isDefault shouldBe true |
| 161 | + it.subscription.hasExplicitSubscription shouldBe false |
| 162 | + true |
| 163 | + } |
| 164 | + |
| 165 | + turbine.cancelAndIgnoreRemainingEvents() |
| 166 | + } |
| 167 | + } |
| 168 | + |
| 169 | + @OptIn(DelicateCoroutinesApi::class) |
| 170 | + @Test |
| 171 | + fun `changes subscriptions dynamically`() = |
| 172 | + databaseTest { |
| 173 | + database.connect(connector, options = getOptions()) |
| 174 | + turbineScope { |
| 175 | + val turbine = database.currentStatus.asFlow().testIn(this) |
| 176 | + turbine.waitFor { it.connected && !it.downloading } |
| 177 | + requestedSyncStreams.clear() |
| 178 | + |
| 179 | + val subscription = database.syncStream("a").subscribe() |
| 180 | + |
| 181 | + // Adding the subscription should reconnect |
| 182 | + turbine.waitFor { it.connected && !it.downloading } |
| 183 | + requestedSyncStreams shouldHaveSingleElement { |
| 184 | + val streams = it.jsonObject["streams"]!!.jsonObject |
| 185 | + val subscriptions = streams["subscriptions"]!!.jsonArray |
| 186 | + |
| 187 | + subscriptions shouldHaveSize 1 |
| 188 | + JsonUtil.json.encodeToString(subscriptions[0]) shouldBe """{"stream":"a","parameters":null,"override_priority":null}""" |
| 189 | + true |
| 190 | + } |
| 191 | + |
| 192 | + // Given that the subscription has a default TTL, unsubscribing should not re-subscribe. |
| 193 | + subscription.unsubscribe() |
| 194 | + delay(100) |
| 195 | + turbine.expectNoEvents() |
| 196 | + |
| 197 | + turbine.cancelAndIgnoreRemainingEvents() |
| 198 | + } |
| 199 | + } |
| 200 | + |
| 201 | + @Test |
| 202 | + fun `subscriptions update while offline`() = |
| 203 | + databaseTest { |
| 204 | + turbineScope { |
| 205 | + val turbine = database.currentStatus.asFlow().testIn(this) |
| 206 | + turbine.awaitItem() // Ignore initial |
| 207 | + |
| 208 | + // Subscribing while offline should add the stream to the subscriptions reported in the |
| 209 | + // status. |
| 210 | + val subscription = database.syncStream("foo").subscribe() |
| 211 | + val status = turbine.awaitItem() |
| 212 | + status.forStream(subscription) shouldNotBeNull {} |
| 213 | + |
| 214 | + turbine.cancelAndIgnoreRemainingEvents() |
| 215 | + } |
| 216 | + } |
| 217 | +} |
| 218 | + |
| 219 | +private fun stream( |
| 220 | + name: String, |
| 221 | + isDefault: Boolean, |
| 222 | +): JsonObject = |
| 223 | + buildJsonObject { |
| 224 | + put("name", name) |
| 225 | + put("is_default", isDefault) |
| 226 | + put("errors", JsonArray(emptyList())) |
| 227 | + } |
| 228 | + |
| 229 | +private fun defaultSubscription(index: Int): JsonObject = buildJsonObject { put("sub", index) } |
0 commit comments