Kotlin: Subscribe to channel

Subscribe to realtime changes in your database.

• Realtime is disabled by default for new Projects for better database performance and security. You can turn it on by managing replication.
• If you want to receive the "previous" data for updates and deletes, you will need to set REPLICA IDENTITY to FULL, like this: ALTER TABLE your_table REPLICA IDENTITY FULL;
• When using a method with a generic type like track, broadcast or broadcastFlow, you have to provide a serializable class as the type parameter.
• Presence, Broadcast and Database updates are sent through a Flow

Examples

Listen to broadcasts

@Serializable
data class Message(val content: String, val sender: String)

val channel = supabase.channel("channelId") \{
    // optional config
\}

val broadcastFlow = channel.broadcastFlow<Message>(event = "message")

// Collect the flow
broadcastFlow.onEach \{ // it: Message
    println(it)
\}.launchIn(coroutineScope) // launch a new coroutine to collect the flow

channel.subscribe(blockUntilSubscribed = true)

channel.broadcast(event = "message", Message("I joined!", "John"))

Listen to presence updates

@Serializable
data class PresenceState(val username: String)

val channel = supabase.channel("channelId") \{
    //optional config
\}

//if you want more control and want to receive the raw data, use the `presenceChangeFlow` method
val presenceFlow = channel.presenceDataFlow<PresenceState>() 

//Collect the flow
presenceFlow.onEach \{
    for (presence in it) \{
        println(presence.username)
    \}
\}.launchIn(coroutineScope) // launch a new coroutine to collect the flow

channel.subscribe(blockUntilSubscribed = true)
//send your own state
channel.track(PresenceState(username = "John"))

Listen to all database changes

val channel = supabase.channel("channelId") \{
    //optional config
\}
val changeFlow = channel.postgresChangeFlow<PostgresAction>(schema = "public")

//Collect the flow
changeFlow.onEach \{
    when(it) \{
        is PostgresAction.Delete -> println("Deleted: $\{it.oldRecord\}")
        is PostgresAction.Insert -> println("Inserted: $\{it.record\}")
        is PostgresAction.Select -> println("Selected: $\{it.record\}")
        is PostgresAction.Update -> println("Updated: $\{it.oldRecord\} with $\{it.record\}")
    \}
\}.launchIn(coroutineScope) // launch a new coroutine to collect the flow

channel.subscribe()

Listen to a specific table

val channel = supabase.channel("channelId") \{
    //optional config
\}

//This flow will automatically emit the initial data of all messages (which can be restricted by specifying a filter) and then listens for updates.
//The data is cached automatically by the primary key you specify
val users: Flow<List<User>> = channel.postgresListDataFlow(schema = "public", table = "users", primaryKey = User::id)

//If you want more control and want to receive the raw data use the `postgresChangeFlow` method:
val changeFlow = channel.postgresChangeFlow<PostgresAction>(schema = "public") \{
    table = "users"
\}

//Collect the flow
changeFlow.onEach \{
    when(it) \{
        is PostgresAction.Delete -> println("Deleted: $\{it.oldRecord\}")
        is PostgresAction.Insert -> println("Inserted: $\{it.record\}")
        is PostgresAction.Select -> println("Selected: $\{it.record\}")
        is PostgresAction.Update -> println("Updated: $\{it.oldRecord\} with $\{it.record\}")
    \}
\}.launchIn(coroutineScope) // launch a new coroutine to collect the flow

channel.subscribe()

Listen to inserts

val channel = supabase.channel("channelId") \{
    //optional config
\}
val changeFlow = channel.postgresChangeFlow<PostgresAction.Insert>(schema = "public") \{
    table = "users"
\}

//Collect the flow
changeFlow.onEach \{
    println(it.record)
\}.launchIn(coroutineScope) // launch a new coroutine to collect the flow

channel.subscribe()

Listen to updates

val channel = supabase.channel("channelId") \{
    //optional config
\}

val changeFlow = channel.postgresChangeFlow<PostgresAction.Update>(schema = "public") \{
    table = "users"
\}

//Collect the flow
changeFlow.onEach \{
    println(it.record)
    println(it.oldRecord)
\}.launchIn(coroutineScope) // launch a new coroutine to collect the flow

channel.subscribe()

Listen to deletes

val channel = supabase.channel("channelId") \{
    //optional config
\}

val changeFlow = channel.postgresChangeFlow<PostgresAction.Delete>(schema = "public") \{
    table = "users"
\}

//Collect the flow
changeFlow.onEach \{
    println(it.oldRecord)
\}.launchIn(coroutineScope) // launch a new coroutine to collect the flow

channel.subscribe()

Listen to row level changes

val channel = supabase.channel("channelId") \{
   //optional config
\}

//This flow will automatically emit the user's initial data and then emits new values on update. The flow ends on deletion
val user: Flow<User> = channel.postgresSingleDataFlow(schema = "public", table = "users", primaryKey = User::id) \{
    //This is the same filter as used in postgrest, so you could use complex queries, but only one entry is fetched
    eq("id", 1)
\}

//If you want more control, use the `postgresChangeFlow` method which works the same as the other examples
val changeFlow = channel.postgresChangeFlow<PostgresAction.Delete>(schema = "public") \{
    table = "users"
    filter("id", FilterOperator.EQ, 1)
\}

//Collect the flow
changeFlow.onEach \{
    println(it.oldRecord)
\}.launchIn(coroutineScope) // launch a new coroutine to collect the flow

channel.subscribe()