Applying Realtime events¶
Realtime stays on supabase-kt: the library never opens a channel. Hand each record a channel
delivers to applyServerRecord or applyServerDelete, and every cached list of the table shows the
change at once, without fetching again, so a change made elsewhere (another device, the dashboard)
does not wait for the next fetch:
From the samples: LiveUpdates.kt · TaskRepository+Server.swift
/** A change to `tasks`: the row decoded into the app's own type and applied, or a delete. */
suspend fun applyTaskChange(change: PostgresAction) {
when (change) {
is PostgresAction.Insert ->
offline.applyServerRecord(Tables.TASKS, change.decodeRecord<TaskRecord>())
is PostgresAction.Update ->
offline.applyServerRecord(Tables.TASKS, change.decodeRecord<TaskRecord>())
// With row-level security, an old record carries the key columns only: enough.
is PostgresAction.Delete -> offline.applyServerDelete(Tables.TASKS, change.oldRecord)
is PostgresAction.Select -> Unit // never sent for these subscriptions
}
}
/** A change to `task_photos`, applied as the JSON the server sent. */
suspend fun applyPhotoChange(change: PostgresAction) {
when (change) {
is PostgresAction.Insert -> offline.applyServerRecord(Tables.TASK_PHOTOS, change.record)
is PostgresAction.Update -> offline.applyServerRecord(Tables.TASK_PHOTOS, change.record)
is PostgresAction.Delete ->
offline.applyServerDelete(Tables.TASK_PHOTOS, change.oldRecord)
is PostgresAction.Select -> Unit
}
}
/// Applies a row change the server reported to every cached list of its table, so the screens
/// show it without asking the server. Tasks use the typed forms, photos the JSON forms.
func apply(_ change: ServerChange) async throws {
switch change {
case .saved(Tables.tasks, let record):
try await client.applyServerRecord(table: Tables.tasks, record: try record.decoded(as: ReportedTaskRow.self))
case .saved(let table, let record):
try await client.applyServerRecord(table: table, record: record)
case .deleted(Tables.tasks, let oldRecord):
try await client.applyServerDelete(table: Tables.tasks, oldRecord: try oldRecord.decoded(as: RowKey.self))
case .deleted(let table, let oldRecord):
try await client.applyServerDelete(table: table, oldRecord: oldRecord)
}
}
applyServerRecord(table, record)replaces the cached row with the same key in every cached list of the table, and adds the row, after the list's own rows and past its order and limit, to each list that admits it and does not hold it.applyServerDelete(table, oldRecord)removes the row with that key from every cached list of the table.
Your user's queued writes are still laid over the result, so a change waiting to be sent keeps
showing over the server's record until it is sent. The typed overload encodes the row with the
client's PostgREST serializer (import io.github.nesfeder.supabase.offline.applyServerRecord). The
sample applies tasks typed and task_photos as JSON; its typed row has every column and no
default values, since the record replaces the cached row whole and supabase-kt's serializer leaves
a property at its default out of what it encodes.
Wiring a channel¶
With supabase-kt's Realtime plugin installed, collect the table's changes and apply each one. The sample subscribes to both tables behind its Live updates switch, on by default, and leaves the channel when the switch goes off, when nobody is signed in, and while the network is down or Simulate offline is on; it joins again when the network is back:
From the samples: LiveUpdates.kt · ServerChanges.swift · AppModel.swift
private val offline = supabase.offline // the plugin, as every call below reaches it
/**
* Subscribes to both tables and applies each change until the caller is canceled, then leaves
* the channel. [onConnected] hears whether the channel is subscribed.
*/
suspend fun applyChangesUntilCanceled(onConnected: (Boolean) -> Unit) {
val channel = supabase.channel(CHANNEL)
val tasks = channel.postgresChangeFlow<PostgresAction>(SCHEMA) { table = Tables.TASKS }
val photos =
channel.postgresChangeFlow<PostgresAction>(SCHEMA) { table = Tables.TASK_PHOTOS }
try {
coroutineScope {
launch(start = CoroutineStart.UNDISPATCHED) { tasks.collect(::applyTaskChange) }
launch(start = CoroutineStart.UNDISPATCHED) { photos.collect(::applyPhotoChange) }
launch {
channel.status.collect { onConnected(it == RealtimeChannel.Status.SUBSCRIBED) }
}
channel.subscribe()
}
} finally {
withContext(NonCancellable) { supabase.realtime.removeChannel(channel) }
}
}
/// Subscribes `channel` to every insert, update and delete of `tables`, says when it has
/// joined, and hands on each change until the subscription is canceled.
private static func listen(
to tables: [String],
on channel: RealtimeChannelV2,
report: @escaping @Sendable (ServerFeedEvent) -> Void
) async throws {
let streams = tables.map { table in (table, channel.postgresChange(AnyAction.self, schema: "public", table: table)) }
try await channel.subscribeWithError()
report(.joined)
try await withThrowingDiscardingTaskGroup { group in
for (table, actions) in streams {
group.addTask {
for await action in actions { report(.change(try ServerChange(action, table: table))) }
}
}
}
}
/// Applies the server's row changes as they arrive, for as long as the calling task runs: the
/// root view runs it while Live updates is on, a user is signed in and the network can be
/// reached. Each time the subscription joins, the lists are read again, to catch up with
/// what changed while it was not there.
func followLiveUpdates() async {
guard liveUpdatesKey.isOn, liveUpdatesKey.userID != nil else { return }
do {
for try await event in changeSource.changes(of: [Tables.tasks, Tables.taskPhotos]) {
switch event {
case .joined: try await repository.readListsAgain()
case .change(let change): try await repository.apply(change)
}
}
} catch is CancellationError {
return
} catch {
report(error)
}
}
Add realtime-kt from the same supabase-kt BOM as the rest, and install Realtime beside the
other plugins (Project setup). Each table must be
in Supabase's supabase_realtime publication, or no change of it is sent;
Set up Supabase adds the samples' two. The library is built on
supabase-kt 3.6 partly because of Realtime: 3.7.0 and 3.8.0 carry a Realtime crash
(why 3.6).
Primary keys¶
A record is matched by the table's primary key: id, unless you set the table's key with
primaryKey(table, columns) in the configuration, or pass primaryKeys to the call. A record that
lacks a key column, or holds one as null, an array or an object, cannot be located: it is logged
and dropped. A DELETE's old record carries the key under row-level security, which is all
applyServerDelete needs. For a table in another schema, pass schema; its key is the one
configured as schema.table, or id.
Limits¶
Realtime does not replay what happened while the device was disconnected. Applying records
keeps the cache current between fetches, never instead of them: read again after a reconnect, for
example with a NetworkOnly read of the same query, as the sample does each time its channel
connects:
From the samples: OfflineTaskRepository.kt · TaskRepository.swift
/**
* Reads both lists from the server once more: Realtime does not replay what happened while it
* was not connected, so each new connection catches up this way.
*/
private suspend fun readListsAgain() {
refreshTasks()
offline.from(Tables.TASK_PHOTOS).select(readStrategy = ReadStrategy.NetworkOnly) {
newestFirst() // the query of allPhotos()
}
}
/// Reads both lists from the server once more: Realtime does not replay what happened while it
/// was not connected, so each new subscription catches up this way.
func readListsAgain() async throws {
_ = try await refreshTasks(TaskQuery())
try await client.from(Tables.taskPhotos)
.select(readStrategy: .networkOnly)
.order("created_at", ascending: false)
.execute()
}
Nothing happens for a table nothing caches.