Discard stale messages from MessagesCache#getMessagesToPersist
This commit is contained in:
parent
3f9863c441
commit
6d30a45017
|
@ -528,7 +528,6 @@ public class MessagesCache {
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
@VisibleForTesting
|
|
||||||
List<MessageProtos.Envelope> getMessagesToPersist(final UUID accountUuid, final byte destinationDevice,
|
List<MessageProtos.Envelope> getMessagesToPersist(final UUID accountUuid, final byte destinationDevice,
|
||||||
final int limit) {
|
final int limit) {
|
||||||
|
|
||||||
|
@ -537,7 +536,7 @@ public class MessagesCache {
|
||||||
final List<byte[]> messages = redisCluster.withBinaryCluster(connection ->
|
final List<byte[]> messages = redisCluster.withBinaryCluster(connection ->
|
||||||
connection.sync().zrange(getMessageQueueKey(accountUuid, destinationDevice), 0, limit));
|
connection.sync().zrange(getMessageQueueKey(accountUuid, destinationDevice), 0, limit));
|
||||||
|
|
||||||
return Flux.fromIterable(messages)
|
final Flux<MessageProtos.Envelope> allMessages = Flux.fromIterable(messages)
|
||||||
.mapNotNull(message -> {
|
.mapNotNull(message -> {
|
||||||
try {
|
try {
|
||||||
return MessageProtos.Envelope.parseFrom(message);
|
return MessageProtos.Envelope.parseFrom(message);
|
||||||
|
@ -555,7 +554,22 @@ public class MessagesCache {
|
||||||
}
|
}
|
||||||
|
|
||||||
return messageMono;
|
return messageMono;
|
||||||
})
|
});
|
||||||
|
|
||||||
|
final Flux<MessageProtos.Envelope> messagesToPersist = allMessages
|
||||||
|
.filter(Predicate.not(envelope ->
|
||||||
|
envelope.getEphemeral() || isStaleMrmMessage(envelope)));
|
||||||
|
|
||||||
|
final Flux<MessageProtos.Envelope> ephemeralMessages = allMessages
|
||||||
|
.filter(MessageProtos.Envelope::getEphemeral);
|
||||||
|
discardStaleMessages(accountUuid, destinationDevice, ephemeralMessages, staleEphemeralMessagesCounter, "ephemeral");
|
||||||
|
|
||||||
|
final Flux<MessageProtos.Envelope> staleMrmMessages = allMessages.filter(MessagesCache::isStaleMrmMessage)
|
||||||
|
// clearing the sharedMrmKey prevents unnecessary calls to update the shared MRM data
|
||||||
|
.map(envelope -> envelope.toBuilder().clearSharedMrmKey().build());
|
||||||
|
discardStaleMessages(accountUuid, destinationDevice, staleMrmMessages, staleMrmMessagesCounter, "mrm");
|
||||||
|
|
||||||
|
return messagesToPersist
|
||||||
.collectList()
|
.collectList()
|
||||||
.doOnTerminate(() -> sample.stop(getMessagesTimer))
|
.doOnTerminate(() -> sample.stop(getMessagesTimer))
|
||||||
.block(Duration.ofSeconds(5));
|
.block(Duration.ofSeconds(5));
|
||||||
|
|
|
@ -267,11 +267,7 @@ public class MessagesManager {
|
||||||
final Device destinationDevice,
|
final Device destinationDevice,
|
||||||
final List<Envelope> messages) {
|
final List<Envelope> messages) {
|
||||||
|
|
||||||
final List<Envelope> nonEphemeralMessages = messages.stream()
|
messagesDynamoDb.store(messages, destinationUuid, destinationDevice);
|
||||||
.filter(envelope -> !envelope.getEphemeral())
|
|
||||||
.collect(Collectors.toList());
|
|
||||||
|
|
||||||
messagesDynamoDb.store(nonEphemeralMessages, destinationUuid, destinationDevice);
|
|
||||||
|
|
||||||
final List<UUID> messageGuids = messages.stream().map(message -> UUID.fromString(message.getServerGuid()))
|
final List<UUID> messageGuids = messages.stream().map(message -> UUID.fromString(message.getServerGuid()))
|
||||||
.collect(Collectors.toList());
|
.collect(Collectors.toList());
|
||||||
|
@ -279,7 +275,7 @@ public class MessagesManager {
|
||||||
try {
|
try {
|
||||||
messagesRemovedFromCache = messagesCache.remove(destinationUuid, destinationDevice.getId(), messageGuids)
|
messagesRemovedFromCache = messagesCache.remove(destinationUuid, destinationDevice.getId(), messageGuids)
|
||||||
.get(30, TimeUnit.SECONDS).size();
|
.get(30, TimeUnit.SECONDS).size();
|
||||||
PERSIST_MESSAGE_COUNTER.increment(nonEphemeralMessages.size());
|
PERSIST_MESSAGE_COUNTER.increment(messages.size());
|
||||||
|
|
||||||
} catch (InterruptedException | ExecutionException | TimeoutException e) {
|
} catch (InterruptedException | ExecutionException | TimeoutException e) {
|
||||||
logger.warn("Failed to remove messages from cache", e);
|
logger.warn("Failed to remove messages from cache", e);
|
||||||
|
|
Loading…
Reference in New Issue