Cloud DB live data changes

get_user_data_changes.kt

    @ExperimentalCoroutinesApi
    suspend fun getUserDataChanges(id : String?) : Flow<Resource<User>> = withContext(ioDispatcher) {
        callbackFlow {

            if (id == null) {
                offer(Resource.Error(Exception("Id must not be null")))
                return@callbackFlow
            }

            // 1- Create query
            val query: CloudDBZoneQuery<User> = CloudDBZoneQuery.where(User::class.java).equalTo("accountId", id)
            // 2 - Register query
            val subscription = cloudDBZone.subscribeSnapshot(query, CloudDBZoneQuery.CloudDBZoneQueryPolicy.POLICY_QUERY_FROM_CLOUD_PRIOR, object : OnSnapshotListener<User> {
                override fun onSnapshot(snapShot: CloudDBZoneSnapshot<User>?, error: AGConnectCloudDBException?) {
                    // Check error
                    if (error != null) {
                        error.printStackTrace()
                        offer(Resource.Error(error))
                        return
                    }
                    // Check data
                    try {
                        val snapShotObjects = snapShot?.snapshotObjects
                        // Get user data from db
                        if (snapShotObjects != null) {

                            // Check item in db exist
                            if (snapShotObjects.size() == 0) {
                                offer(Resource.Error(Exception("User not exists in Cloud DB!")))
                                return
                            }

                            while (snapShotObjects.hasNext()) {
                                val user : User = snapShotObjects.next()
                                offer(Resource.Success(user))
                            }
                        }
                    } catch (e : Exception) {
                        e.printStackTrace()
                        offer(Resource.Error(e))
                    } finally {
                        snapShot?.release()
                    }
                }
            })

            // 3 - Remove subscription
            awaitClose {
                subscription.remove()
            }
        }
    }