Fixed Coroutine Cancellation Exception for concurrent close and offer

Author: Daeda88

firebase-auth/src/androidMain/kotlin/dev/gitlive/firebase/auth/auth.kt

@@ -26,14 +26,14 @@ actual class FirebaseAuth internal constructor(val android: com.google.firebase.
         get() = android.currentUser?.let { FirebaseUser(it) }
 
     actual val authStateChanged get() = callbackFlow {
-        val listener = AuthStateListener { auth -> offer(auth.currentUser?.let { FirebaseUser(it) }) }
+        val listener = AuthStateListener { auth -> if (!isClosedForSend) offer(auth.currentUser?.let { FirebaseUser(it) }) }
         android.addAuthStateListener(listener)
         awaitClose { android.removeAuthStateListener(listener) }
     }
 
     actual val idTokenChanged: Flow<FirebaseUser?>
         get() = callbackFlow {
-            val listener = com.google.firebase.auth.FirebaseAuth.IdTokenListener { auth -> offer(auth.currentUser?.let { FirebaseUser(it) })}
+            val listener = com.google.firebase.auth.FirebaseAuth.IdTokenListener { auth -> if (!isClosedForSend) offer(auth.currentUser?.let { FirebaseUser(it) })}
             android.addIdTokenListener(listener)
             awaitClose { android.removeIdTokenListener(listener) }
         }

firebase-auth/src/iosMain/kotlin/dev/gitlive/firebase/auth/auth.kt

@@ -27,12 +27,12 @@ actual class FirebaseAuth internal constructor(val ios: FIRAuth) {
         get() = ios.currentUser?.let { FirebaseUser(it) }
 
     actual val authStateChanged get() = callbackFlow {
-        val handle = ios.addAuthStateDidChangeListener { _, user -> offer(user?.let { FirebaseUser(it) }) }
+        val handle = ios.addAuthStateDidChangeListener { _, user -> if (!isClosedForSend) offer(user?.let { FirebaseUser(it) }) }
         awaitClose { ios.removeAuthStateDidChangeListener(handle) }
     }
 
     actual val idTokenChanged get() = callbackFlow {
-        val handle = ios.addIDTokenDidChangeListener { _, user -> offer(user?.let { FirebaseUser(it) }) }
+        val handle = ios.addIDTokenDidChangeListener { _, user -> if (!isClosedForSend) offer(user?.let { FirebaseUser(it) }) }
         awaitClose { ios.removeIDTokenDidChangeListener(handle) }
     }
 

firebase-auth/src/jsMain/kotlin/dev/gitlive/firebase/auth/auth.kt

@@ -22,14 +22,16 @@ actual class FirebaseAuth internal constructor(val js: firebase.auth.Auth) {
 
     actual val authStateChanged get() = callbackFlow {
         val unsubscribe = js.onAuthStateChanged {
-            offer(it?.let { FirebaseUser(it) })
+            if (!isClosedForSend)
+                offer(it?.let { FirebaseUser(it) })
         }
         awaitClose { unsubscribe() }
     }
 
     actual val idTokenChanged get() = callbackFlow {
         val unsubscribe = js.onIdTokenChanged {
-            offer(it?.let { FirebaseUser(it) })
+            if (!isClosedForSend)
+                offer(it?.let { FirebaseUser(it) })
         }
         awaitClose { unsubscribe() }
     }

firebase-database/src/androidMain/kotlin/dev/gitlive/firebase/database/database.kt

@@ -91,7 +91,8 @@ actual open class Query internal constructor(
         get() = callbackFlow {
         val listener = object : ValueEventListener {
             override fun onDataChange(snapshot: com.google.firebase.database.DataSnapshot) {
-                offer(DataSnapshot(snapshot))
+                if (!isClosedForSend)
+                    offer(DataSnapshot(snapshot))
             }
 
             override fun onCancelled(error: com.google.firebase.database.DatabaseError) {
@@ -107,22 +108,22 @@ actual open class Query internal constructor(
 
             val moved by lazy { types.contains(Type.MOVED) }
             override fun onChildMoved(snapshot: com.google.firebase.database.DataSnapshot, previousChildName: String?) {
-                if(moved) offer(ChildEvent(DataSnapshot(snapshot), Type.MOVED, previousChildName))
+                if(moved && !isClosedForSend) offer(ChildEvent(DataSnapshot(snapshot), Type.MOVED, previousChildName))
             }
 
             val changed by lazy { types.contains(Type.CHANGED) }
             override fun onChildChanged(snapshot: com.google.firebase.database.DataSnapshot, previousChildName: String?) {
-                if(changed) offer(ChildEvent(DataSnapshot(snapshot), Type.CHANGED, previousChildName))
+                if(changed && !isClosedForSend) offer(ChildEvent(DataSnapshot(snapshot), Type.CHANGED, previousChildName))
             }
 
             val added by lazy { types.contains(Type.ADDED) }
             override fun onChildAdded(snapshot: com.google.firebase.database.DataSnapshot, previousChildName: String?) {
-                if(added) offer(ChildEvent(DataSnapshot(snapshot), Type.ADDED, previousChildName))
+                if(added && !isClosedForSend) offer(ChildEvent(DataSnapshot(snapshot), Type.ADDED, previousChildName))
             }
 
             val removed by lazy { types.contains(Type.REMOVED) }
             override fun onChildRemoved(snapshot: com.google.firebase.database.DataSnapshot) {
-                if(removed) offer(ChildEvent(DataSnapshot(snapshot), Type.REMOVED, null))
+                if(removed && !isClosedForSend) offer(ChildEvent(DataSnapshot(snapshot), Type.REMOVED, null))
             }
 
             override fun onCancelled(error: com.google.firebase.database.DatabaseError) {

firebase-database/src/iosMain/kotlin/dev/gitlive/firebase/database/database.kt

@@ -78,7 +78,7 @@ actual open class Query internal constructor(
     actual val valueEvents get() = callbackFlow {
         val handle = ios.observeEventType(
             FIRDataEventTypeValue,
-            withBlock = { offer(DataSnapshot(it!!)) }
+            withBlock = { if (!isClosedForSend) offer(DataSnapshot(it!!)) }
         ) { close(DatabaseException(it.toString())) }
         awaitClose { ios.removeObserverWithHandle(handle) }
     }
@@ -87,7 +87,7 @@ actual open class Query internal constructor(
         val handles = types.map { type ->
             ios.observeEventType(
                 type.toEventType(),
-                andPreviousSiblingKeyWithBlock = { it, key -> offer(ChildEvent(DataSnapshot(it!!), type, key)) }
+                andPreviousSiblingKeyWithBlock = { it, key -> if (!isClosedForSend) offer(ChildEvent(DataSnapshot(it!!), type, key)) }
             ) { close(DatabaseException(it.toString())) }
         }
         awaitClose {

firebase-database/src/jsMain/kotlin/dev/gitlive/firebase/database/database.kt

@@ -44,7 +44,7 @@ actual open class Query internal constructor(open val js: firebase.database.Quer
         val listener = rethrow {
             js.on(
                 "value",
-                { it, _ -> offer(DataSnapshot(it)) },
+                { it, _ -> if (!isClosedForSend) offer(DataSnapshot(it)) },
                 { close(DatabaseException(it)).run { Unit } }
             )
         }
@@ -58,7 +58,7 @@ actual open class Query internal constructor(open val js: firebase.database.Quer
                     eventType to js.on(
                         eventType,
                         { snapshot, previousChildName ->
-                            offer(
+                            if (!isClosedForSend) offer(
                                 ChildEvent(
                                     DataSnapshot(snapshot),
                                     type,

firebase-firestore/src/jsMain/kotlin/dev/gitlive/firebase/firestore/firestore.kt

@@ -238,7 +238,7 @@ actual class DocumentReference(val js: firebase.firestore.DocumentReference) {
 
     actual val snapshots get() = callbackFlow {
         val unsubscribe = js.onSnapshot(
-            { offer(DocumentSnapshot(it)) },
+            { if (!isClosedForSend) offer(DocumentSnapshot(it)) },
             { close(errorToException(it)) }
         )
         awaitClose { unsubscribe() }
@@ -289,7 +289,7 @@ actual open class Query(open val js: firebase.firestore.Query) {
     actual val snapshots get() = callbackFlow {
         val unsubscribe = rethrow {
             js.onSnapshot(
-                { offer(QuerySnapshot(it)) },
+                { if (!isClosedForSend) offer(QuerySnapshot(it)) },
                 { close(errorToException(it)) }
             )
         }