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

Refactor: move sendNextMessagePage into its own method.

This commit is contained in:
Jon Chambers 2020-09-10 18:06:04 -04:00 committed by Jon Chambers
parent 6f9ff3be37
commit 158e5004b7
1 changed files with 32 additions and 28 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -205,19 +205,42 @@ public class WebSocketConnection implements DispatchChannel, MessageAvailability
cachedMessagesOnly = lastPersistedState <= lastDatabaseClearedState; cachedMessagesOnly = lastPersistedState <= lastDatabaseClearedState;
} }
OutgoingMessageEntityList messages = messagesManager.getMessagesForDevice(account.getNumber(), account.getUuid(), device.getId(), client.getUserAgent(), cachedMessagesOnly); sendNextMessagePage(cachedMessagesOnly).thenAccept(hasMoreStoredMessages -> {
CompletableFuture<?>[] sendFutures = new CompletableFuture[messages.getMessages().size()]; final boolean mayHaveMoreMessages;
synchronized (this) {
processingStoredMessages = false;
mayHaveMoreMessages = hasMoreStoredMessages || storedMessageState > processedState;
}
if (mayHaveMoreMessages) {
processStoredMessages();
} else {
synchronized (this) {
lastDatabaseClearedState = processedState;
}
if (sentInitialQueueEmptyMessage.compareAndSet(false, true)) {
client.sendRequest("PUT", "/api/v1/queue/empty", Collections.singletonList(TimestampHeaderUtil.getTimestampHeader()), Optional.empty());
}
}
});
}
private CompletableFuture<Boolean> sendNextMessagePage(final boolean cachedMessagesOnly) {
final OutgoingMessageEntityList messages = messagesManager.getMessagesForDevice(account.getNumber(), account.getUuid(), device.getId(), client.getUserAgent(), cachedMessagesOnly);
final CompletableFuture<?>[] sendFutures = new CompletableFuture[messages.getMessages().size()];
for (int i = 0; i < messages.getMessages().size(); i++) { for (int i = 0; i < messages.getMessages().size(); i++) {
OutgoingMessageEntity message = messages.getMessages().get(i); final OutgoingMessageEntity message = messages.getMessages().get(i);
Envelope.Builder builder = Envelope.newBuilder() final Envelope.Builder builder = Envelope.newBuilder()
.setType(Envelope.Type.valueOf(message.getType())) .setType(Envelope.Type.valueOf(message.getType()))
.setTimestamp(message.getTimestamp()) .setTimestamp(message.getTimestamp())
.setServerTimestamp(message.getServerTimestamp()); .setServerTimestamp(message.getServerTimestamp());
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) {
@ -235,26 +258,7 @@ public class WebSocketConnection implements DispatchChannel, MessageAvailability
sendFutures[i] = sendMessage(builder.build(), Optional.of(new StoredMessageInfo(message.getId(), message.isCached()))); sendFutures[i] = sendMessage(builder.build(), Optional.of(new StoredMessageInfo(message.getId(), message.isCached())));
} }
CompletableFuture.allOf(sendFutures).whenComplete((v, cause) -> { return CompletableFuture.allOf(sendFutures).handle((v, cause) -> messages.hasMore());
final boolean mayHaveMoreMessages;
synchronized (this) {
processingStoredMessages = false;
mayHaveMoreMessages = messages.hasMore() || storedMessageState > processedState;
}
if (mayHaveMoreMessages) {
processStoredMessages();
} else {
synchronized (this) {
lastDatabaseClearedState = processedState;
}
if (sentInitialQueueEmptyMessage.compareAndSet(false, true)) {
client.sendRequest("PUT", "/api/v1/queue/empty", Collections.singletonList(TimestampHeaderUtil.getTimestampHeader()), Optional.empty());
}
}
});
} }
@Override @Override