Signal/Android
1
0
Fork 0
mirror of https://github.com/signalapp/Signal-Android.git synced 2025-12-23 04:28:35 +00:00
Code Issues Packages Projects Releases Wiki Activity

Reschedule job if background web socket message retrieval fails.

Browse Source 
Closes #12971
This commit is contained in:
g1a55er
2023-05-27 19:33:26 -04:00
committed by Cody Henthorne
parent f3b629bc06
commit 03a212eee4
1 changed files with 16 additions and 5 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
app/src/main/java/org/thoughtcrime/securesms/messages/WebSocketStrategy.java
View File

@@ -39,14 +39,19 @@ public class WebSocketStrategy extends MessageRetrievalStrategy {
jobManager.addListener(job -> job.getParameters().getQueue() != null && job.getParameters().getQueue().startsWith(PushProcessMessageJob.QUEUE_PREFIX), queueListener); jobManager.addListener(job -> job.getParameters().getQueue() != null && job.getParameters().getQueue().startsWith(PushProcessMessageJob.QUEUE_PREFIX), queueListener);
blockUntilWebsocketDrained(observer); if (!blockUntilWebsocketDrained(observer)) {
return false;
}
stopwatch.split("decryptions-drained"); stopwatch.split("decryptions-drained");
Set<String> processQueues = queueListener.getQueues(); Set<String> processQueues = queueListener.getQueues();
Log.d(TAG, "Discovered " + processQueues.size() + " queue(s): " + processQueues); Log.d(TAG, "Discovered " + processQueues.size() + " queue(s): " + processQueues);
for (String queue : processQueues) { for (String queue : processQueues) {
blockUntilJobQueueDrained(queue, QUEUE_TIMEOUT); if (!blockUntilJobQueueDrained(queue, QUEUE_TIMEOUT)) {
return false;
}
} }
stopwatch.split("process-drained"); stopwatch.split("process-drained");
@@ -58,7 +63,7 @@ public class WebSocketStrategy extends MessageRetrievalStrategy {
} }
} }
private static void blockUntilWebsocketDrained(IncomingMessageObserver observer) { private static boolean blockUntilWebsocketDrained(IncomingMessageObserver observer) {
CountDownLatch latch = new CountDownLatch(1); CountDownLatch latch = new CountDownLatch(1);
observer.addDecryptionDrainedListener(new Runnable() { observer.addDecryptionDrainedListener(new Runnable() {
@@ -69,15 +74,19 @@ public class WebSocketStrategy extends MessageRetrievalStrategy {
}); });
try { try {
if (!latch.await(1, TimeUnit.MINUTES)) { if (latch.await(1, TimeUnit.MINUTES)) {
return true;
} else {
Log.w(TAG, "Hit timeout while waiting for decryptions to drain!"); Log.w(TAG, "Hit timeout while waiting for decryptions to drain!");
return false;
} }
} catch (InterruptedException e) { } catch (InterruptedException e) {
Log.w(TAG, "Interrupted!", e); Log.w(TAG, "Interrupted!", e);
return false;
} }
} }
private static void blockUntilJobQueueDrained(@NonNull String queue, long timeoutMs) { private static boolean blockUntilJobQueueDrained(@NonNull String queue, long timeoutMs) {
long startTime = System.currentTimeMillis(); long startTime = System.currentTimeMillis();
final JobManager jobManager = ApplicationDependencies.getJobManager(); final JobManager jobManager = ApplicationDependencies.getJobManager();
final MarkerJob markerJob = new MarkerJob(queue); final MarkerJob markerJob = new MarkerJob(queue);
@@ -86,12 +95,14 @@ public class WebSocketStrategy extends MessageRetrievalStrategy {
if (!jobState.isPresent()) { if (!jobState.isPresent()) {
Log.w(TAG, "Timed out waiting for " + queue + " job(s) to finish!"); Log.w(TAG, "Timed out waiting for " + queue + " job(s) to finish!");
return false;
} }
long endTime = System.currentTimeMillis(); long endTime = System.currentTimeMillis();
long duration = endTime - startTime; long duration = endTime - startTime;
Log.d(TAG, "Waited " + duration + " ms for the " + queue + " job(s) to finish."); Log.d(TAG, "Waited " + duration + " ms for the " + queue + " job(s) to finish.");
return true;
} }
@Override @Override