|
| 1 | +package com.powersync.sync |
| 2 | + |
| 3 | +import com.powersync.ExperimentalPowerSyncAPI |
| 4 | +import com.powersync.bucket.PowerSyncControlArguments |
| 5 | +import com.powersync.connectors.PowerSyncCredentials |
| 6 | +import com.powersync.utils.JsonUtil |
| 7 | +import io.ktor.client.HttpClient |
| 8 | +import io.ktor.client.plugins.websocket.webSocketSession |
| 9 | +import io.ktor.http.URLBuilder |
| 10 | +import io.ktor.http.URLProtocol |
| 11 | +import io.ktor.http.takeFrom |
| 12 | +import io.rsocket.kotlin.core.RSocketConnector |
| 13 | +import io.rsocket.kotlin.keepalive.KeepAlive |
| 14 | +import io.rsocket.kotlin.payload.PayloadMimeType |
| 15 | +import io.rsocket.kotlin.payload.buildPayload |
| 16 | +import io.rsocket.kotlin.payload.data |
| 17 | +import io.rsocket.kotlin.payload.metadata |
| 18 | +import io.rsocket.kotlin.transport.RSocketClientTarget |
| 19 | +import io.rsocket.kotlin.transport.RSocketConnection |
| 20 | +import io.rsocket.kotlin.transport.RSocketTransportApi |
| 21 | +import io.rsocket.kotlin.transport.ktor.websocket.internal.KtorWebSocketConnection |
| 22 | +import kotlinx.coroutines.Dispatchers |
| 23 | +import kotlinx.coroutines.IO |
| 24 | +import kotlinx.coroutines.currentCoroutineContext |
| 25 | +import kotlinx.coroutines.flow.Flow |
| 26 | +import kotlinx.coroutines.flow.emitAll |
| 27 | +import kotlinx.coroutines.flow.flow |
| 28 | +import kotlinx.coroutines.flow.flowOn |
| 29 | +import kotlinx.coroutines.flow.map |
| 30 | +import kotlinx.io.readByteArray |
| 31 | +import kotlinx.serialization.SerialName |
| 32 | +import kotlinx.serialization.Serializable |
| 33 | +import kotlinx.serialization.json.JsonElement |
| 34 | +import kotlin.coroutines.CoroutineContext |
| 35 | +import kotlin.time.Duration.Companion.seconds |
| 36 | + |
| 37 | +/** |
| 38 | + * Connects to the RSocket endpoint for receiving sync lines. |
| 39 | + * |
| 40 | + * Note that we reconstruct the transport layer for RSocket by opening a WebSocket connection |
| 41 | + * manually instead of using the high-level RSocket Ktor integration. |
| 42 | + * The reason is that every request to the sync service needs its own metadata and data payload |
| 43 | + * (e.g. to transmit the token), but the Ktor integration only supports setting a single payload for |
| 44 | + * the entire client. |
| 45 | + */ |
| 46 | +@OptIn(RSocketTransportApi::class, ExperimentalPowerSyncAPI::class) |
| 47 | +internal fun HttpClient.rSocketSyncStream( |
| 48 | + userAgent: String, |
| 49 | + req: JsonElement, |
| 50 | + credentials: PowerSyncCredentials, |
| 51 | +): Flow<PowerSyncControlArguments> = |
| 52 | + flow { |
| 53 | + val flowContext = currentCoroutineContext() |
| 54 | + |
| 55 | + val websocketUri = |
| 56 | + URLBuilder(credentials.endpointUri("sync/stream")).apply { |
| 57 | + protocol = |
| 58 | + when (protocolOrNull) { |
| 59 | + URLProtocol.HTTP -> URLProtocol.WS |
| 60 | + else -> URLProtocol.WSS |
| 61 | + } |
| 62 | + } |
| 63 | + |
| 64 | + // Note: We're using a custom connector here because we need to set options for each request |
| 65 | + // without creating a new HTTP client each time. The recommended approach would be to add an |
| 66 | + // RSocket extension to the HTTP client, but that only allows us to set the SETUP metadata for |
| 67 | + // all connections (bad because we need a short-lived token in there). |
| 68 | + // https://github.com/rsocket/rsocket-kotlin/issues/311 |
| 69 | + val target = |
| 70 | + object : RSocketClientTarget { |
| 71 | + @RSocketTransportApi |
| 72 | + override suspend fun connectClient(): RSocketConnection { |
| 73 | + val ws = |
| 74 | + webSocketSession { |
| 75 | + url.takeFrom(websocketUri) |
| 76 | + } |
| 77 | + return KtorWebSocketConnection(ws) |
| 78 | + } |
| 79 | + |
| 80 | + override val coroutineContext: CoroutineContext |
| 81 | + get() = flowContext |
| 82 | + } |
| 83 | + |
| 84 | + val connector = |
| 85 | + RSocketConnector { |
| 86 | + connectionConfig { |
| 87 | + payloadMimeType = |
| 88 | + PayloadMimeType( |
| 89 | + metadata = "application/json", |
| 90 | + data = "application/json", |
| 91 | + ) |
| 92 | + |
| 93 | + setupPayload { |
| 94 | + buildPayload { |
| 95 | + data("{}") |
| 96 | + metadata( |
| 97 | + JsonUtil.json.encodeToString( |
| 98 | + ConnectionSetupMetadata( |
| 99 | + token = "Bearer ${credentials.token}", |
| 100 | + userAgent = userAgent, |
| 101 | + ), |
| 102 | + ), |
| 103 | + ) |
| 104 | + } |
| 105 | + } |
| 106 | + |
| 107 | + keepAlive = KeepAlive(interval = 20.0.seconds, maxLifetime = 30.0.seconds) |
| 108 | + } |
| 109 | + } |
| 110 | + |
| 111 | + val rSocket = connector.connect(target) |
| 112 | + emit(PowerSyncControlArguments.ConnectionEstablished) |
| 113 | + val syncStream = |
| 114 | + rSocket.requestStream( |
| 115 | + buildPayload { |
| 116 | + data(JsonUtil.json.encodeToString(req)) |
| 117 | + metadata(JsonUtil.json.encodeToString(RequestStreamMetadata("/sync/stream"))) |
| 118 | + }, |
| 119 | + ) |
| 120 | + |
| 121 | + emitAll( |
| 122 | + syncStream |
| 123 | + .map { |
| 124 | + PowerSyncControlArguments.BinaryLine(it.data.readByteArray()) |
| 125 | + }.flowOn(Dispatchers.IO), |
| 126 | + ) |
| 127 | + emit(PowerSyncControlArguments.ResponseStreamEnd) |
| 128 | + } |
| 129 | + |
| 130 | +/** |
| 131 | + * The metadata payload we need to use when connecting with RSocket. |
| 132 | + * |
| 133 | + * This corresponds to `RSocketContextMeta` on the sync service. |
| 134 | + */ |
| 135 | +@Serializable |
| 136 | +private class ConnectionSetupMetadata( |
| 137 | + val token: String, |
| 138 | + @SerialName("user_agent") |
| 139 | + val userAgent: String, |
| 140 | +) |
| 141 | + |
| 142 | +/** |
| 143 | + * The metadata payload we send for the `REQUEST_STREAM` frame. |
| 144 | + */ |
| 145 | +@Serializable |
| 146 | +private class RequestStreamMetadata( |
| 147 | + val path: String, |
| 148 | +) |
0 commit comments