TheArchive/session-android
1
0
Fork 0
mirror of https://github.com/oxen-io/session-android.git synced 2024-11-23 18:15:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'ry/poller-fix' into on-3

Browse Source
This commit is contained in:
bemusementpark 2024-07-05 16:55:18 +09:30
commit fa5b10e2e1
4 changed files with 139 additions and 57 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
app/src/main/java/org/thoughtcrime/securesms/ApplicationContext.java
View File

@ -453,6 +453,13 @@ public class ApplicationContext extends Application implements DefaultLifecycleO
ClosedGroupPollerV2.getShared().start();
}
public void retrieveUserProfile() {
setUpPollingIfNeeded();
if (poller != null) {
poller.retrieveUserProfile();
}
}
private void resubmitProfilePictureIfNeeded() {
// Files expire on the file server after a while, so we simply re-upload the user's profile picture
// at a certain interval to ensure it's always available.

2
app/src/main/java/org/thoughtcrime/securesms/onboarding/manager/LoadAccountManager.kt
View File

@ -52,7 +52,7 @@ class LoadAccountManager @Inject constructor(
setHasViewedSeed(true)
}
ApplicationContext.getInstance(context).apply { startPollingIfNeeded() }
ApplicationContext.getInstance(context).retrieveUserProfile()
}
}
}

185
libsession/src/main/java/org/session/libsession/messaging/sending_receiving/pollers/Poller.kt
View File

@ -62,6 +62,16 @@ class Poller(private val configFactory: ConfigFactoryProtocol, debounceTimer: Ti
hasStarted = false
usedSnodes.clear()
}
fun retrieveUserProfile() {
Log.d("Loki", "Retrieving user profile.")
SnodeAPI.getSwarm(userPublicKey).bind {
usedSnodes.clear()
deferred<Unit, Exception>().also {
pollNextSnode(userProfileOnly = true, it)
}.promise
}
}
// endregion
// region Private API
@ -71,7 +81,7 @@ class Poller(private val configFactory: ConfigFactoryProtocol, debounceTimer: Ti
SnodeAPI.getSwarm(userPublicKey).bind {
usedSnodes.clear()
val deferred = deferred<Unit, Exception>()
pollNextSnode(deferred)
pollNextSnode(deferred = deferred)
deferred.promise
}.success {
val nextDelay = if (isCaughtUp) retryInterval else 0
@ -90,7 +100,7 @@ class Poller(private val configFactory: ConfigFactoryProtocol, debounceTimer: Ti
}
}
private fun pollNextSnode(deferred: Deferred<Unit, Exception>) {
private fun pollNextSnode(userProfileOnly: Boolean = false, deferred: Deferred<Unit, Exception>) {
val swarm = SnodeModule.shared.storage.getSwarm(userPublicKey) ?: setOf()
val unusedSnodes = swarm.subtract(usedSnodes)
if (unusedSnodes.isNotEmpty()) {
@ -98,13 +108,13 @@ class Poller(private val configFactory: ConfigFactoryProtocol, debounceTimer: Ti
val nextSnode = unusedSnodes.elementAt(index)
usedSnodes.add(nextSnode)
Log.d("Loki", "Polling $nextSnode.")
poll(nextSnode, deferred).fail { exception ->
poll(userProfileOnly, nextSnode, deferred).fail { exception ->
if (exception is PromiseCanceledException) {
Log.d("Loki", "Polling $nextSnode canceled.")
} else {
Log.d("Loki", "Polling $nextSnode failed; dropping it and switching to next snode.")
SnodeAPI.dropSnodeFromSwarmIfNeeded(nextSnode, userPublicKey)
pollNextSnode(deferred)
pollNextSnode(userProfileOnly, deferred)
}
}
} else {
@ -158,6 +168,65 @@ class Poller(private val configFactory: ConfigFactoryProtocol, debounceTimer: Ti
}
}
private fun poll(userProfileOnly: Boolean, snode: Snode, deferred: Deferred<Unit, Exception>): Promise<Unit, Exception> {
if (userProfileOnly) {
return pollUserProfile(snode, deferred)
}
return poll(snode, deferred)
}
private fun pollUserProfile(snode: Snode, deferred: Deferred<Unit, Exception>): Promise<Unit, Exception> = task {
runBlocking(Dispatchers.IO) {
val requests = mutableListOf<SnodeAPI.SnodeBatchRequestInfo>()
val hashesToExtend = mutableSetOf<String>()
configFactory.user?.let { config ->
hashesToExtend += config.currentHashes()
SnodeAPI.buildAuthenticatedRetrieveBatchRequest(
snode, userPublicKey,
config.configNamespace(),
maxSize = -8
)
}?.let { request ->
requests += request
}
if (hashesToExtend.isNotEmpty()) {
SnodeAPI.buildAuthenticatedAlterTtlBatchRequest(
messageHashes = hashesToExtend.toList(),
publicKey = userPublicKey,
newExpiry = SnodeAPI.nowWithOffset + 14.days.inWholeMilliseconds,
extend = true
)?.let { extensionRequest ->
requests += extensionRequest
}
}
if (requests.isNotEmpty()) {
SnodeAPI.getRawBatchResponse(snode, userPublicKey, requests).bind { rawResponses ->
isCaughtUp = true
if (!deferred.promise.isDone()) {
val responseList = (rawResponses["results"] as List<RawResponse>)
responseList.getOrNull(0)?.let { rawResponse ->
if (rawResponse["code"] as? Int != 200) {
Log.e("Loki", "Batch sub-request had non-200 response code, returned code ${(rawResponse["code"] as? Int) ?: "[unknown]"}")
} else {
val body = rawResponse["body"] as? RawResponse
if (body == null) {
Log.e("Loki", "Batch sub-request didn't contain a body")
} else {
processConfig(snode, body, configFactory.user!!.configNamespace(), configFactory.user)
}
}
}
}
Promise.ofSuccess(Unit)
}.fail {
Log.e("Loki", "Failed to get raw batch response", it)
}
}
}
}
private fun poll(snode: Snode, deferred: Deferred<Unit, Exception>): Promise<Unit, Exception> {
if (!hasStarted) { return Promise.ofFail(PromiseCanceledException()) }
return task {
@ -196,67 +265,73 @@ class Poller(private val configFactory: ConfigFactoryProtocol, debounceTimer: Ti
}
}
SnodeAPI.getRawBatchResponse(snode, userPublicKey, requests).bind { rawResponses ->
isCaughtUp = true
if (deferred.promise.isDone()) {
return@bind Promise.ofSuccess(Unit)
} else {
val responseList = (rawResponses["results"] as List<RawResponse>)
// in case we had null configs, the array won't be fully populated
// index of the sparse array key iterator should be the request index, with the key being the namespace
listOfNotNull(
configFactory.user?.configNamespace(),
configFactory.contacts?.configNamespace(),
configFactory.userGroups?.configNamespace(),
configFactory.convoVolatile?.configNamespace()
).map {
it to requestSparseArray.indexOfKey(it)
}.filter { (_, i) -> i >= 0 }.forEach { (key, requestIndex) ->
responseList.getOrNull(requestIndex)?.let { rawResponse ->
if (rawResponse["code"] as? Int != 200) {
Log.e("Loki", "Batch sub-request had non-200 response code, returned code ${(rawResponse["code"] as? Int) ?: "[unknown]"}")
return@forEach
}
val body = rawResponse["body"] as? RawResponse
if (body == null) {
Log.e("Loki", "Batch sub-request didn't contain a body")
return@forEach
}
if (key == Namespace.DEFAULT) {
return@forEach // continue, skip default namespace
} else {
when (ConfigBase.kindFor(key)) {
UserProfile::class.java -> processConfig(snode, body, key, configFactory.user)
Contacts::class.java -> processConfig(snode, body, key, configFactory.contacts)
ConversationVolatileConfig::class.java -> processConfig(snode, body, key, configFactory.convoVolatile)
UserGroupsConfig::class.java -> processConfig(snode, body, key, configFactory.userGroups)
if (requests.isNotEmpty()) {
SnodeAPI.getRawBatchResponse(snode, userPublicKey, requests).bind { rawResponses ->
isCaughtUp = true
if (deferred.promise.isDone()) {
return@bind Promise.ofSuccess(Unit)
} else {
val responseList = (rawResponses["results"] as List<RawResponse>)
// in case we had null configs, the array won't be fully populated
// index of the sparse array key iterator should be the request index, with the key being the namespace
listOfNotNull(
configFactory.user?.configNamespace(),
configFactory.contacts?.configNamespace(),
configFactory.userGroups?.configNamespace(),
configFactory.convoVolatile?.configNamespace()
).map {
it to requestSparseArray.indexOfKey(it)
}.filter { (_, i) -> i >= 0 }.forEach { (key, requestIndex) ->
responseList.getOrNull(requestIndex)?.let { rawResponse ->
if (rawResponse["code"] as? Int != 200) {
Log.e("Loki", "Batch sub-request had non-200 response code, returned code ${(rawResponse["code"] as? Int) ?: "[unknown]"}")
return@forEach
}
}
}
}
// the first response will be the personal messages (we want these to be processed after config messages)
val personalResponseIndex = requestSparseArray.indexOfKey(Namespace.DEFAULT)
if (personalResponseIndex >= 0) {
responseList.getOrNull(personalResponseIndex)?.let { rawResponse ->
if (rawResponse["code"] as? Int != 200) {
Log.e("Loki", "Batch sub-request for personal messages had non-200 response code, returned code ${(rawResponse["code"] as? Int) ?: "[unknown]"}")
} else {
val body = rawResponse["body"] as? RawResponse
if (body == null) {
Log.e("Loki", "Batch sub-request for personal messages didn't contain a body")
Log.e("Loki", "Batch sub-request didn't contain a body")
return@forEach
}
if (key == Namespace.DEFAULT) {
return@forEach // continue, skip default namespace
} else {
processPersonalMessages(snode, body)
when (ConfigBase.kindFor(key)) {
UserProfile::class.java -> processConfig(snode, body, key, configFactory.user)
Contacts::class.java -> processConfig(snode, body, key, configFactory.contacts)
ConversationVolatileConfig::class.java -> processConfig(snode, body, key, configFactory.convoVolatile)
UserGroupsConfig::class.java -> processConfig(snode, body, key, configFactory.userGroups)
}
}
}
}
}
// the first response will be the personal messages (we want these to be processed after config messages)
val personalResponseIndex = requestSparseArray.indexOfKey(Namespace.DEFAULT)
if (personalResponseIndex >= 0) {
responseList.getOrNull(personalResponseIndex)?.let { rawResponse ->
if (rawResponse["code"] as? Int != 200) {
Log.e("Loki", "Batch sub-request for personal messages had non-200 response code, returned code ${(rawResponse["code"] as? Int) ?: "[unknown]"}")
// If we got a non-success response then the snode might be bad so we should try rotate
// to a different one just in case
pollNextSnode(deferred = deferred)
return@bind Promise.ofSuccess(Unit)
} else {
val body = rawResponse["body"] as? RawResponse
if (body == null) {
Log.e("Loki", "Batch sub-request for personal messages didn't contain a body")
} else {
processPersonalMessages(snode, body)
}
}
}
}
poll(snode, deferred)
}
}.fail {
Log.e("Loki", "Failed to get raw batch response", it)
poll(snode, deferred)
}
}.fail {
Log.e("Loki", "Failed to get raw batch response", it)
poll(snode, deferred)
}
}
}

2
libsession/src/main/java/org/session/libsession/snode/SnodeAPI.kt
View File

@ -520,7 +520,7 @@ object SnodeAPI {
Log.w("Loki", "response code was not 200")
handleSnodeError(
response["code"] as? Int ?: 0,
response,
response["body"] as? Map<*, *>,
snode,
publicKey
)