MirrorMirror
/
Signal-Server
mirror of https://github.com/signalapp/Signal-Server.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Don't attempt to send more messages if sending failed for any reason.

This commit is contained in:
Jon Chambers 2020-09-17 15:05:55 -04:00 committed by Jon Chambers
parent e9e18afb4a
commit e146135bd1
1 changed files with 27 additions and 24 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

51
service/src/main/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnection.java
View File

@ -126,7 +126,7 @@ public class WebSocketConnection implements DispatchChannel, MessageAvailability
processStoredMessages(); processStoredMessages();
} }
private CompletableFuture<Void> sendMessage(final Envelope message, final Optional<StoredMessageInfo> storedMessageInfo) { private CompletableFuture<WebSocketResponseMessage> sendMessage(final Envelope message, final Optional<StoredMessageInfo> storedMessageInfo) {
try { try {
String header; String header;
Optional<byte[]> body; Optional<byte[]> body;
@ -141,25 +141,24 @@ public class WebSocketConnection implements DispatchChannel, MessageAvailability
sendMessageMeter.mark(); sendMessageMeter.mark();
return client.sendRequest("PUT", "/api/v1/message", List.of(header, TimestampHeaderUtil.getTimestampHeader()), body) return client.sendRequest("PUT", "/api/v1/message", List.of(header, TimestampHeaderUtil.getTimestampHeader()), body).whenComplete((response, throwable) -> {
.thenAccept(response -> { if (throwable == null) {
boolean isReceipt = message.getType() == Envelope.Type.RECEIPT; boolean isReceipt = message.getType() == Envelope.Type.RECEIPT;
if (isSuccessResponse(response) && !isReceipt) { if (isSuccessResponse(response) && !isReceipt) {
messageTime.update(System.currentTimeMillis() - message.getTimestamp()); messageTime.update(System.currentTimeMillis() - message.getTimestamp());
} }
if (isSuccessResponse(response)) { if (isSuccessResponse(response)) {
if (storedMessageInfo.isPresent()) messagesManager.delete(account.getNumber(), account.getUuid(), device.getId(), storedMessageInfo.get().id, storedMessageInfo.get().cached); if (storedMessageInfo.isPresent()) messagesManager.delete(account.getNumber(), account.getUuid(), device.getId(), storedMessageInfo.get().id, storedMessageInfo.get().cached);
if (!isReceipt) sendDeliveryReceiptFor(message); if (!isReceipt) sendDeliveryReceiptFor(message);
} else if (!isSuccessResponse(response) && !storedMessageInfo.isPresent()) { } else if (!isSuccessResponse(response) && !storedMessageInfo.isPresent()) {
requeueMessage(message); requeueMessage(message);
} }
}) } else {
.exceptionally(throwable -> { if (!storedMessageInfo.isPresent()) requeueMessage(message);
if (!storedMessageInfo.isPresent()) requeueMessage(message); }
return null; });
});
} catch (CryptoEncodingException e) { } catch (CryptoEncodingException e) {
logger.warn("Bad signaling key", e); logger.warn("Bad signaling key", e);
return CompletableFuture.failedFuture(e); return CompletableFuture.failedFuture(e);
@ -201,13 +200,13 @@ public class WebSocketConnection implements DispatchChannel, MessageAvailability
sendNextMessagePage(state != StoredMessageState.PERSISTED_NEW_MESSAGES_AVAILABLE, queueClearedFuture); sendNextMessagePage(state != StoredMessageState.PERSISTED_NEW_MESSAGES_AVAILABLE, queueClearedFuture);
queueClearedFuture.whenComplete((v, cause) -> { queueClearedFuture.whenComplete((v, cause) -> {
if (sentInitialQueueEmptyMessage.compareAndSet(false, true)) { if (cause == null && sentInitialQueueEmptyMessage.compareAndSet(false, true)) {
client.sendRequest("PUT", "/api/v1/queue/empty", Collections.singletonList(TimestampHeaderUtil.getTimestampHeader()), Optional.empty()); client.sendRequest("PUT", "/api/v1/queue/empty", Collections.singletonList(TimestampHeaderUtil.getTimestampHeader()), Optional.empty());
} }
processStoredMessagesSemaphore.release(); processStoredMessagesSemaphore.release();
if (storedMessageState.get() != StoredMessageState.EMPTY) { if (cause == null && storedMessageState.get() != StoredMessageState.EMPTY) {
processStoredMessages(); processStoredMessages();
} }
}); });
@ -227,7 +226,7 @@ public class WebSocketConnection implements DispatchChannel, MessageAvailability
if (!Util.isEmpty(message.getSource())) { if (!Util.isEmpty(message.getSource())) {
builder.setSource(message.getSource()) builder.setSource(message.getSource())
.setSourceDevice(message.getSourceDevice()); .setSourceDevice(message.getSourceDevice());
} }
if (message.getMessage() != null) { if (message.getMessage() != null) {
@ -246,10 +245,14 @@ public class WebSocketConnection implements DispatchChannel, MessageAvailability
} }
CompletableFuture.allOf(sendFutures).whenComplete((v, cause) -> { CompletableFuture.allOf(sendFutures).whenComplete((v, cause) -> {
if (messages.hasMore()) { if (cause == null) {
sendNextMessagePage(cachedMessagesOnly, queueClearedFuture); if (messages.hasMore()) {
sendNextMessagePage(cachedMessagesOnly, queueClearedFuture);
} else {
queueClearedFuture.complete(null);
}
} else { } else {
queueClearedFuture.complete(null); queueClearedFuture.completeExceptionally(cause);
} }
}); });
} }