mirror of
https://github.com/signalapp/Signal-Android.git
synced 2026-04-19 16:19:33 +01:00
Separate and kotlinize websockets.
This commit is contained in:
Cody Henthorne
committed by
Michelle Tang
Michelle Tang
parent
6c9acf4657
commit
93d18c1763
@@ -55,6 +55,6 @@ public final class DeviceTransferBlockingInterceptor implements Interceptor {
|
||||
|
||||
public void unblockNetwork() {
|
||||
blockNetworking = false;
|
||||
AppDependencies.getIncomingMessageObserver();
|
||||
AppDependencies.startNetwork();
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,207 +0,0 @@
|
||||
package org.thoughtcrime.securesms.net;
|
||||
|
||||
import android.app.Application;
|
||||
|
||||
import androidx.annotation.NonNull;
|
||||
|
||||
import org.signal.core.util.logging.Log;
|
||||
import org.thoughtcrime.securesms.util.TextSecurePreferences;
|
||||
import org.whispersystems.signalservice.api.SignalWebSocket;
|
||||
import org.whispersystems.signalservice.api.util.Preconditions;
|
||||
import org.whispersystems.signalservice.api.util.SleepTimer;
|
||||
import org.whispersystems.signalservice.api.websocket.HealthMonitor;
|
||||
import org.whispersystems.signalservice.api.websocket.WebSocketConnectionState;
|
||||
import org.whispersystems.signalservice.internal.websocket.OkHttpWebSocketConnection;
|
||||
|
||||
import java.util.concurrent.Executor;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import io.reactivex.rxjava3.schedulers.Schedulers;
|
||||
|
||||
/**
|
||||
* Monitors the health of the identified and unidentified WebSockets. If either one appears to be
|
||||
* unhealthy, will trigger restarting both.
|
||||
* <p>
|
||||
* The monitor is also responsible for sending heartbeats/keep-alive messages to prevent
|
||||
* timeouts.
|
||||
*/
|
||||
public final class SignalWebSocketHealthMonitor implements HealthMonitor {
|
||||
|
||||
private static final String TAG = Log.tag(SignalWebSocketHealthMonitor.class);
|
||||
|
||||
/**
|
||||
* This is the amount of time in between sent keep alives. Must be greater than {@link SignalWebSocketHealthMonitor#KEEP_ALIVE_TIMEOUT}
|
||||
*/
|
||||
private static final long KEEP_ALIVE_SEND_CADENCE = TimeUnit.SECONDS.toMillis(OkHttpWebSocketConnection.KEEPALIVE_FREQUENCY_SECONDS);
|
||||
|
||||
/**
|
||||
* This is the amount of time we will wait for a response to the keep alive before we consider the websockets dead.
|
||||
* It is required that this value be less than {@link SignalWebSocketHealthMonitor#KEEP_ALIVE_SEND_CADENCE}
|
||||
*/
|
||||
private static final long KEEP_ALIVE_TIMEOUT = TimeUnit.SECONDS.toMillis(20);
|
||||
|
||||
private final Executor executor = Executors.newSingleThreadExecutor();
|
||||
|
||||
private final Application context;
|
||||
private SignalWebSocket signalWebSocket;
|
||||
private final SleepTimer sleepTimer;
|
||||
|
||||
private KeepAliveSender keepAliveSender;
|
||||
|
||||
private final HealthState identified = new HealthState();
|
||||
private final HealthState unidentified = new HealthState();
|
||||
|
||||
public SignalWebSocketHealthMonitor(@NonNull Application context, @NonNull SleepTimer sleepTimer) {
|
||||
this.context = context;
|
||||
this.sleepTimer = sleepTimer;
|
||||
}
|
||||
|
||||
public void monitor(@NonNull SignalWebSocket signalWebSocket) {
|
||||
executor.execute(() -> {
|
||||
Preconditions.checkNotNull(signalWebSocket);
|
||||
Preconditions.checkArgument(this.signalWebSocket == null, "monitor can only be called once");
|
||||
|
||||
this.signalWebSocket = signalWebSocket;
|
||||
|
||||
//noinspection ResultOfMethodCallIgnored
|
||||
signalWebSocket.getWebSocketState()
|
||||
.subscribeOn(Schedulers.computation())
|
||||
.observeOn(Schedulers.computation())
|
||||
.distinctUntilChanged()
|
||||
.subscribe(s -> onStateChange(s, identified, true));
|
||||
|
||||
//noinspection ResultOfMethodCallIgnored
|
||||
signalWebSocket.getUnidentifiedWebSocketState()
|
||||
.subscribeOn(Schedulers.computation())
|
||||
.observeOn(Schedulers.computation())
|
||||
.distinctUntilChanged()
|
||||
.subscribe(s -> onStateChange(s, unidentified, false));
|
||||
});
|
||||
}
|
||||
|
||||
private void onStateChange(WebSocketConnectionState connectionState, HealthState healthState, boolean isIdentified) {
|
||||
executor.execute(() -> {
|
||||
switch (connectionState) {
|
||||
case CONNECTED:
|
||||
if (isIdentified) {
|
||||
TextSecurePreferences.setUnauthorizedReceived(context, false);
|
||||
break;
|
||||
}
|
||||
case AUTHENTICATION_FAILED:
|
||||
if (isIdentified) {
|
||||
TextSecurePreferences.setUnauthorizedReceived(context, true);
|
||||
break;
|
||||
}
|
||||
case FAILED:
|
||||
break;
|
||||
}
|
||||
|
||||
healthState.needsKeepAlive = connectionState == WebSocketConnectionState.CONNECTED;
|
||||
|
||||
if (keepAliveSender == null && isKeepAliveNecessary()) {
|
||||
keepAliveSender = new KeepAliveSender();
|
||||
keepAliveSender.start();
|
||||
} else if (keepAliveSender != null && !isKeepAliveNecessary()) {
|
||||
keepAliveSender.shutdown();
|
||||
keepAliveSender = null;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onKeepAliveResponse(long sentTimestamp, boolean isIdentifiedWebSocket) {
|
||||
final long keepAliveTime = System.currentTimeMillis();
|
||||
executor.execute(() -> {
|
||||
if (isIdentifiedWebSocket) {
|
||||
identified.lastKeepAliveReceived = keepAliveTime;
|
||||
} else {
|
||||
unidentified.lastKeepAliveReceived = keepAliveTime;
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onMessageError(int status, boolean isIdentifiedWebSocket) {
|
||||
executor.execute(() -> {
|
||||
if (status == 409) {
|
||||
HealthState healthState = (isIdentifiedWebSocket ? identified : unidentified);
|
||||
if (healthState.mismatchErrorTracker.addSample(System.currentTimeMillis())) {
|
||||
Log.w(TAG, "Received too many mismatch device errors, forcing new websockets.");
|
||||
signalWebSocket.forceNewWebSockets();
|
||||
}
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
private boolean isKeepAliveNecessary() {
|
||||
return identified.needsKeepAlive || unidentified.needsKeepAlive;
|
||||
}
|
||||
|
||||
private static class HealthState {
|
||||
private final HttpErrorTracker mismatchErrorTracker = new HttpErrorTracker(5, TimeUnit.MINUTES.toMillis(1));
|
||||
|
||||
private volatile boolean needsKeepAlive;
|
||||
private volatile long lastKeepAliveReceived;
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends periodic heartbeats/keep-alives over both WebSockets to prevent connection timeouts. If
|
||||
* either WebSocket fails to get a return heartbeat after {@link SignalWebSocketHealthMonitor#KEEP_ALIVE_TIMEOUT} seconds, both are forced to be recreated.
|
||||
*/
|
||||
private class KeepAliveSender extends Thread {
|
||||
|
||||
private volatile boolean shouldKeepRunning = true;
|
||||
|
||||
public void run() {
|
||||
Log.d(TAG, "[KeepAliveSender] started");
|
||||
identified.lastKeepAliveReceived = System.currentTimeMillis();
|
||||
unidentified.lastKeepAliveReceived = System.currentTimeMillis();
|
||||
|
||||
long keepAliveSendTime = System.currentTimeMillis();
|
||||
while (shouldKeepRunning && isKeepAliveNecessary()) {
|
||||
try {
|
||||
long nextKeepAliveSendTime = (keepAliveSendTime + KEEP_ALIVE_SEND_CADENCE);
|
||||
sleepUntil(nextKeepAliveSendTime);
|
||||
|
||||
if (shouldKeepRunning && isKeepAliveNecessary()) {
|
||||
keepAliveSendTime = System.currentTimeMillis();
|
||||
signalWebSocket.sendKeepAlive();
|
||||
}
|
||||
|
||||
final long responseRequiredTime = keepAliveSendTime + KEEP_ALIVE_TIMEOUT;
|
||||
sleepUntil(responseRequiredTime);
|
||||
|
||||
if (shouldKeepRunning && isKeepAliveNecessary()) {
|
||||
if (identified.lastKeepAliveReceived < keepAliveSendTime || unidentified.lastKeepAliveReceived < keepAliveSendTime) {
|
||||
Log.w(TAG, "Missed keep alives, identified last: " + identified.lastKeepAliveReceived +
|
||||
" unidentified last: " + unidentified.lastKeepAliveReceived +
|
||||
" needed by: " + responseRequiredTime);
|
||||
signalWebSocket.forceNewWebSockets();
|
||||
}
|
||||
}
|
||||
} catch (Throwable e) {
|
||||
Log.w(TAG, e);
|
||||
}
|
||||
}
|
||||
Log.d(TAG, "[KeepAliveSender] ended");
|
||||
}
|
||||
|
||||
private void sleepUntil(long timeMs) {
|
||||
while (System.currentTimeMillis() < timeMs) {
|
||||
long waitTime = timeMs - System.currentTimeMillis();
|
||||
if (waitTime > 0) {
|
||||
try {
|
||||
sleepTimer.sleep(waitTime);
|
||||
} catch (InterruptedException e) {
|
||||
Log.w(TAG, e);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
public void shutdown() {
|
||||
shouldKeepRunning = false;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,172 @@
|
||||
/*
|
||||
* Copyright 2025 Signal Messenger, LLC
|
||||
* SPDX-License-Identifier: AGPL-3.0-only
|
||||
*/
|
||||
|
||||
package org.thoughtcrime.securesms.net
|
||||
|
||||
import io.reactivex.rxjava3.kotlin.subscribeBy
|
||||
import io.reactivex.rxjava3.schedulers.Schedulers
|
||||
import org.signal.core.util.logging.Log
|
||||
import org.thoughtcrime.securesms.dependencies.AppDependencies
|
||||
import org.thoughtcrime.securesms.util.TextSecurePreferences
|
||||
import org.whispersystems.signalservice.api.util.SleepTimer
|
||||
import org.whispersystems.signalservice.api.websocket.HealthMonitor
|
||||
import org.whispersystems.signalservice.api.websocket.SignalWebSocket
|
||||
import org.whispersystems.signalservice.api.websocket.WebSocketConnectionState
|
||||
import org.whispersystems.signalservice.internal.websocket.OkHttpWebSocketConnection
|
||||
import java.util.concurrent.Executor
|
||||
import java.util.concurrent.Executors
|
||||
import kotlin.concurrent.Volatile
|
||||
import kotlin.time.Duration
|
||||
import kotlin.time.Duration.Companion.milliseconds
|
||||
import kotlin.time.Duration.Companion.seconds
|
||||
|
||||
class SignalWebSocketHealthMonitor(
|
||||
private val sleepTimer: SleepTimer
|
||||
) : HealthMonitor {
|
||||
|
||||
companion object {
|
||||
private val TAG = Log.tag(SignalWebSocketHealthMonitor::class)
|
||||
|
||||
/**
|
||||
* This is the amount of time in between sent keep alives. Must be greater than [KEEP_ALIVE_TIMEOUT]
|
||||
*/
|
||||
private val KEEP_ALIVE_SEND_CADENCE: Duration = OkHttpWebSocketConnection.KEEPALIVE_FREQUENCY_SECONDS.seconds
|
||||
|
||||
/**
|
||||
* This is the amount of time we will wait for a response to the keep alive before we consider the websockets dead.
|
||||
* It is required that this value be less than [KEEP_ALIVE_SEND_CADENCE]
|
||||
*/
|
||||
private val KEEP_ALIVE_TIMEOUT: Duration = 20.seconds
|
||||
}
|
||||
|
||||
private val executor: Executor = Executors.newSingleThreadExecutor()
|
||||
|
||||
private var webSocket: SignalWebSocket? = null
|
||||
|
||||
private var keepAliveSender: KeepAliveSender? = null
|
||||
private var needsKeepAlive = false
|
||||
private var lastKeepAliveReceived: Duration = 0.seconds
|
||||
|
||||
@Suppress("CheckResult")
|
||||
fun monitor(webSocket: SignalWebSocket) {
|
||||
executor.execute {
|
||||
check(this.webSocket == null)
|
||||
|
||||
this.webSocket = webSocket
|
||||
|
||||
webSocket
|
||||
.state
|
||||
.subscribeOn(Schedulers.computation())
|
||||
.observeOn(Schedulers.computation())
|
||||
.distinctUntilChanged()
|
||||
.subscribeBy { onStateChanged(it) }
|
||||
|
||||
webSocket.keepAliveChangedListener = this::updateKeepAliveSenderStatus
|
||||
}
|
||||
}
|
||||
|
||||
private fun onStateChanged(connectionState: WebSocketConnectionState) {
|
||||
executor.execute {
|
||||
when (connectionState) {
|
||||
WebSocketConnectionState.CONNECTED -> {
|
||||
if (webSocket is SignalWebSocket.AuthenticatedWebSocket) {
|
||||
TextSecurePreferences.setUnauthorizedReceived(AppDependencies.application, false)
|
||||
}
|
||||
}
|
||||
WebSocketConnectionState.AUTHENTICATION_FAILED -> {
|
||||
if (webSocket is SignalWebSocket.AuthenticatedWebSocket) {
|
||||
TextSecurePreferences.setUnauthorizedReceived(AppDependencies.application, true)
|
||||
}
|
||||
}
|
||||
else -> Unit
|
||||
}
|
||||
|
||||
needsKeepAlive = connectionState == WebSocketConnectionState.CONNECTED
|
||||
|
||||
updateKeepAliveSenderStatus()
|
||||
}
|
||||
}
|
||||
|
||||
override fun onKeepAliveResponse(sentTimestamp: Long, isIdentifiedWebSocket: Boolean) {
|
||||
val keepAliveTime = System.currentTimeMillis().milliseconds
|
||||
executor.execute {
|
||||
lastKeepAliveReceived = keepAliveTime
|
||||
}
|
||||
}
|
||||
|
||||
override fun onMessageError(status: Int, isIdentifiedWebSocket: Boolean) = Unit
|
||||
|
||||
private fun updateKeepAliveSenderStatus() {
|
||||
if (keepAliveSender == null && sendKeepAlives()) {
|
||||
keepAliveSender = KeepAliveSender()
|
||||
keepAliveSender!!.start()
|
||||
} else if (keepAliveSender != null && !sendKeepAlives()) {
|
||||
keepAliveSender!!.shutdown()
|
||||
keepAliveSender = null
|
||||
}
|
||||
}
|
||||
|
||||
private fun sendKeepAlives(): Boolean {
|
||||
return needsKeepAlive && webSocket?.shouldSendKeepAlives == true
|
||||
}
|
||||
|
||||
/**
|
||||
* Sends periodic heartbeats/keep-alives over the WebSocket to prevent connection timeouts. If
|
||||
* the WebSocket fails to get a return heartbeat after [KEEP_ALIVE_TIMEOUT] seconds, it is forced to be recreated.
|
||||
*/
|
||||
private inner class KeepAliveSender : Thread() {
|
||||
|
||||
@Volatile
|
||||
private var shouldKeepRunning = true
|
||||
|
||||
override fun run() {
|
||||
Log.d(TAG, "[KeepAliveSender($id)] started")
|
||||
lastKeepAliveReceived = System.currentTimeMillis().milliseconds
|
||||
|
||||
var keepAliveSendTime = System.currentTimeMillis().milliseconds
|
||||
while (shouldKeepRunning && sendKeepAlives()) {
|
||||
try {
|
||||
val nextKeepAliveSendTime: Duration = keepAliveSendTime + KEEP_ALIVE_SEND_CADENCE
|
||||
sleepUntil(nextKeepAliveSendTime)
|
||||
|
||||
if (shouldKeepRunning && sendKeepAlives()) {
|
||||
keepAliveSendTime = System.currentTimeMillis().milliseconds
|
||||
webSocket?.sendKeepAlive()
|
||||
}
|
||||
|
||||
val responseRequiredTime: Duration = keepAliveSendTime + KEEP_ALIVE_TIMEOUT
|
||||
sleepUntil(responseRequiredTime)
|
||||
|
||||
if (shouldKeepRunning && sendKeepAlives()) {
|
||||
if (lastKeepAliveReceived < keepAliveSendTime) {
|
||||
Log.w(TAG, "Missed keep alive, last: ${lastKeepAliveReceived.inWholeMilliseconds} needed by: ${responseRequiredTime.inWholeMilliseconds}")
|
||||
webSocket?.forceNewWebSocket()
|
||||
}
|
||||
}
|
||||
} catch (e: Throwable) {
|
||||
Log.w(TAG, e)
|
||||
}
|
||||
}
|
||||
Log.d(TAG, "[KeepAliveSender($id)] ended")
|
||||
}
|
||||
|
||||
fun sleepUntil(time: Duration) {
|
||||
while (System.currentTimeMillis().milliseconds < time) {
|
||||
val waitTime = time - System.currentTimeMillis().milliseconds
|
||||
if (waitTime.isPositive()) {
|
||||
try {
|
||||
sleepTimer.sleep(waitTime.inWholeMilliseconds)
|
||||
} catch (e: InterruptedException) {
|
||||
Log.w(TAG, e)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fun shutdown() {
|
||||
shouldKeepRunning = false
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user