Skip to content

Commit a6e92e0

Browse files
committed
Squashed changes from ExponentialBackoffResetOnNetworkRestored, which includes #8421 (add logic to integrate network state changes with exponential backoff logic) and #8423 (Use LongAdder to count concurrent invocations instead of AtomicInteger), also #8425 (readability improvements to DataConnectBidiConnectStream.kt)
1 parent 40f7541 commit a6e92e0

18 files changed

Lines changed: 2805 additions & 46 deletions

File tree

firebase-dataconnect/CHANGELOG.md

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,8 @@
22

33
- [changed] Realtime query subscriptions now retry connecting using an
44
exponential backoff strategy.
5-
([#8381](https://github.com/firebase/firebase-android-sdk/pull/8381))
5+
([#8381](https://github.com/firebase/firebase-android-sdk/pull/8381),
6+
[#8421](https://github.com/firebase/firebase-android-sdk/pull/8421))
67

78
# 17.3.2
89

firebase-dataconnect/src/main/kotlin/com/google/firebase/dataconnect/core/DataConnectBidiConnectStream.kt

Lines changed: 68 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -21,6 +21,7 @@ import com.google.firebase.dataconnect.core.DataConnectAuth.AuthUid
2121
import com.google.firebase.dataconnect.core.DataConnectAuth.GetAuthTokenResult
2222
import com.google.firebase.dataconnect.core.DataConnectGrpcMetadata.Companion.FIREBASE_AUTH_TOKEN_HEADER
2323
import com.google.firebase.dataconnect.core.LoggerGlobals.debug
24+
import com.google.firebase.dataconnect.core.LoggerGlobals.warn
2425
import com.google.firebase.dataconnect.util.CoroutineUtils.completedFlow
2526
import com.google.firebase.dataconnect.util.CoroutineUtils.mergeColdAndHotFlow
2627
import com.google.firebase.dataconnect.util.GrpcBidiFlow
@@ -40,6 +41,7 @@ import google.firebase.dataconnect.proto.StreamRequest as StreamRequestProto
4041
import google.firebase.dataconnect.proto.StreamResponse as StreamResponseProto
4142
import java.util.concurrent.atomic.AtomicReference
4243
import kotlin.time.Duration.Companion.milliseconds
44+
import kotlinx.coroutines.CoroutineName
4345
import kotlinx.coroutines.CoroutineScope
4446
import kotlinx.coroutines.CoroutineStart
4547
import kotlinx.coroutines.Job
@@ -48,6 +50,7 @@ import kotlinx.coroutines.channels.BufferOverflow
4850
import kotlinx.coroutines.channels.Channel
4951
import kotlinx.coroutines.channels.SendChannel
5052
import kotlinx.coroutines.channels.onFailure
53+
import kotlinx.coroutines.coroutineScope
5154
import kotlinx.coroutines.currentCoroutineContext
5255
import kotlinx.coroutines.ensureActive
5356
import kotlinx.coroutines.flow.Flow
@@ -56,7 +59,9 @@ import kotlinx.coroutines.flow.MutableStateFlow
5659
import kotlinx.coroutines.flow.SharingStarted
5760
import kotlinx.coroutines.flow.asSharedFlow
5861
import kotlinx.coroutines.flow.buffer
62+
import kotlinx.coroutines.flow.emitAll
5963
import kotlinx.coroutines.flow.filter
64+
import kotlinx.coroutines.flow.flow
6065
import kotlinx.coroutines.flow.map
6166
import kotlinx.coroutines.flow.mapNotNull
6267
import kotlinx.coroutines.flow.merge
@@ -98,6 +103,7 @@ internal class DataConnectBidiConnectStream(
98103
>,
99104
authToken: Flow<SequencedReference<GetAuthTokenResult?>>,
100105
shouldRetry: suspend (Throwable) -> RetryStrategy,
106+
networkConnectivityRestoredFlow: Flow<NetworkConnectivityRestored>,
101107
idStringGenerator: IdStringGenerator,
102108
private val grpcMetadata: DataConnectGrpcMetadata,
103109
private val coroutineScope: CoroutineScope,
@@ -131,7 +137,7 @@ internal class DataConnectBidiConnectStream(
131137
)
132138
val connectionStateUpdater = ConnectionStateUpdater(idStringGenerator)
133139

134-
val sharedFlow =
140+
val physicalConnectionFlow =
135141
mergeGrpcAndAuth(flow, authToken)
136142
.onStart { resetAndRetryEvent.clear() }
137143
.mapNotNull { event ->
@@ -157,36 +163,74 @@ internal class DataConnectBidiConnectStream(
157163
}
158164
throw throwable ?: Exception("to be handled by retryWhen")
159165
}
160-
.retryWhen { cause, _ ->
161-
resetAndRetryEvent.clear()
162-
val retryStrategy =
163-
try {
164-
shouldRetry(cause)
165-
} catch (e: Throwable) {
166-
currentCoroutineContext().ensureActive()
167-
_permanentFailureFlow.emit(e)
168-
throw e
169-
}
170166

171-
when (retryStrategy) {
172-
RetryStrategy.RETRY_IMMEDIATELY -> {
173-
retryBackoff.reset()
174-
true
167+
val logicalConnectionFlow =
168+
physicalConnectionFlow.retryWhen { cause, _ ->
169+
resetAndRetryEvent.clear()
170+
val retryStrategy =
171+
try {
172+
shouldRetry(cause)
173+
} catch (e: Throwable) {
174+
currentCoroutineContext().ensureActive()
175+
_permanentFailureFlow.emit(e)
176+
throw e
177+
}
178+
179+
when (retryStrategy) {
180+
RetryStrategy.RETRY_IMMEDIATELY -> {
181+
retryBackoff.reset()
182+
true
183+
}
184+
RetryStrategy.RETRY_AFTER_BACKOFF -> {
185+
val backoffMs = retryBackoff.next()
186+
logger.debug {
187+
"waiting ${backoffMs}ms before retrying connection, which failed due to $cause"
175188
}
176-
RetryStrategy.RETRY_AFTER_BACKOFF -> {
177-
val backoffMs = retryBackoff.next()
178-
logger.debug {
179-
"waiting ${backoffMs}ms before retrying connection, which failed due to $cause"
189+
withTimeoutOrNull(backoffMs.milliseconds) {
190+
onRetryBackoffForTesting.get()?.invoke(backoffMs)
191+
resetAndRetryEvent.await()
192+
}
193+
resetAndRetryEvent.clear()
194+
true
195+
}
196+
}
197+
}
198+
199+
val logicalConnectionWithNetworkMonitoringFlow = flow {
200+
coroutineScope {
201+
// Monitor for network connectivity restored events concurrently with the downstream flow
202+
// collection so that retryWhen can eagerly retry the connection if connectivity is
203+
// potentially restored, rather than waiting for the entire backoff duration.
204+
val networkConnectivityRestoredFlowCollectJob =
205+
launch(CoroutineName("NetworkConnectivityRestoredFlowCollectJob")) {
206+
val collectResult = runCatching {
207+
networkConnectivityRestoredFlow.collect {
208+
retryBackoff.reset()
209+
resetAndRetryEvent.signal()
180210
}
181-
withTimeoutOrNull(backoffMs.milliseconds) {
182-
onRetryBackoffForTesting.get()?.invoke(backoffMs)
183-
resetAndRetryEvent.await()
211+
}
212+
213+
ensureActive()
214+
215+
collectResult.onFailure {
216+
logger.warn(it) {
217+
"WARNING: monitoring network connectivity failed; " +
218+
"automatic re-connection to the Data Connect server " +
219+
"may take longer than strictly necessary [qhrqw5xvxc]"
184220
}
185-
resetAndRetryEvent.clear()
186-
true
187221
}
188222
}
223+
224+
try {
225+
emitAll(logicalConnectionFlow)
226+
} finally {
227+
networkConnectivityRestoredFlowCollectJob.cancel()
189228
}
229+
}
230+
}
231+
232+
val sharedFlow =
233+
logicalConnectionWithNetworkMonitoringFlow
190234
.buffer(capacity = 64) // Use a finite buffer to activate gRPC flow control, when needed
191235
.shareIn(
192236
coroutineScope,

firebase-dataconnect/src/main/kotlin/com/google/firebase/dataconnect/core/DataConnectGrpcRPCs.kt

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -100,6 +100,7 @@ internal class DataConnectGrpcRPCs(
100100
@get:VisibleForTesting val grpcMetadata: DataConnectGrpcMetadata,
101101
private val cache: DataConnectCache?,
102102
parentLogger: Logger,
103+
private val networkConnectivityRestoredFlow: Flow<NetworkConnectivityRestored>,
103104
) {
104105
private val logger =
105106
Logger("DataConnectGrpcRPCs").apply {
@@ -525,6 +526,7 @@ internal class DataConnectGrpcRPCs(
525526
flow,
526527
tokenManager.authToken,
527528
shouldRetry = shouldRetry,
529+
networkConnectivityRestoredFlow,
528530
idStringGenerator,
529531
grpcMetadata,
530532
connectCoroutineScope,

firebase-dataconnect/src/main/kotlin/com/google/firebase/dataconnect/core/FirebaseDataConnectFactory.kt

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -41,6 +41,8 @@ internal class FirebaseDataConnectFactory(
4141
// all instances generate unique IDs.
4242
private val idStringGenerator = IdStringGenerator(Random.Default)
4343

44+
private val networkConnectivityRestoredFlow = networkConnectivityRestoredFlow(context)
45+
4446
init {
4547
firebaseApp.addLifecycleEventListener { _, _ -> close() }
4648
}
@@ -92,6 +94,7 @@ internal class FirebaseDataConnectFactory(
9294
creator = this@FirebaseDataConnectFactory,
9395
settings = settings ?: DataConnectSettings(),
9496
idStringGenerator = idStringGenerator,
97+
networkConnectivityRestoredFlow = networkConnectivityRestoredFlow,
9598
)
9699

97100
fun remove(instance: FirebaseDataConnect) {

firebase-dataconnect/src/main/kotlin/com/google/firebase/dataconnect/core/FirebaseDataConnectImpl.kt

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -53,6 +53,7 @@ import kotlinx.coroutines.GlobalScope
5353
import kotlinx.coroutines.asCoroutineDispatcher
5454
import kotlinx.coroutines.async
5555
import kotlinx.coroutines.cancel
56+
import kotlinx.coroutines.flow.Flow
5657
import kotlinx.coroutines.flow.MutableStateFlow
5758
import kotlinx.coroutines.flow.collect
5859
import kotlinx.coroutines.flow.update
@@ -96,6 +97,7 @@ internal class FirebaseDataConnectImpl(
9697
private val creator: FirebaseDataConnectFactory,
9798
override val settings: DataConnectSettings,
9899
override val idStringGenerator: IdStringGenerator,
100+
private val networkConnectivityRestoredFlow: Flow<NetworkConnectivityRestored>,
99101
) : FirebaseDataConnectInternal {
100102

101103
override val logger =
@@ -309,6 +311,7 @@ internal class FirebaseDataConnectImpl(
309311
grpcMetadata = grpcMetadata,
310312
cache = cache,
311313
parentLogger = logger,
314+
networkConnectivityRestoredFlow = networkConnectivityRestoredFlow,
312315
)
313316

314317
if (backendInfo.isEmulator) {
Lines changed: 176 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,176 @@
1+
/*
2+
* Copyright 2026 Google LLC
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package com.google.firebase.dataconnect.core
17+
18+
import android.content.Context
19+
import android.content.Context.CONNECTIVITY_SERVICE
20+
import android.net.ConnectivityManager
21+
import android.net.Network
22+
import android.net.NetworkCapabilities
23+
import android.net.NetworkCapabilities.NET_CAPABILITY_INTERNET
24+
import android.net.NetworkCapabilities.NET_CAPABILITY_VALIDATED
25+
import android.net.NetworkRequest
26+
import android.os.Build
27+
import androidx.annotation.RequiresApi
28+
import com.google.firebase.dataconnect.util.coroutines.ConflatedSignal
29+
import com.google.firebase.dataconnect.util.coroutines.signalIfNotNull
30+
import java.util.concurrent.locks.ReentrantLock
31+
import kotlin.concurrent.withLock
32+
import kotlinx.coroutines.flow.Flow
33+
import kotlinx.coroutines.flow.emitAll
34+
import kotlinx.coroutines.flow.flow
35+
36+
internal data object NetworkConnectivityRestored
37+
38+
/**
39+
* Creates a cold [Flow] of [NetworkConnectivityRestored] events that monitors network connectivity
40+
* changes that potentially result in a restoration of network connectivity.
41+
*
42+
* Emissions from this flow signify that the network connectivity status has changed in a way that
43+
* suggests a network connection might now be successfully established.
44+
*
45+
* A downstream connection retry backoff mechanism can collect this flow to interrupt or abort a
46+
* retry suspension (backoff delay). Instead of waiting for the full backoff timeout to expire, the
47+
* mechanism can immediately re-attempt the connection upon receiving a
48+
* [NetworkConnectivityRestored] event.
49+
*
50+
* If the [Flow] is collected while a network appears to be available then an event will be emitted
51+
* immediately. Namely, it will not wait for some other network to become available to emit the
52+
* first event if a network is _already_ available.
53+
*
54+
* @param context The [Context] used to retrieve the [ConnectivityManager] system service.
55+
* @return A cold [Flow] emitting [NetworkConnectivityRestored] on network state transitions that
56+
* suggest that network connectivity is now (or continues to be) available.
57+
*/
58+
internal fun networkConnectivityRestoredFlow(context: Context): Flow<NetworkConnectivityRestored> {
59+
return if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.N) {
60+
networkConnectivityRestoredFlowAPI24(context)
61+
} else {
62+
networkConnectivityRestoredFlowAPI23(context)
63+
}
64+
}
65+
66+
private fun networkConnectivityRestoredFlowAPI23(
67+
context: Context
68+
): Flow<NetworkConnectivityRestored> =
69+
networkConnectivityRestoredFlow(context) {
70+
val request = NetworkRequest.Builder().addCapability(NET_CAPABILITY_INTERNET).build()
71+
registerNetworkCallback(request, it)
72+
}
73+
74+
@RequiresApi(Build.VERSION_CODES.N)
75+
private fun networkConnectivityRestoredFlowAPI24(
76+
context: Context
77+
): Flow<NetworkConnectivityRestored> =
78+
networkConnectivityRestoredFlow(context) { registerDefaultNetworkCallback(it) }
79+
80+
private fun networkConnectivityRestoredFlow(
81+
context: Context,
82+
registerCallback: ConnectivityManager.(ConnectivityManager.NetworkCallback) -> Unit
83+
): Flow<NetworkConnectivityRestored> = flow {
84+
val connectivityManager = context.getSystemService(CONNECTIVITY_SERVICE)
85+
checkNotNull(connectivityManager) {
86+
"getSystemService(CONNECTIVITY_SERVICE) returned null; " +
87+
"try adding android.permission.ACCESS_NETWORK_STATE to AndroidManifest.xml [yxzng5zfyg]"
88+
}
89+
check(connectivityManager is ConnectivityManager) {
90+
"internal error rfq632nm3m: getSystemService(CONNECTIVITY_SERVICE) should have returned " +
91+
"an instance of ${ConnectivityManager::class.qualifiedName}, but got $connectivityManager"
92+
}
93+
94+
val callback = NetworkCallbackImpl()
95+
registerCallback(connectivityManager, callback)
96+
97+
try {
98+
emitAll(callback.signal.signals)
99+
} finally {
100+
connectivityManager.unregisterNetworkCallback(callback)
101+
}
102+
}
103+
104+
private class NetworkCallbackImpl : ConnectivityManager.NetworkCallback() {
105+
106+
val signal = ConflatedSignal<NetworkConnectivityRestored>()
107+
108+
private val lock = ReentrantLock()
109+
private val availableNetworks = mutableSetOf<Network>()
110+
private val validatedNetworks = mutableSetOf<Network>()
111+
112+
override fun onAvailable(network: Network) {
113+
val signalOrNull =
114+
lock.withLock {
115+
val networkAdded = availableNetworks.add(network)
116+
if (networkAdded) NetworkConnectivityRestored else null
117+
}
118+
119+
signal.signalIfNotNull(signalOrNull)
120+
}
121+
122+
override fun onLost(network: Network) {
123+
val signalOrNull =
124+
lock.withLock {
125+
availableNetworks.remove(network)
126+
val wasValidated = validatedNetworks.remove(network)
127+
// Only signal if we have other validated networks that can take over (API 23 failover)
128+
if (wasValidated && validatedNetworks.isNotEmpty()) NetworkConnectivityRestored else null
129+
}
130+
131+
signal.signalIfNotNull(signalOrNull)
132+
}
133+
134+
/**
135+
* Notifies the callback when the network block status changes (e.g. background data saver rules
136+
* are toggled). We signal when the network becomes unblocked (`blocked == false`) so consumers
137+
* can attempt to reconnect.
138+
*/
139+
@RequiresApi(Build.VERSION_CODES.Q)
140+
override fun onBlockedStatusChanged(network: Network, blocked: Boolean) {
141+
lock.withLock { availableNetworks.add(network) }
142+
if (!blocked) {
143+
signal.signal(NetworkConnectivityRestored)
144+
}
145+
}
146+
147+
/**
148+
* Handles capability changes for a network.
149+
*
150+
* On Android 9 (API 28) and above, this callback is invoked frequently for minor capability
151+
* updates such as signal strength (RSSI) fluctuations or bandwidth estimate changes, even if the
152+
* connection remains active and stable.
153+
*
154+
* To prevent excessive signals and unnecessary reconnection attempts by downstream consumers,
155+
* this method filters out those redundant updates and only signals when the network's validation
156+
* status (presence of [NET_CAPABILITY_VALIDATED]) actually transitions (either gained or lost).
157+
*/
158+
override fun onCapabilitiesChanged(network: Network, networkCapabilities: NetworkCapabilities) {
159+
val isValidated = networkCapabilities.hasCapability(NET_CAPABILITY_VALIDATED)
160+
161+
val signalOrNull =
162+
lock.withLock {
163+
availableNetworks.add(network)
164+
if (isValidated) {
165+
val networkAdded = validatedNetworks.add(network)
166+
if (networkAdded) NetworkConnectivityRestored else null
167+
} else {
168+
val networkRemoved = validatedNetworks.remove(network)
169+
if (networkRemoved && validatedNetworks.isNotEmpty()) NetworkConnectivityRestored
170+
else null
171+
}
172+
}
173+
174+
signal.signalIfNotNull(signalOrNull)
175+
}
176+
}

0 commit comments

Comments
 (0)