-
Notifications
You must be signed in to change notification settings - Fork 30
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Create new implementation of PersistableSubscriptionStore (#34)
- Loading branch information
1 parent
4ea455b
commit a789796
Showing
7 changed files
with
281 additions
and
18 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
7 changes: 6 additions & 1 deletion
7
mqtt-client/src/main/java/com/gojek/mqtt/client/config/ExperimentConfigs.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -1,16 +1,21 @@ | ||
| package com.gojek.mqtt.client.config | ||
|
|
||
| import com.gojek.mqtt.client.config.SubscriptionStore.PERSISTABLE | ||
| import com.gojek.mqtt.constants.DEFAULT_ACTIVITY_CHECK_INTERVAL_SECS | ||
| import com.gojek.mqtt.constants.DEFAULT_INACTIVITY_TIMEOUT_SECS | ||
| import com.gojek.mqtt.constants.DEFAULT_POLICY_RESET_TIME_SECS | ||
| import com.gojek.mqtt.model.AdaptiveKeepAliveConfig | ||
|
|
||
| data class ExperimentConfigs( | ||
| val isPersistentSubscriptionStoreEnabled: Boolean = true, | ||
| val subscriptionStore: SubscriptionStore = PERSISTABLE, | ||
| val adaptiveKeepAliveConfig: AdaptiveKeepAliveConfig? = null, | ||
| val activityCheckIntervalSeconds: Int = DEFAULT_ACTIVITY_CHECK_INTERVAL_SECS, | ||
| val inactivityTimeoutSeconds: Int = DEFAULT_INACTIVITY_TIMEOUT_SECS, | ||
| val policyResetTimeSeconds: Int = DEFAULT_POLICY_RESET_TIME_SECS, | ||
| val incomingMessagesTTLSecs: Long = 360, | ||
| val incomingMessagesCleanupIntervalSecs: Long = 60 | ||
| ) | ||
|
|
||
| enum class SubscriptionStore { | ||
| IN_MEMORY, PERSISTABLE, PERSISTABLE_V2 | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
92 changes: 92 additions & 0 deletions
92
mqtt-client/src/main/java/com/gojek/mqtt/subscription/PersistableSubscriptionStoreV2.kt
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,92 @@ | ||
| package com.gojek.mqtt.subscription | ||
|
|
||
| import android.content.Context | ||
| import androidx.annotation.VisibleForTesting | ||
| import com.gojek.courier.QoS | ||
| import com.gojek.courier.extensions.toImmutableMap | ||
| import com.gojek.courier.extensions.toImmutableSet | ||
|
|
||
| internal class PersistableSubscriptionStoreV2(context: Context) : SubscriptionStore { | ||
| private lateinit var state: State | ||
| private val persistence = Persistence(context) | ||
| private val listener = object : SubscriptionStoreListener { | ||
| override fun onTopicsUnsubscribed(topics: Set<String>) { | ||
| onTopicsUnsubscribedInternal(topics) | ||
| } | ||
| } | ||
|
|
||
| private data class State( | ||
| val subscriptionTopics: Map<String, QoS>, | ||
| val pendingUnsubscribeTopics: Set<String> | ||
| ) | ||
|
|
||
| init { | ||
| restoreState() | ||
| } | ||
|
|
||
| @Synchronized | ||
| override fun getSubscribeTopics(): Map<String, QoS> { | ||
| return state.subscriptionTopics.toImmutableMap() | ||
| } | ||
|
|
||
| @Synchronized | ||
| override fun getUnsubscribeTopics(cleanSession: Boolean): Set<String> { | ||
| if (cleanSession) { | ||
| persistence.put(PREF_KEY_PENDING_UNSUBSCRIBES, emptySet()) | ||
| state = state.copy(pendingUnsubscribeTopics = emptySet()) | ||
| } | ||
| return state.pendingUnsubscribeTopics.toImmutableSet() | ||
| } | ||
|
|
||
| @Synchronized | ||
| override fun subscribeTopics(topicMap: Map<String, QoS>): Map<String, QoS> { | ||
| state = state.copy( | ||
| subscriptionTopics = state.subscriptionTopics + topicMap, | ||
| pendingUnsubscribeTopics = state.pendingUnsubscribeTopics - topicMap.keys | ||
| ) | ||
| persistence.put(PREF_KEY_PENDING_UNSUBSCRIBES, state.pendingUnsubscribeTopics) | ||
| return topicMap | ||
| } | ||
|
|
||
| @Synchronized | ||
| override fun unsubscribeTopics(topics: List<String>): Set<String> { | ||
| state = state.copy( | ||
| subscriptionTopics = state.subscriptionTopics - topics, | ||
| pendingUnsubscribeTopics = state.pendingUnsubscribeTopics + topics | ||
| ) | ||
| persistence.put(PREF_KEY_PENDING_UNSUBSCRIBES, state.pendingUnsubscribeTopics) | ||
| return topics.toSet() | ||
| } | ||
|
|
||
| override fun getListener(): SubscriptionStoreListener { | ||
| return listener | ||
| } | ||
|
|
||
| @Synchronized | ||
| override fun clear() { | ||
| persistence.put(PREF_KEY_PENDING_UNSUBSCRIBES, emptySet()) | ||
| state = State( | ||
| subscriptionTopics = emptyMap(), | ||
| pendingUnsubscribeTopics = emptySet() | ||
| ) | ||
| } | ||
|
|
||
| @VisibleForTesting | ||
| internal fun restoreState() { | ||
| state = State( | ||
| subscriptionTopics = emptyMap(), | ||
| pendingUnsubscribeTopics = persistence.get(PREF_KEY_PENDING_UNSUBSCRIBES, emptySet()) | ||
| ) | ||
| } | ||
|
|
||
| @Synchronized | ||
| private fun onTopicsUnsubscribedInternal(topics: Set<String>) { | ||
| state = state.copy( | ||
| subscriptionTopics = state.subscriptionTopics, | ||
| pendingUnsubscribeTopics = state.pendingUnsubscribeTopics - topics | ||
| ) | ||
| persistence.put(PREF_KEY_PENDING_UNSUBSCRIBES, state.pendingUnsubscribeTopics) | ||
| } | ||
| } | ||
|
|
||
| private const val PREF_KEY_PENDING_UNSUBSCRIBES = "PendingUnsubscribes" |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.