Rewatch feeds on socket reconnection #61
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,88 @@ | ||
| /* | ||
| * Copyright (c) 2014-2025 Stream.io Inc. All rights reserved. | ||
| * | ||
| * Licensed under the Stream License; | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * https://github.com/GetStream/stream-feeds-android/blob/main/LICENSE | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
| package io.getstream.feeds.android.client.internal.client.reconnect | ||
|
|
||
| import io.getstream.android.core.api.model.StreamRetryPolicy | ||
| import io.getstream.android.core.api.model.connection.StreamConnectionState | ||
| import io.getstream.android.core.api.model.exceptions.StreamClientException | ||
| import io.getstream.android.core.api.processing.StreamRetryProcessor | ||
| import io.getstream.feeds.android.client.api.model.FeedId | ||
| import io.getstream.feeds.android.client.api.state.query.FeedQuery | ||
| import io.getstream.feeds.android.client.internal.repository.FeedsRepository | ||
| import io.getstream.feeds.android.client.internal.repository.GetOrCreateInfo | ||
| import java.util.concurrent.ConcurrentHashMap | ||
| import kotlinx.coroutines.CoroutineScope | ||
| import kotlinx.coroutines.async | ||
| import kotlinx.coroutines.coroutineScope | ||
| import kotlinx.coroutines.flow.Flow | ||
| import kotlinx.coroutines.flow.MutableSharedFlow | ||
| import kotlinx.coroutines.flow.filterIsInstance | ||
| import kotlinx.coroutines.launch | ||
|
|
||
| /** | ||
| * Handles re-watching feeds upon reconnection. | ||
| * | ||
| * Keeps track of feeds that are being watched and re-subscribes to them when the connection is | ||
| * re-established. | ||
| * | ||
| * @property connectionState A [Flow] that emits events when the connection state changes to | ||
| * [StreamConnectionState.Connected]. | ||
| * @property feedsRepository The [FeedsRepository] used to rewatch feeds on connection. | ||
| * @property scope The [CoroutineScope] in which to launch coroutines for re-watching feeds. | ||
| */ | ||
| internal class FeedWatchHandler( | ||
| private val connectionState: Flow<StreamConnectionState>, | ||
|
There was a problem hiding this comment. Question: Maybe its better for the There was a problem hiding this comment. To do that, we'd need to add a There was a problem hiding this comment. The |
||
| private val feedsRepository: FeedsRepository, | ||
| private val retryProcessor: StreamRetryProcessor, | ||
| private val errorBus: MutableSharedFlow<StreamClientException>, | ||
| private val scope: CoroutineScope, | ||
| ) { | ||
| private val watched = ConcurrentHashMap<FeedId, Unit>() | ||
|
|
||
| init { | ||
| scope.launch { | ||
| connectionState.filterIsInstance<StreamConnectionState.Connected>().collect { | ||
| rewatchAll() | ||
|
There was a problem hiding this comment.
There was a problem hiding this comment. I know but what's the alternative? I'm going to change it to |
||
| } | ||
| } | ||
| } | ||
|
|
||
| fun onStartWatching(feedId: FeedId) { | ||
|
There was a problem hiding this comment. Nit:maybe There was a problem hiding this comment. I put |
||
| watched[feedId] = Unit | ||
| } | ||
|
|
||
| fun onStopWatching(feedId: FeedId) { | ||
|
There was a problem hiding this comment. Nit: same as the other |
||
| watched -= feedId | ||
| } | ||
|
|
||
| private suspend fun rewatchAll() { | ||
| coroutineScope { watched.keys.map { id -> async { rewatch(id) } } } | ||
| } | ||
|
|
||
| private suspend fun rewatch(id: FeedId): Result<GetOrCreateInfo> = | ||
| retryProcessor | ||
| .retry(retryPolicy) { | ||
| feedsRepository.getOrCreateFeed(FeedQuery(id, watch = true)).getOrThrow() | ||
| } | ||
| .onFailure { errorBus.emit(StreamFeedRewatchException(id)) } | ||
|
|
||
| companion object { | ||
| private val retryPolicy = StreamRetryPolicy.exponential(maxRetries = 3) | ||
| } | ||
| } | ||
|
|
||
| internal class StreamFeedRewatchException(val id: FeedId, cause: Throwable? = null) : | ||
| StreamClientException("Failed to rewatch feed", cause) | ||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Isn't it better to just call
rewatchAllhere in case we have non empty list of watched feeds. Since ifconnectgoes trough in thecoreClientwe will definitely have a new connection ID which will require re-watching??There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This is actually a change that is not directly tied to the rewatch logic. I discovered that we lost this call when migrating to the core client, so we were never reconnecting the socket automatically.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I am not concerned about
connectionRecoveryHandler.start()it should be there.My point is that right after that we can call
rewatchAllinstead of initing a collector in the watch handler.