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

Mirror persistence operations from the new persister to the old persister.

This commit is contained in:
Jon Chambers 2020-09-01 16:20:30 -04:00 committed by Jon Chambers
parent 2b50367d7f
commit 8409986ef5
5 changed files with 24 additions and 15 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
service/src/main/java/org/whispersystems/textsecuregcm/WhisperServerService.java
View File

@ -333,7 +333,7 @@ public class WhisperServerService extends Application<WhisperServerConfiguration
RecaptchaClient recaptchaClient = new RecaptchaClient(config.getRecaptchaConfiguration().getSecret()); RecaptchaClient recaptchaClient = new RecaptchaClient(config.getRecaptchaConfiguration().getSecret());
MessagePersister messagePersister = new MessagePersister(messagesClient, messagesManager, pubSubManager, pushSender, accountsManager, featureFlagsManager, config.getMessageCacheConfiguration().getPersistDelayMinutes(), TimeUnit.MINUTES); MessagePersister messagePersister = new MessagePersister(messagesClient, messagesManager, pubSubManager, pushSender, accountsManager, featureFlagsManager, config.getMessageCacheConfiguration().getPersistDelayMinutes(), TimeUnit.MINUTES);
RedisClusterMessagePersister clusterMessagePersister = new RedisClusterMessagePersister(clusterMessagesCache, messages, pubSubManager, pushSender, accountsManager, featureFlagsManager, Duration.ofMinutes(config.getMessageCacheConfiguration().getPersistDelayMinutes())); RedisClusterMessagePersister clusterMessagePersister = new RedisClusterMessagePersister(clusterMessagesCache, messagesManager, pubSubManager, pushSender, accountsManager, featureFlagsManager, Duration.ofMinutes(config.getMessageCacheConfiguration().getPersistDelayMinutes()));
DirectoryReconciliationClient directoryReconciliationClient = new DirectoryReconciliationClient(config.getDirectoryConfiguration().getDirectoryServerConfiguration()); DirectoryReconciliationClient directoryReconciliationClient = new DirectoryReconciliationClient(config.getDirectoryConfiguration().getDirectoryServerConfiguration());

2
service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagePersister.java
View File

@ -169,7 +169,7 @@ public class MessagePersister implements Managed, Runnable {
envelope = envelope.toBuilder().clearServerGuid().build(); envelope = envelope.toBuilder().clearServerGuid().build();
messagesManager.persistMessage(key.getAddress(), destinationUuid, envelope, guid, key.getDeviceId(), score); messagesManager.persistMessage(key.getAddress(), destinationUuid, envelope, guid, key.getDeviceId());
} catch (InvalidProtocolBufferException e) { } catch (InvalidProtocolBufferException e) {
logger.error("Error parsing envelope", e); logger.error("Error parsing envelope", e);
} }

6
service/src/main/java/org/whispersystems/textsecuregcm/storage/MessagesManager.java
View File

@ -120,11 +120,9 @@ public class MessagesManager {
} }
} }
public void persistMessage(String destination, UUID destinationUuid, Envelope envelope, UUID messageGuid, long deviceId, long id) { public void persistMessage(String destination, UUID destinationUuid, Envelope envelope, UUID messageGuid, long deviceId) {
messages.store(messageGuid, envelope, destination, deviceId); messages.store(messageGuid, envelope, destination, deviceId);
delete(destination, destinationUuid, deviceId, messageGuid);
final Optional<OutgoingMessageEntity> maybeRemovedMessage = messagesCache.remove(destination, destinationUuid, deviceId, id);
removeByIdExperiment.compareSupplierResult(maybeRemovedMessage, () -> clusterMessagesCache.remove(destination, destinationUuid, deviceId, id));
} }
public void addMessageAvailabilityListener(final UUID destinationUuid, final long deviceId, final MessageAvailabilityListener listener) { public void addMessageAvailabilityListener(final UUID destinationUuid, final long deviceId, final MessageAvailabilityListener listener) {

12
service/src/main/java/org/whispersystems/textsecuregcm/storage/RedisClusterMessagePersister.java
View File

@ -26,7 +26,7 @@ import static com.codahale.metrics.MetricRegistry.name;
public class RedisClusterMessagePersister implements Managed { public class RedisClusterMessagePersister implements Managed {
private final RedisClusterMessagesCache messagesCache; private final RedisClusterMessagesCache messagesCache;
private final Messages messagesDatabase; private final MessagesManager messagesManager;
private final PubSubManager pubSubManager; private final PubSubManager pubSubManager;
private final PushSender pushSender; private final PushSender pushSender;
private final AccountsManager accountsManager; private final AccountsManager accountsManager;
@ -51,9 +51,9 @@ public class RedisClusterMessagePersister implements Managed {
private static final Logger logger = LoggerFactory.getLogger(RedisClusterMessagePersister.class); private static final Logger logger = LoggerFactory.getLogger(RedisClusterMessagePersister.class);
public RedisClusterMessagePersister(final RedisClusterMessagesCache messagesCache, final Messages messagesDatabase, final PubSubManager pubSubManager, final PushSender pushSender, final AccountsManager accountsManager, final FeatureFlagsManager featureFlagsManager, final Duration persistDelay) { public RedisClusterMessagePersister(final RedisClusterMessagesCache messagesCache, final MessagesManager messagesManager, final PubSubManager pubSubManager, final PushSender pushSender, final AccountsManager accountsManager, final FeatureFlagsManager featureFlagsManager, final Duration persistDelay) {
this.messagesCache = messagesCache; this.messagesCache = messagesCache;
this.messagesDatabase = messagesDatabase; this.messagesManager = messagesManager;
this.pubSubManager = pubSubManager; this.pubSubManager = pubSubManager;
this.pushSender = pushSender; this.pushSender = pushSender;
this.accountsManager = accountsManager; this.accountsManager = accountsManager;
@ -143,11 +143,7 @@ public class RedisClusterMessagePersister implements Managed {
messages = messagesCache.getMessagesToPersist(accountUuid, deviceId, MESSAGE_BATCH_LIMIT); messages = messagesCache.getMessagesToPersist(accountUuid, deviceId, MESSAGE_BATCH_LIMIT);
for (final MessageProtos.Envelope message : messages) { for (final MessageProtos.Envelope message : messages) {
final UUID uuid = UUID.fromString(message.getServerGuid()); messagesManager.persistMessage(accountNumber, accountUuid, message, UUID.fromString(message.getServerGuid()), deviceId);
messagesDatabase.store(uuid, message, accountNumber, deviceId);
messagesCache.remove(accountNumber, accountUuid, deviceId, uuid);
messageCount++; messageCount++;
} }
} while (!messages.isEmpty()); } while (!messages.isEmpty());

17
service/src/test/java/org/whispersystems/textsecuregcm/storage/RedisClusterMessagePersisterTest.java
View File

@ -22,6 +22,7 @@ import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong; import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString; import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.ArgumentMatchers.eq; import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.never; import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times; import static org.mockito.Mockito.times;
@ -48,6 +49,7 @@ public class RedisClusterMessagePersisterTest extends AbstractRedisClusterTest {
public void setUp() throws Exception { public void setUp() throws Exception {
super.setUp(); super.setUp();
final MessagesManager messagesManager = mock(MessagesManager.class);
final FeatureFlagsManager featureFlagsManager = mock(FeatureFlagsManager.class); final FeatureFlagsManager featureFlagsManager = mock(FeatureFlagsManager.class);
when(featureFlagsManager.isFeatureFlagActive(RedisClusterMessagePersister.ENABLE_PERSISTENCE_FLAG)).thenReturn(true); when(featureFlagsManager.isFeatureFlagActive(RedisClusterMessagePersister.ENABLE_PERSISTENCE_FLAG)).thenReturn(true);
@ -62,7 +64,20 @@ public class RedisClusterMessagePersisterTest extends AbstractRedisClusterTest {
notificationExecutorService = Executors.newSingleThreadExecutor(); notificationExecutorService = Executors.newSingleThreadExecutor();
messagesCache = new RedisClusterMessagesCache(getRedisCluster(), notificationExecutorService); messagesCache = new RedisClusterMessagesCache(getRedisCluster(), notificationExecutorService);
messagePersister = new RedisClusterMessagePersister(messagesCache, messagesDatabase, pubSubManager, mock(PushSender.class), accountsManager, featureFlagsManager, PERSIST_DELAY); messagePersister = new RedisClusterMessagePersister(messagesCache, messagesManager, pubSubManager, mock(PushSender.class), accountsManager, featureFlagsManager, PERSIST_DELAY);
doAnswer(invocation -> {
final String destination = invocation.getArgument(0, String.class);
final UUID destinationUuid = invocation.getArgument(1, UUID.class);
final MessageProtos.Envelope message = invocation.getArgument(2, MessageProtos.Envelope.class);
final UUID messageGuid = invocation.getArgument(3, UUID.class);
final long deviceId = invocation.getArgument(4, Long.class);
messagesDatabase.store(messageGuid, message, destination, deviceId);
messagesCache.remove(destination, destinationUuid, deviceId, messageGuid);
return null;
}).when(messagesManager).persistMessage(anyString(), any(UUID.class), any(MessageProtos.Envelope.class), any(UUID.class), anyLong());
} }
@Override @Override