From 5f5bab74a799efe963f7d630a42db7800e2a39f9 Mon Sep 17 00:00:00 2001 From: Dimitris Papavasiliou Date: Sun, 23 Sep 2018 16:25:38 +0300 Subject: [PATCH] Detect and recycle dangling websockets. Co-authored-by: Dimitris Papavasiliou Co-authored-by: Ralf Kohrt --- .../api/util/RealtimeSleepTimer.java | 13 +--- .../api/SignalServiceMessagePipe.java | 4 +- .../websocket/WebSocketConnection.java | 77 +++++++++++-------- 3 files changed, 51 insertions(+), 43 deletions(-) diff --git a/android/src/main/java/org/whispersystems/signalservice/api/util/RealtimeSleepTimer.java b/android/src/main/java/org/whispersystems/signalservice/api/util/RealtimeSleepTimer.java index 55c2eb7d8..6b93aaa86 100644 --- a/android/src/main/java/org/whispersystems/signalservice/api/util/RealtimeSleepTimer.java +++ b/android/src/main/java/org/whispersystems/signalservice/api/util/RealtimeSleepTimer.java @@ -31,21 +31,14 @@ public RealtimeSleepTimer(Context context) { } @Override - public void sleep(long millis) { + public void sleep(long millis) throws InterruptedException { context.registerReceiver(alarmReceiver, new IntentFilter(AlarmReceiver.WAKE_UP_THREAD_ACTION)); - final long startTime = System.currentTimeMillis(); alarmReceiver.setAlarm(millis); - while (System.currentTimeMillis() - startTime < millis) { - try { - synchronized (this) { - wait(millis - System.currentTimeMillis() + startTime); - } - } catch (InterruptedException e) { - Log.w(TAG, e); - } + synchronized (this) { + wait(millis); } context.unregisterReceiver(alarmReceiver); diff --git a/java/src/main/java/org/whispersystems/signalservice/api/SignalServiceMessagePipe.java b/java/src/main/java/org/whispersystems/signalservice/api/SignalServiceMessagePipe.java index 019820170..ba0d1650b 100644 --- a/java/src/main/java/org/whispersystems/signalservice/api/SignalServiceMessagePipe.java +++ b/java/src/main/java/org/whispersystems/signalservice/api/SignalServiceMessagePipe.java @@ -63,7 +63,7 @@ public class SignalServiceMessagePipe { * @throws TimeoutException */ public SignalServiceEnvelope read(long timeout, TimeUnit unit) - throws InvalidVersionException, IOException, TimeoutException + throws InvalidVersionException, IOException, TimeoutException, InterruptedException { return read(timeout, unit, new NullMessagePipeCallback()); } @@ -86,7 +86,7 @@ public SignalServiceEnvelope read(long timeout, TimeUnit unit) * @throws InvalidVersionException */ public SignalServiceEnvelope read(long timeout, TimeUnit unit, MessagePipeCallback callback) - throws TimeoutException, IOException, InvalidVersionException + throws TimeoutException, IOException, InvalidVersionException, InterruptedException { while (true) { WebSocketRequestMessage request = websocket.readRequest(unit.toMillis(timeout)); diff --git a/java/src/main/java/org/whispersystems/signalservice/internal/websocket/WebSocketConnection.java b/java/src/main/java/org/whispersystems/signalservice/internal/websocket/WebSocketConnection.java index 132ae51cb..2b7e9f101 100644 --- a/java/src/main/java/org/whispersystems/signalservice/internal/websocket/WebSocketConnection.java +++ b/java/src/main/java/org/whispersystems/signalservice/internal/websocket/WebSocketConnection.java @@ -19,6 +19,7 @@ import java.util.Iterator; import java.util.LinkedList; import java.util.Map; +import java.util.concurrent.ExecutionException; import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; @@ -116,22 +117,20 @@ public synchronized void disconnect() { } if (keepAliveSender != null) { - keepAliveSender.shutdown(); + keepAliveSender.interrupt(); keepAliveSender = null; } } public synchronized WebSocketRequestMessage readRequest(long timeoutMillis) - throws TimeoutException, IOException + throws TimeoutException, IOException, InterruptedException { if (client == null) { throw new IOException("Connection closed!"); } - long startTime = System.currentTimeMillis(); - - while (client != null && incomingRequests.isEmpty() && elapsedTime(startTime) < timeoutMillis) { - Util.wait(this, Math.max(1, timeoutMillis - elapsedTime(startTime))); + if (client != null && incomingRequests.isEmpty()) { + wait(timeoutMillis); } if (incomingRequests.isEmpty() && client == null) throw new IOException("Connection closed!"); @@ -172,20 +171,16 @@ public synchronized void sendResponse(WebSocketResponseMessage response) throws } } - private synchronized void sendKeepAlive() throws IOException { + private synchronized Future> sendKeepAlive() throws IOException { if (keepAliveSender != null && client != null) { - byte[] message = WebSocketMessage.newBuilder() - .setType(WebSocketMessage.Type.REQUEST) - .setRequest(WebSocketRequestMessage.newBuilder() - .setId(System.currentTimeMillis()) - .setPath("/v1/keepalive") - .setVerb("GET") - .build()).build() - .toByteArray(); - - if (!client.send(ByteString.of(message))) { - throw new IOException("Write failed!"); - } + WebSocketRequestMessage request = WebSocketRequestMessage.newBuilder() + .setId(System.currentTimeMillis()) + .setPath("/v1/keepalive") + .setVerb("GET") + .build(); + return sendRequest(request); + } else { + return null; } } @@ -238,7 +233,7 @@ public synchronized void onClosed(WebSocket webSocket, int code, String reason) } if (keepAliveSender != null) { - keepAliveSender.shutdown(); + keepAliveSender.interrupt(); keepAliveSender = null; } @@ -301,23 +296,43 @@ private Pair createTlsSocketFactory(TrustSto private class KeepAliveSender extends Thread { - private AtomicBoolean stop = new AtomicBoolean(false); - public void run() { - while (!stop.get()) { + Future future = null; + boolean severed = false; + + while (!interrupted()) { try { sleepTimer.sleep(TimeUnit.SECONDS.toMillis(KEEPALIVE_TIMEOUT_SECONDS)); - Log.w(TAG, "Sending keep alive..."); - sendKeepAlive(); - } catch (Throwable e) { - Log.w(TAG, e); + if (future != null) { + try { + future.get(0L, TimeUnit.SECONDS); + } catch (ExecutionException | TimeoutException e){ + severed = true; + } + } + } catch (InterruptedException e) { + Log.d(TAG, "Keep alive sender interrupted; exiting loop."); + break; } - } - } - public void shutdown() { - stop.set(true); + if (severed) { + Log.d(TAG, "No response to previous keep-alive; forcing new connection."); + + disconnect(); + synchronized(WebSocketConnection.this) { + WebSocketConnection.this.notifyAll(); + } + } else { + Log.d(TAG, "Sending keep alive..."); + + try { + future = sendKeepAlive(); + } catch (IOException e) { + Log.d(TAG, "Failed to send keep alive: " + e.getMessage()); + } + } + } } }