-
-
Notifications
You must be signed in to change notification settings - Fork 38
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #696 from supabase-community/realtime-event-refactor
Refactor Realtime event system
- Loading branch information
Showing
17 changed files
with
367 additions
and
127 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
23 changes: 23 additions & 0 deletions
23
Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/event/RBroadcastEvent.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,23 @@ | ||
| package io.github.jan.supabase.realtime.event | ||
|
|
||
| import io.github.jan.supabase.realtime.RealtimeChannel | ||
| import io.github.jan.supabase.realtime.RealtimeMessage | ||
| import kotlinx.serialization.json.JsonObject | ||
| import kotlinx.serialization.json.jsonObject | ||
| import kotlinx.serialization.json.jsonPrimitive | ||
|
|
||
| /** | ||
| * Handles broadcast events | ||
| */ | ||
| data object RBroadcastEvent : RealtimeEvent { | ||
|
|
||
| override suspend fun handle(channel: RealtimeChannel, message: RealtimeMessage) { | ||
| val event = message.payload["event"]?.jsonPrimitive?.content ?: "" | ||
| channel.callbackManager.triggerBroadcast(event, message.payload["payload"]?.jsonObject ?: JsonObject(mutableMapOf())) | ||
| } | ||
|
|
||
| override fun appliesTo(message: RealtimeMessage): Boolean { | ||
| return message.event == RealtimeChannel.CHANNEL_EVENT_BROADCAST | ||
| } | ||
|
|
||
| } |
22 changes: 22 additions & 0 deletions
22
Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/event/RCloseEvent.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,22 @@ | ||
| package io.github.jan.supabase.realtime.event | ||
|
|
||
| import io.github.jan.supabase.logging.d | ||
| import io.github.jan.supabase.realtime.Realtime | ||
| import io.github.jan.supabase.realtime.RealtimeChannel | ||
| import io.github.jan.supabase.realtime.RealtimeMessage | ||
|
|
||
| /** | ||
| * Event that handles the closing of a channel | ||
| */ | ||
| data object RCloseEvent : RealtimeEvent { | ||
|
|
||
| override suspend fun handle(channel: RealtimeChannel, message: RealtimeMessage) { | ||
| channel.realtime.removeChannel(channel) | ||
| Realtime.logger.d { "Unsubscribed from channel ${message.topic}" } | ||
| } | ||
|
|
||
| override fun appliesTo(message: RealtimeMessage): Boolean { | ||
| return message.event == RealtimeChannel.CHANNEL_EVENT_CLOSE | ||
| } | ||
|
|
||
| } |
21 changes: 21 additions & 0 deletions
21
Realtime/src/commonMain/kotlin/io/github/jan/supabase/realtime/event/RErrorEvent.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,21 @@ | ||
| package io.github.jan.supabase.realtime.event | ||
|
|
||
| import io.github.jan.supabase.logging.e | ||
| import io.github.jan.supabase.realtime.Realtime | ||
| import io.github.jan.supabase.realtime.RealtimeChannel | ||
| import io.github.jan.supabase.realtime.RealtimeMessage | ||
|
|
||
| /** | ||
| * Event that handles an error event | ||
| */ | ||
| data object RErrorEvent : RealtimeEvent { | ||
|
|
||
| override suspend fun handle(channel: RealtimeChannel, message: RealtimeMessage) { | ||
| Realtime.logger.e { "Received an error in channel ${message.topic}. That could be as a result of an invalid access token" } | ||
| } | ||
|
|
||
| override fun appliesTo(message: RealtimeMessage): Boolean { | ||
| return message.event == RealtimeChannel.CHANNEL_EVENT_ERROR | ||
| } | ||
|
|
||
| } |
59 changes: 59 additions & 0 deletions
59
...time/src/commonMain/kotlin/io/github/jan/supabase/realtime/event/RPostgresChangesEvent.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,59 @@ | ||
| package io.github.jan.supabase.realtime.event | ||
|
|
||
| import io.github.jan.supabase.realtime.PostgresAction | ||
| import io.github.jan.supabase.realtime.RealtimeChannel | ||
| import io.github.jan.supabase.realtime.RealtimeMessage | ||
| import io.github.jan.supabase.realtime.data.PostgresActionData | ||
| import io.github.jan.supabase.realtime.realtime | ||
| import io.github.jan.supabase.supabaseJson | ||
| import kotlinx.serialization.json.decodeFromJsonElement | ||
| import kotlinx.serialization.json.jsonArray | ||
| import kotlinx.serialization.json.jsonObject | ||
| import kotlinx.serialization.json.jsonPrimitive | ||
| import kotlinx.serialization.json.longOrNull | ||
|
|
||
| /** | ||
| * Handles postgres changes events | ||
| */ | ||
| data object RPostgresChangesEvent : RealtimeEvent { | ||
|
|
||
| override suspend fun handle(channel: RealtimeChannel, message: RealtimeMessage) { | ||
| val data = message.payload["data"]?.jsonObject ?: return | ||
| val ids = message.payload["ids"]?.jsonArray?.mapNotNull { it.jsonPrimitive.longOrNull } ?: emptyList() //the ids of the matching postgres changes | ||
| val postgresAction = supabaseJson.decodeFromJsonElement<PostgresActionData>(data) | ||
| val action = when(data["type"]?.jsonPrimitive?.content ?: "") { | ||
| "UPDATE" -> PostgresAction.Update( | ||
| postgresAction.record ?: error("Received no record on update event"), | ||
| postgresAction.oldRecord ?: error("Received no old record on update event"), | ||
| postgresAction.columns, | ||
| postgresAction.commitTimestamp, | ||
| channel.supabaseClient.realtime.serializer | ||
| ) | ||
| "DELETE" -> PostgresAction.Delete( | ||
| postgresAction.oldRecord ?: error("Received no old record on delete event"), | ||
| postgresAction.columns, | ||
| postgresAction.commitTimestamp, | ||
| channel.supabaseClient.realtime.serializer | ||
| ) | ||
| "INSERT" -> PostgresAction.Insert( | ||
| postgresAction.record ?: error("Received no record on update event"), | ||
| postgresAction.columns, | ||
| postgresAction.commitTimestamp, | ||
| channel.supabaseClient.realtime.serializer | ||
| ) | ||
| "SELECT" -> PostgresAction.Select( | ||
| postgresAction.record ?: error("Received no record on update event"), | ||
| postgresAction.columns, | ||
| postgresAction.commitTimestamp, | ||
| channel.supabaseClient.realtime.serializer | ||
| ) | ||
| else -> error("Unknown event type ${message.event}") | ||
| } | ||
| channel.callbackManager.triggerPostgresChange(ids, action) | ||
| } | ||
|
|
||
| override fun appliesTo(message: RealtimeMessage): Boolean { | ||
| return message.event == RealtimeChannel.CHANNEL_EVENT_POSTGRES_CHANGES | ||
| } | ||
|
|
||
| } |
Oops, something went wrong.