Signal/Signal-Server
1
0
Fork 0
mirror of https://github.com/signalapp/Signal-Server synced 2026-04-22 20:48:03 +01:00
Code Issues Packages Projects Releases Wiki Activity

Extract common message-sending methods into a shared utility class

Browse Source
This commit is contained in:
Jon Chambers
2025-04-08 17:39:45 -04:00
committed by GitHub
parent caa81b4885
commit 8d8a2a5583
3 changed files with 150 additions and 115 deletions
Download Patch File Download Diff File Expand all files Collapse all files

59
service/src/main/java/org/whispersystems/textsecuregcm/grpc/MessagesAnonymousGrpcService.java
View File

@@ -8,8 +8,6 @@ package org.whispersystems.textsecuregcm.grpc;
import io.grpc.Status;
import io.grpc.StatusException;
import java.time.Clock;
import java.util.Arrays;
import java.util.Collections;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;
@@ -37,17 +35,13 @@ import org.whispersystems.textsecuregcm.limits.CardinalityEstimator;
import org.whispersystems.textsecuregcm.limits.RateLimiters;
import org.whispersystems.textsecuregcm.push.MessageSender;
import org.whispersystems.textsecuregcm.push.MessageTooLargeException;
import org.whispersystems.textsecuregcm.push.MessageUtil;
import org.whispersystems.textsecuregcm.spam.GrpcResponse;
import org.whispersystems.textsecuregcm.spam.MessageType;
import org.whispersystems.textsecuregcm.spam.SpamCheckResult;
import org.whispersystems.textsecuregcm.spam.SpamChecker;
import org.whispersystems.textsecuregcm.storage.Account;
import org.whispersystems.textsecuregcm.storage.AccountsManager;
import org.whispersystems.textsecuregcm.storage.Device;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.util.function.Tuple2;
import reactor.util.function.Tuples;
public class MessagesAnonymousGrpcService extends SimpleMessagesAnonymousGrpc.MessagesAnonymousImplBase {
@@ -61,8 +55,6 @@ public class MessagesAnonymousGrpcService extends SimpleMessagesAnonymousGrpc.Me
private static final SendMessageResponse SEND_MESSAGE_SUCCESS_RESPONSE = SendMessageResponse.newBuilder().build();
private static final int MAX_FETCH_ACCOUNT_CONCURRENCY = 8;
public MessagesAnonymousGrpcService(final AccountsManager accountsManager,
final RateLimiters rateLimiters,
final MessageSender messageSender,
@@ -256,7 +248,7 @@ public class MessagesAnonymousGrpcService extends SimpleMessagesAnonymousGrpc.Me
// At this point, the caller has at least superficially provided the information needed to send a multi-recipient
// message. Attempt to resolve the destination service identifiers to Signal accounts.
final Map<SealedSenderMultiRecipientMessage.Recipient, Account> resolvedRecipients =
resolveRecipients(multiRecipientMessage);
MessageUtil.resolveRecipients(accountsManager, multiRecipientMessage);
try {
messageSender.sendMultiRecipientMessage(multiRecipientMessage,
@@ -269,10 +261,7 @@ public class MessagesAnonymousGrpcService extends SimpleMessagesAnonymousGrpc.Me
final SendMultiRecipientMessageResponse.Builder responseBuilder = SendMultiRecipientMessageResponse.newBuilder();
multiRecipientMessage.getRecipients().entrySet()
.stream()
.filter(entry -> !resolvedRecipients.containsKey(entry.getValue()))
.map(entry -> ServiceIdentifier.fromLibsignal(entry.getKey()))
MessageUtil.getUnresolvedRecipients(multiRecipientMessage, resolvedRecipients).stream()
.map(ServiceIdentifierUtil::toGrpcServiceIdentifier)
.forEach(responseBuilder::addUnresolvedRecipients);
@@ -308,49 +297,13 @@ public class MessagesAnonymousGrpcService extends SimpleMessagesAnonymousGrpc.Me
// Check that the request is well-formed and doesn't contain repeated entries for the same device for the same
// recipient
validateNoDuplicateDevices(multiRecipientMessage);
if (MessageUtil.hasDuplicateDevices(multiRecipientMessage)) {
throw Status.INVALID_ARGUMENT.withDescription("Multi-recipient message contains duplicate recipient").asException();
}
return multiRecipientMessage;
}
private void validateNoDuplicateDevices(final SealedSenderMultiRecipientMessage multiRecipientMessage)
throws StatusException {
final boolean[] usedDeviceIds = new boolean[Device.MAXIMUM_DEVICE_ID + 1];
for (final SealedSenderMultiRecipientMessage.Recipient recipient : multiRecipientMessage.getRecipients().values()) {
if (recipient.getDevices().length == 1) {
// A recipient can't have repeated devices if they only have one device
continue;
}
Arrays.fill(usedDeviceIds, false);
for (final byte deviceId : recipient.getDevices()) {
if (usedDeviceIds[deviceId]) {
throw Status.INVALID_ARGUMENT.withDescription("Request contains repeated device entries").asException();
}
usedDeviceIds[deviceId] = true;
}
}
}
private Map<SealedSenderMultiRecipientMessage.Recipient, Account> resolveRecipients(final SealedSenderMultiRecipientMessage multiRecipientMessage) {
return Flux.fromIterable(multiRecipientMessage.getRecipients().entrySet())
.flatMap(serviceIdAndRecipient -> {
final ServiceIdentifier serviceIdentifier =
ServiceIdentifier.fromLibsignal(serviceIdAndRecipient.getKey());
return Mono.fromFuture(() -> accountsManager.getByServiceIdentifierAsync(serviceIdentifier))
.flatMap(Mono::justOrEmpty)
.map(account -> Tuples.of(serviceIdAndRecipient.getValue(), account));
}, MAX_FETCH_ACCOUNT_CONCURRENCY)
.collectMap(Tuple2::getT1, Tuple2::getT2)
.blockOptional()
.orElse(Collections.emptyMap());
}
private MismatchedDevices buildMismatchedDevices(final ServiceIdentifier serviceIdentifier,
org.whispersystems.textsecuregcm.controllers.MismatchedDevices mismatchedDevices) {