Is this the correct way to implement Realtime-kt?

[Premjit-Developer]

I'm trying to integrate real-time functionality into my app using the chat demo as a reference, but I'm struggling to do so correctly. Can I receive guidance on implementing real-time functionality into my app?

DI

@Provides
@Singleton
fun providesRealtime(client: SupabaseClient): RealtimeChannel = client.realtime.channel("orders")

Cart Screen

 LaunchedEffect(Unit) {
    
        viewModel.connectToRealtime(this)
      
    }

Cart ViewModel

fun connectToRealtime(scope: CoroutineScope) = viewModelScope.launch{

    try {

       // val realtimeChannel = realtime.channel("orders")

        val changeFlow = realtimeChannel.postgresChangeFlow<PostgresAction>(schema = "public") {
            table = "orders"
        }
        changeFlow.onEach {
            when (it) {
                is PostgresAction.Delete -> {
                    val updatedOrder = it.decodeOldRecord<Order>()
                    val updatedOrderList = _orderLines.value.orderList + updatedOrder
                    _orderLines.value = _orderLines.value.copy(
                        orderList = updatedOrderList
                    )

                    Log.d(TAG, "DeleteItem : ${it.oldRecord}")

                }

                is PostgresAction.Insert -> {
                    val updatedOrder = it.decodeRecord<Order>()
                    val updatedOrderList = _orderLines.value.orderList + updatedOrder
                    _orderLines.value = _orderLines.value.copy(
                        orderList = updatedOrderList
                    )

                    Log.d(TAG, "InsertItem : ${it.record}")
                }

                is PostgresAction.Select -> {
                    val updatedOrder = it.decodeRecord<Order>()
                    val updatedOrderList = _orderLines.value.orderList + updatedOrder
                    _orderLines.value = _orderLines.value.copy(
                        orderList = updatedOrderList
                    )

                    Log.d(TAG, "SelectedItem : ${it.record}")
                }

                is PostgresAction.Update -> {
                    val updatedOrder = it.decodeRecord<Order>()
                    val updatedOrderList = _orderLines.value.orderList + updatedOrder
                    _orderLines.value = _orderLines.value.copy(
                        orderList = updatedOrderList
                    )

                    Log.d(TAG, "UpdateItem : ${it.record}")
                }

            }
        }.launchIn(scope)

        realtimeChannel.subscribe()
    }catch (e: RestException){
        e.printStackTrace()
    }
}

Cart UI State

data class CartUiState(
    val orderList: List<Order> = listOf()
)

[jan-tennert]

fun connectToRealtime(scope: CoroutineScope) = viewModelScope.launch{

I wouldn't use the scope from LaunchedEffect, you are already in the viewModelScope so you can use it. You can also take a look at a new API which doesn't require channels: https://supabase.com/docs/reference/kotlin/stream

[Premjit-Developer]

I forgot to add getting this error:

java.lang.IllegalStateException: You cannot call postgresChangeFlow after joining the channel
               	at com.foodz.ui.feature.home.cart.CartVM$connectToRealtime$1.invokeSuspend(CartVM.kt:239)
               	at kotlin.coroutines.jvm.internal.BaseContinuationImpl.resumeWith(ContinuationImpl.kt:33)
               	at kotlinx.coroutines.internal.DispatchedContinuationKt.resumeCancellableWith(DispatchedContinuation.kt:363)
               	at kotlinx.coroutines.intrinsics.CancellableKt.startCoroutineCancellable(Cancellable.kt:26)
               	at kotlinx.coroutines.intrinsics.CancellableKt.startCoroutineCancellable$default(Cancellable.kt:21)
               	at kotlinx.coroutines.CoroutineStart.invoke(CoroutineStart.kt:88)
               	at kotlinx.coroutines.AbstractCoroutine.start(AbstractCoroutine.kt:123)
               	at kotlinx.coroutines.BuildersKt__Builders_commonKt.launch(Builders.common.kt:52)
               	at kotlinx.coroutines.BuildersKt.launch(Unknown Source:1)
               	at kotlinx.coroutines.BuildersKt__Builders_commonKt.launch$default(Builders.common.kt:43)
               	at kotlinx.coroutines.BuildersKt.launch$default(Unknown Source:1)
               	at com.foodz.ui.feature.home.cart.CartVM.connectToRealtime(CartVM.kt:167)

[jan-tennert]

You might be calling subscribe before calling this method or reusing a subscribed channel. Can't see much in your snippet.

[Premjit-Developer]

Maybe this ViewModel implementation part can help to identify this error

@HiltViewModel
class CartVM @Inject constructor(
    val authRepository: AuthRepository,
    val repository: DatabaseRepository,
    val realtimeChannel: RealtimeChannel
) : ViewModel() {


    private val _orderLines = MutableStateFlow(CartUiState())
    val uiState: StateFlow<CartUiState> = _orderLines.asStateFlow()

    val userId = authRepository.isCurrentUser()?.id.orEmpty()

    init {
        getAllOrders()

    }

    fun getAllOrders() = viewModelScope.launch {
        repository.getAllOrders(userId).collect {
            when (it) {
                is ResultState.Loading -> {
                    _orderLines.value = _orderLines.value.copy(
                        isLoading = true,
                    )
                }

                is ResultState.Success -> {
                    _orderLines.value = _orderLines.value.copy(
                        orderList = it.data,
                        isLoading = false
                    )
                }

                is ResultState.Error -> {
                    _orderLines.value = _orderLines.value.copy(
                        isLoading = false,
                        isError = true
                    )
                }
            }
        }
    }
    
       fun connectToRealtime() = viewModelScope.launch {

        try {

            // val realtimeChannel = realtime.channel("orders")

            val changeFlow = realtimeChannel.postgresChangeFlow<PostgresAction>(schema = "public") {
                table = "orders"
            }
            changeFlow.onEach {
                when (it) {
                    is PostgresAction.Delete -> {
                        val updatedOrder = it.decodeOldRecord<Order>()
                        val updatedOrderList = _orderLines.value.orderList + updatedOrder
                        _orderLines.value = _orderLines.value.copy(
                            orderList = updatedOrderList
                        )

                        Log.d(TAG, "DeleteItem : ${it.oldRecord}")

                    }

                    is PostgresAction.Insert -> {
                        val updatedOrder = it.decodeRecord<Order>()
                        val updatedOrderList = _orderLines.value.orderList + updatedOrder
                        _orderLines.value = _orderLines.value.copy(
                            orderList = updatedOrderList
                        )

                        Log.d(TAG, "InsertItem : ${it.record}")
                    }

                    is PostgresAction.Select -> {
                        val updatedOrder = it.decodeRecord<Order>()
                        val updatedOrderList = _orderLines.value.orderList + updatedOrder
                        _orderLines.value = _orderLines.value.copy(
                            orderList = updatedOrderList
                        )

                        Log.d(TAG, "SelectedItem : ${it.record}")
                    }

                    is PostgresAction.Update -> {
                        val updatedOrder = it.decodeRecord<Order>()
                        val updatedOrderList = _orderLines.value.orderList + updatedOrder
                        _orderLines.value = _orderLines.value.copy(
                            orderList = updatedOrderList
                        )

                        Log.d(TAG, "UpdateItem : ${it.record}")
                    }

                }
            }

            realtimeChannel.subscribe()
        } catch (e: RestException) {
            e.printStackTrace()
        }
    }
}

[Premjit-Developer]

Currently not possible with Realtime apis?

[jan-tennert]

Yes, realtime only supports listening to tables separately. You'll have to join them yourself.

[Premjit-Developer]

I'm saying that direct join like select() columns is it possible currently with realtime apis or in furture?

[jan-tennert]

No, that's not possible, and I don't know when or if they implement something like this, same with multiple filters.
See supabase/realtime#303.

[Premjit-Developer]

Oh..that means Supabase Realtime Engine currently doesn't support join queries. By the way, thanks for the replies.