FirestoreUserEventDataSource.kt

/* Copyright 2019 Google LLC.	
   SPDX-License-Identifier: Apache-2.0 */
override fun getObservableUserEvent(userId: String, eventId: SessionId): Flow<UserEventResult> {

    // 1) Create Flow with channelFlow
    return channelFlow<UserEventResult> {

        val eventDocument = firestore.collection(USERS_COLLECTION)
            .document(userId)
            .collection(EVENTS_COLLECTION)
            .document(eventId)
    
        // 1) Register callback to the API
        val subscription = eventDocument.addSnapshotListener { snapshot, _ ->
            val userEvent = if (snapshot.exists()) {
                parseUserEvent(snapshot)
            } else { null }

            // 2) Send items to the Flow
            channel.offer(UserEventResult(userEvent))
        }

        // 3) Don't close the stream of data, keep it open until the consumer
        // stops listening or the API calls onCompleted or onError. 
        // When that happens, cancel the subscription to the 3P library
        awaitClose { subscription.remove() }
    }
}