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

Persist messages in batches.

This commit is contained in:
Jon Chambers 2020-09-23 12:23:34 -04:00 committed by Jon Chambers
parent 6041a9d094
commit fc71ced660
6 changed files with 118 additions and 106 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -123,10 +123,8 @@ public class MessagePersister implements Managed {
do { do {
messages = messagesCache.getMessagesToPersist(accountUuid, deviceId, MESSAGE_BATCH_LIMIT); messages = messagesCache.getMessagesToPersist(accountUuid, deviceId, MESSAGE_BATCH_LIMIT);
for (final MessageProtos.Envelope message : messages) { messagesManager.persistMessages(accountNumber, accountUuid, deviceId, messages);
messagesManager.persistMessage(accountNumber, accountUuid, message, UUID.fromString(message.getServerGuid()), deviceId); messageCount += messages.size();
messageCount++;
}
} while (!messages.isEmpty()); } while (!messages.isEmpty());
queueSizeHistogram.update(messageCount); queueSizeHistogram.update(messageCount);

25
service/src/main/java/org/whispersystems/textsecuregcm/storage/Messages.java
View File

@ -1,9 +1,11 @@
package org.whispersystems.textsecuregcm.storage; package org.whispersystems.textsecuregcm.storage;
import com.codahale.metrics.Histogram;
import com.codahale.metrics.Meter; import com.codahale.metrics.Meter;
import com.codahale.metrics.MetricRegistry; import com.codahale.metrics.MetricRegistry;
import com.codahale.metrics.SharedMetricRegistries; import com.codahale.metrics.SharedMetricRegistries;
import com.codahale.metrics.Timer; import com.codahale.metrics.Timer;
import org.jdbi.v3.core.statement.PreparedBatch;
import org.whispersystems.textsecuregcm.entities.MessageProtos.Envelope; import org.whispersystems.textsecuregcm.entities.MessageProtos.Envelope;
import org.whispersystems.textsecuregcm.entities.OutgoingMessageEntity; import org.whispersystems.textsecuregcm.entities.OutgoingMessageEntity;
import org.whispersystems.textsecuregcm.storage.mappers.OutgoingMessageEntityRowMapper; import org.whispersystems.textsecuregcm.storage.mappers.OutgoingMessageEntityRowMapper;
@ -43,6 +45,7 @@ public class Messages {
private final Timer clearTimer = metricRegistry.timer(name(Messages.class, "clear" )); private final Timer clearTimer = metricRegistry.timer(name(Messages.class, "clear" ));
private final Timer vacuumTimer = metricRegistry.timer(name(Messages.class, "vacuum")); private final Timer vacuumTimer = metricRegistry.timer(name(Messages.class, "vacuum"));
private final Meter insertNullGuidMeter = metricRegistry.meter(name(Messages.class, "insertNullGuid")); private final Meter insertNullGuidMeter = metricRegistry.meter(name(Messages.class, "insertNullGuid"));
private final Histogram storeSizeHistogram = metricRegistry.histogram(name(Messages.class, "storeBatchSize"));
private final FaultTolerantDatabase database; private final FaultTolerantDatabase database;
@ -51,16 +54,18 @@ public class Messages {
this.database.getDatabase().registerRowMapper(new OutgoingMessageEntityRowMapper()); this.database.getDatabase().registerRowMapper(new OutgoingMessageEntityRowMapper());
} }
public void store(UUID guid, Envelope message, String destination, long destinationDevice) { public void store(final List<Envelope> messages, final String destination, final long destinationDevice) {
if (guid == null) { database.use(jdbi -> jdbi.useTransaction(handle -> {
try (final Timer.Context ignored = storeTimer.time()) {
final PreparedBatch batch = handle.prepareBatch("INSERT INTO messages (" + GUID + ", " + TYPE + ", " + RELAY + ", " + TIMESTAMP + ", " + SERVER_TIMESTAMP + ", " + SOURCE + ", " + SOURCE_UUID + ", " + SOURCE_DEVICE + ", " + DESTINATION + ", " + DESTINATION_DEVICE + ", " + MESSAGE + ", " + CONTENT + ") " +
"VALUES (:guid, :type, :relay, :timestamp, :server_timestamp, :source, :source_uuid, :source_device, :destination, :destination_device, :message, :content)");
for (final Envelope message : messages) {
if (message.getServerGuid() == null) {
insertNullGuidMeter.mark(); insertNullGuidMeter.mark();
} }
database.use(jdbi ->jdbi.useHandle(handle -> { batch.bind("guid", UUID.fromString(message.getServerGuid()))
try (Timer.Context ignored = storeTimer.time()) {
handle.createUpdate("INSERT INTO messages (" + GUID + ", " + TYPE + ", " + RELAY + ", " + TIMESTAMP + ", " + SERVER_TIMESTAMP + ", " + SOURCE + ", " + SOURCE_UUID + ", " + SOURCE_DEVICE + ", " + DESTINATION + ", " + DESTINATION_DEVICE + ", " + MESSAGE + ", " + CONTENT + ") " +
"VALUES (:guid, :type, :relay, :timestamp, :server_timestamp, :source, :source_uuid, :source_device, :destination, :destination_device, :message, :content)")
.bind("guid", guid)
.bind("destination", destination) .bind("destination", destination)
.bind("destination_device", destinationDevice) .bind("destination_device", destinationDevice)
.bind("type", message.getType().getNumber()) .bind("type", message.getType().getNumber())
@ -72,7 +77,11 @@ public class Messages {
.bind("source_device", message.hasSourceDevice() ? message.getSourceDevice() : null) .bind("source_device", message.hasSourceDevice() ? message.getSourceDevice() : null)
.bind("message", message.hasLegacyMessage() ? message.getLegacyMessage().toByteArray() : null) .bind("message", message.hasLegacyMessage() ? message.getLegacyMessage().toByteArray() : null)
.bind("content", message.hasContent() ? message.getContent().toByteArray() : null) .bind("content", message.hasContent() ? message.getContent().toByteArray() : null)
.execute(); .add();
}
batch.execute();
storeSizeHistogram.update(messages.size());
} }
})); }));
} }

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

@ -117,9 +117,12 @@ public class MessagesManager {
} }
} }
public void persistMessage(String destination, UUID destinationUuid, Envelope envelope, UUID messageGuid, long deviceId) { public void persistMessages(final String destination, final UUID destinationUuid, final long destinationDeviceId, final List<Envelope> messages) {
messages.store(messageGuid, envelope, destination, deviceId); this.messages.store(messages, destination, destinationDeviceId);
messagesCache.remove(destinationUuid, deviceId, messageGuid);
for (final Envelope message : messages) {
messagesCache.remove(destinationUuid, destinationDeviceId, UUID.fromString(message.getServerGuid()));
}
} }
public void addMessageAvailabilityListener(final UUID destinationUuid, final long deviceId, final MessageAvailabilityListener listener) { public void addMessageAvailabilityListener(final UUID destinationUuid, final long deviceId, final MessageAvailabilityListener listener) {

31
service/src/test/java/org/whispersystems/textsecuregcm/storage/MessagePersisterTest.java
View File

@ -5,12 +5,14 @@ import io.lettuce.core.cluster.SlotHash;
import org.apache.commons.lang3.RandomStringUtils; import org.apache.commons.lang3.RandomStringUtils;
import org.junit.Before; import org.junit.Before;
import org.junit.Test; import org.junit.Test;
import org.mockito.ArgumentCaptor;
import org.whispersystems.textsecuregcm.entities.MessageProtos; import org.whispersystems.textsecuregcm.entities.MessageProtos;
import org.whispersystems.textsecuregcm.redis.AbstractRedisClusterTest; import org.whispersystems.textsecuregcm.redis.AbstractRedisClusterTest;
import java.nio.charset.StandardCharsets; import java.nio.charset.StandardCharsets;
import java.time.Duration; import java.time.Duration;
import java.time.Instant; import java.time.Instant;
import java.util.List;
import java.util.Optional; import java.util.Optional;
import java.util.UUID; import java.util.UUID;
import java.util.concurrent.ExecutorService; import java.util.concurrent.ExecutorService;
@ -18,14 +20,15 @@ import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import static org.junit.Assert.assertEquals;
import static org.mockito.ArgumentMatchers.any; 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.atLeastOnce;
import static org.mockito.Mockito.doAnswer; 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.verify; import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when; import static org.mockito.Mockito.when;
@ -69,15 +72,17 @@ public class MessagePersisterTest extends AbstractRedisClusterTest {
doAnswer(invocation -> { doAnswer(invocation -> {
final String destination = invocation.getArgument(0, String.class); final String destination = invocation.getArgument(0, String.class);
final UUID destinationUuid = invocation.getArgument(1, UUID.class); final UUID destinationUuid = invocation.getArgument(1, UUID.class);
final MessageProtos.Envelope message = invocation.getArgument(2, MessageProtos.Envelope.class); final long deviceId = invocation.getArgument(2, Long.class);
final UUID messageGuid = invocation.getArgument(3, UUID.class); final List<MessageProtos.Envelope> messages = invocation.getArgument(3, List.class);
final long deviceId = invocation.getArgument(4, Long.class);
messagesDatabase.store(messageGuid, message, destination, deviceId); messagesDatabase.store(messages, destination, deviceId);
messagesCache.remove(destinationUuid, deviceId, messageGuid);
for (final MessageProtos.Envelope message : messages) {
messagesCache.remove(destinationUuid, deviceId, UUID.fromString(message.getServerGuid()));
}
return null; return null;
}).when(messagesManager).persistMessage(anyString(), any(UUID.class), any(MessageProtos.Envelope.class), any(UUID.class), anyLong()); }).when(messagesManager).persistMessages(anyString(), any(UUID.class), anyLong(), any());
} }
@Override @Override
@ -109,7 +114,10 @@ public class MessagePersisterTest extends AbstractRedisClusterTest {
messagePersister.persistNextQueues(now.plus(messagePersister.getPersistDelay())); messagePersister.persistNextQueues(now.plus(messagePersister.getPersistDelay()));
verify(messagesDatabase, times(messageCount)).store(any(UUID.class), any(MessageProtos.Envelope.class), eq(DESTINATION_ACCOUNT_NUMBER), eq(DESTINATION_DEVICE_ID)); final ArgumentCaptor<List<MessageProtos.Envelope>> messagesCaptor = ArgumentCaptor.forClass(List.class);
verify(messagesDatabase, atLeastOnce()).store(messagesCaptor.capture(), eq(DESTINATION_ACCOUNT_NUMBER), eq(DESTINATION_DEVICE_ID));
assertEquals(messageCount, messagesCaptor.getAllValues().stream().mapToInt(List::size).sum());
} }
@Test @Test
@ -123,7 +131,7 @@ public class MessagePersisterTest extends AbstractRedisClusterTest {
messagePersister.persistNextQueues(now); messagePersister.persistNextQueues(now);
verify(messagesDatabase, never()).store(any(UUID.class), any(MessageProtos.Envelope.class), anyString(), anyLong()); verify(messagesDatabase, never()).store(any(), anyString(), anyLong());
} }
@Test @Test
@ -151,7 +159,10 @@ public class MessagePersisterTest extends AbstractRedisClusterTest {
messagePersister.persistNextQueues(now.plus(messagePersister.getPersistDelay())); messagePersister.persistNextQueues(now.plus(messagePersister.getPersistDelay()));
verify(messagesDatabase, times(queueCount * messagesPerQueue)).store(any(UUID.class), any(MessageProtos.Envelope.class), anyString(), anyLong()); final ArgumentCaptor<List<MessageProtos.Envelope>> messagesCaptor = ArgumentCaptor.forClass(List.class);
verify(messagesDatabase, atLeastOnce()).store(messagesCaptor.capture(), anyString(), anyLong());
assertEquals(queueCount * messagesPerQueue, messagesCaptor.getAllValues().stream().mapToInt(List::size).sum());
} }
@SuppressWarnings("SameParameterValue") @SuppressWarnings("SameParameterValue")

89
service/src/test/java/org/whispersystems/textsecuregcm/tests/storage/MessagesTest.java
View File

@ -44,9 +44,8 @@ public class MessagesTest {
@Test @Test
public void testStore() throws SQLException { public void testStore() throws SQLException {
Envelope envelope = generateEnvelope(); Envelope envelope = generateEnvelope();
UUID guid = UUID.randomUUID();
messages.store(guid, envelope, "+14151112222", 1); messages.store(List.of(envelope), "+14151112222", 1);
PreparedStatement statement = db.getTestDatabase().getConnection().prepareStatement("SELECT * FROM messages WHERE destination = ?"); PreparedStatement statement = db.getTestDatabase().getConnection().prepareStatement("SELECT * FROM messages WHERE destination = ?");
statement.setString(1, "+14151112222"); statement.setString(1, "+14151112222");
@ -54,7 +53,7 @@ public class MessagesTest {
ResultSet resultSet = statement.executeQuery(); ResultSet resultSet = statement.executeQuery();
assertThat(resultSet.next()).isTrue(); assertThat(resultSet.next()).isTrue();
assertThat(resultSet.getString("guid")).isEqualTo(guid.toString()); assertThat(resultSet.getString("guid")).isEqualTo(envelope.getServerGuid());
assertThat(resultSet.getInt("type")).isEqualTo(envelope.getType().getNumber()); assertThat(resultSet.getInt("type")).isEqualTo(envelope.getType().getNumber());
assertThat(resultSet.getString("relay")).isNullOrEmpty(); assertThat(resultSet.getString("relay")).isNullOrEmpty();
assertThat(resultSet.getLong("timestamp")).isEqualTo(envelope.getTimestamp()); assertThat(resultSet.getLong("timestamp")).isEqualTo(envelope.getTimestamp());
@ -71,36 +70,29 @@ public class MessagesTest {
@Test @Test
public void testLoad() { public void testLoad() {
List<MessageToStore> inserted = new ArrayList<>(50); List<Envelope> inserted = insertRandom("+14151112222", 1);
for (int i=0;i<50;i++) { inserted.sort(Comparator.comparingLong(Envelope::getTimestamp));
MessageToStore message = generateMessageToStore();
inserted.add(message);
messages.store(message.guid, message.envelope, "+14151112222", 1);
}
inserted.sort(Comparator.comparingLong(o -> o.envelope.getTimestamp()));
List<OutgoingMessageEntity> retrieved = messages.load("+14151112222", 1); List<OutgoingMessageEntity> retrieved = messages.load("+14151112222", 1);
assertThat(retrieved.size()).isEqualTo(inserted.size()); assertThat(retrieved.size()).isEqualTo(inserted.size());
for (int i=0;i<retrieved.size();i++) { for (int i=0;i<retrieved.size();i++) {
verifyExpected(retrieved.get(i), inserted.get(i).envelope, inserted.get(i).guid); verifyExpected(retrieved.get(i), inserted.get(i), UUID.fromString(inserted.get(i).getServerGuid()));
} }
} }
@Test @Test
public void removeBySourceDestinationTimestamp() { public void removeBySourceDestinationTimestamp() {
List<MessageToStore> inserted = insertRandom("+14151112222", 1); List<Envelope> inserted = insertRandom("+14151112222", 1);
List<MessageToStore> unrelated = insertRandom("+14151114444", 3); List<Envelope> unrelated = insertRandom("+14151114444", 3);
MessageToStore toRemove = inserted.remove(new Random(System.currentTimeMillis()).nextInt(inserted.size() - 1)); Envelope toRemove = inserted.remove(new Random(System.currentTimeMillis()).nextInt(inserted.size() - 1));
Optional<OutgoingMessageEntity> removed = messages.remove("+14151112222", 1, toRemove.envelope.getSource(), toRemove.envelope.getTimestamp()); Optional<OutgoingMessageEntity> removed = messages.remove("+14151112222", 1, toRemove.getSource(), toRemove.getTimestamp());
assertThat(removed.isPresent()).isTrue(); assertThat(removed.isPresent()).isTrue();
verifyExpected(removed.get(), toRemove.envelope, toRemove.guid); verifyExpected(removed.get(), toRemove, UUID.fromString(toRemove.getServerGuid()));
verifyInTact(inserted, "+14151112222", 1); verifyInTact(inserted, "+14151112222", 1);
verifyInTact(unrelated, "+14151114444", 3); verifyInTact(unrelated, "+14151114444", 3);
@ -108,13 +100,13 @@ public class MessagesTest {
@Test @Test
public void removeByDestinationGuid() { public void removeByDestinationGuid() {
List<MessageToStore> unrelated = insertRandom("+14151113333", 2); List<Envelope> unrelated = insertRandom("+14151113333", 2);
List<MessageToStore> inserted = insertRandom("+14151112222", 1); List<Envelope> inserted = insertRandom("+14151112222", 1);
MessageToStore toRemove = inserted.remove(new Random(System.currentTimeMillis()).nextInt(inserted.size() - 1)); Envelope toRemove = inserted.remove(new Random(System.currentTimeMillis()).nextInt(inserted.size() - 1));
Optional<OutgoingMessageEntity> removed = messages.remove("+14151112222", toRemove.guid); Optional<OutgoingMessageEntity> removed = messages.remove("+14151112222", UUID.fromString(toRemove.getServerGuid()));
assertThat(removed.isPresent()).isTrue(); assertThat(removed.isPresent()).isTrue();
verifyExpected(removed.get(), toRemove.envelope, toRemove.guid); verifyExpected(removed.get(), toRemove, UUID.fromString(toRemove.getServerGuid()));
verifyInTact(inserted, "+14151112222", 1); verifyInTact(inserted, "+14151112222", 1);
verifyInTact(unrelated, "+14151113333", 2); verifyInTact(unrelated, "+14151113333", 2);
@ -122,10 +114,10 @@ public class MessagesTest {
@Test @Test
public void removeByDestinationRowId() { public void removeByDestinationRowId() {
List<MessageToStore> unrelatedInserted = insertRandom("+14151111111", 1); List<Envelope> unrelatedInserted = insertRandom("+14151111111", 1);
List<MessageToStore> inserted = insertRandom("+14151112222", 1); List<Envelope> inserted = insertRandom("+14151112222", 1);
inserted.sort(Comparator.comparingLong(o -> o.envelope.getTimestamp())); inserted.sort(Comparator.comparingLong(Envelope::getTimestamp));
List<OutgoingMessageEntity> retrieved = messages.load("+14151112222", 1); List<OutgoingMessageEntity> retrieved = messages.load("+14151112222", 1);
@ -141,9 +133,8 @@ public class MessagesTest {
@Test @Test
public void testLoadEmpty() { public void testLoadEmpty() {
List<MessageToStore> inserted = insertRandom("+14151112222", 1); insertRandom("+14151112222", 1);
List<OutgoingMessageEntity> loaded = messages.load("+14159999999", 1); assertThat(messages.load("+14159999999", 1).isEmpty()).isTrue();
assertThat(loaded.isEmpty()).isTrue();
} }
@Test @Test
@ -151,7 +142,7 @@ public class MessagesTest {
insertRandom("+14151112222", 1); insertRandom("+14151112222", 1);
insertRandom("+14151112222", 2); insertRandom("+14151112222", 2);
List<MessageToStore> unrelated = insertRandom("+14151111111", 1); List<Envelope> unrelated = insertRandom("+14151111111", 1);
messages.clear("+14151112222"); messages.clear("+14151112222");
@ -163,9 +154,9 @@ public class MessagesTest {
@Test @Test
public void testClearDestinationDevice() { public void testClearDestinationDevice() {
insertRandom("+14151112222", 1); insertRandom("+14151112222", 1);
List<MessageToStore> inserted = insertRandom("+14151112222", 2); List<Envelope> inserted = insertRandom("+14151112222", 2);
List<MessageToStore> unrelated = insertRandom("+14151111111", 1); List<Envelope> unrelated = insertRandom("+14151111111", 1);
messages.clear("+14151112222", 1); messages.clear("+14151112222", 1);
@ -177,33 +168,32 @@ public class MessagesTest {
@Test @Test
public void testVacuum() { public void testVacuum() {
List<MessageToStore> inserted = insertRandom("+14151112222", 2); List<Envelope> inserted = insertRandom("+14151112222", 2);
messages.vacuum(); messages.vacuum();
verifyInTact(inserted, "+14151112222", 2); verifyInTact(inserted, "+14151112222", 2);
} }
private List<MessageToStore> insertRandom(String destination, int destinationDevice) { private List<Envelope> insertRandom(String destination, int destinationDevice) {
List<MessageToStore> inserted = new ArrayList<>(50); List<Envelope> inserted = new ArrayList<>(50);
for (int i=0;i<50;i++) { for (int i=0;i<50;i++) {
MessageToStore message = generateMessageToStore(); inserted.add(generateEnvelope());
inserted.add(message);
messages.store(message.guid, message.envelope, destination, destinationDevice);
} }
messages.store(inserted, destination, destinationDevice);
return inserted; return inserted;
} }
private void verifyInTact(List<MessageToStore> inserted, String destination, int destinationDevice) { private void verifyInTact(List<Envelope> inserted, String destination, int destinationDevice) {
inserted.sort(Comparator.comparingLong(o -> o.envelope.getTimestamp())); inserted.sort(Comparator.comparingLong(Envelope::getTimestamp));
List<OutgoingMessageEntity> retrieved = messages.load(destination, destinationDevice); List<OutgoingMessageEntity> retrieved = messages.load(destination, destinationDevice);
assertThat(retrieved.size()).isEqualTo(inserted.size()); assertThat(retrieved.size()).isEqualTo(inserted.size());
for (int i=0;i<retrieved.size();i++) { for (int i=0;i<retrieved.size();i++) {
verifyExpected(retrieved.get(i), inserted.get(i).envelope, inserted.get(i).guid); verifyExpected(retrieved.get(i), inserted.get(i), UUID.fromString(inserted.get(i).getServerGuid()));
} }
} }
@ -220,10 +210,6 @@ public class MessagesTest {
assertThat(retrieved.getSourceDevice()).isEqualTo(inserted.getSourceDevice()); assertThat(retrieved.getSourceDevice()).isEqualTo(inserted.getSourceDevice());
} }
private MessageToStore generateMessageToStore() {
return new MessageToStore(UUID.randomUUID(), generateEnvelope());
}
private Envelope generateEnvelope() { private Envelope generateEnvelope() {
Random random = new Random(); Random random = new Random();
byte[] content = new byte[256]; byte[] content = new byte[256];
@ -233,6 +219,7 @@ public class MessagesTest {
Arrays.fill(legacy, (byte)random.nextInt(255)); Arrays.fill(legacy, (byte)random.nextInt(255));
return Envelope.newBuilder() return Envelope.newBuilder()
.setServerGuid(UUID.randomUUID().toString())
.setSourceDevice(random.nextInt(10000)) .setSourceDevice(random.nextInt(10000))
.setSource("testSource" + random.nextInt()) .setSource("testSource" + random.nextInt())
.setTimestamp(serialTimestamp++) .setTimestamp(serialTimestamp++)
@ -243,14 +230,4 @@ public class MessagesTest {
.setServerGuid(UUID.randomUUID().toString()) .setServerGuid(UUID.randomUUID().toString())
.build(); .build();
} }
private static class MessageToStore {
private final UUID guid;
private final Envelope envelope;
private MessageToStore(UUID guid, Envelope envelope) {
this.guid = guid;
this.envelope = envelope;
}
}
} }

30
service/src/test/java/org/whispersystems/textsecuregcm/websocket/WebSocketConnectionIntegrationTest.java
View File

@ -63,6 +63,8 @@ public class WebSocketConnectionIntegrationTest extends AbstractRedisClusterTest
private WebSocketClient webSocketClient; private WebSocketClient webSocketClient;
private WebSocketConnection webSocketConnection; private WebSocketConnection webSocketConnection;
private long serialTimestamp = System.currentTimeMillis();
@Before @Before
public void setupAccountsDao() { public void setupAccountsDao() {
} }
@ -108,14 +110,19 @@ public class WebSocketConnectionIntegrationTest extends AbstractRedisClusterTest
final List<MessageProtos.Envelope> expectedMessages = new ArrayList<>(persistedMessageCount + cachedMessageCount); final List<MessageProtos.Envelope> expectedMessages = new ArrayList<>(persistedMessageCount + cachedMessageCount);
for (int i = 0; i < persistedMessageCount; i++) { {
final UUID messageGuid = UUID.randomUUID(); final List<MessageProtos.Envelope> persistedMessages = new ArrayList<>(persistedMessageCount);
final MessageProtos.Envelope envelope = generateRandomMessage(messageGuid);
messages.store(messageGuid, envelope, account.getNumber(), device.getId()); for (int i = 0; i < persistedMessageCount; i++) {
final MessageProtos.Envelope envelope = generateRandomMessage(UUID.randomUUID());
persistedMessages.add(envelope);
expectedMessages.add(envelope.toBuilder().clearServerGuid().build()); expectedMessages.add(envelope.toBuilder().clearServerGuid().build());
} }
messages.store(persistedMessages, account.getNumber(), device.getId());
}
for (int i = 0; i < cachedMessageCount; i++) { for (int i = 0; i < cachedMessageCount; i++) {
final UUID messageGuid = UUID.randomUUID(); final UUID messageGuid = UUID.randomUUID();
final MessageProtos.Envelope envelope = generateRandomMessage(messageGuid); final MessageProtos.Envelope envelope = generateRandomMessage(messageGuid);
@ -172,9 +179,14 @@ public class WebSocketConnectionIntegrationTest extends AbstractRedisClusterTest
final int persistedMessageCount = 207; final int persistedMessageCount = 207;
final int cachedMessageCount = 173; final int cachedMessageCount = 173;
{
final List<MessageProtos.Envelope> persistedMessages = new ArrayList<>(persistedMessageCount);
for (int i = 0; i < persistedMessageCount; i++) { for (int i = 0; i < persistedMessageCount; i++) {
final UUID messageGuid = UUID.randomUUID(); persistedMessages.add(generateRandomMessage(UUID.randomUUID()));
messages.store(messageGuid, generateRandomMessage(messageGuid), account.getNumber(), device.getId()); }
messages.store(persistedMessages, account.getNumber(), device.getId());
} }
for (int i = 0; i < cachedMessageCount; i++) { for (int i = 0; i < cachedMessageCount; i++) {
@ -191,9 +203,11 @@ public class WebSocketConnectionIntegrationTest extends AbstractRedisClusterTest
} }
private MessageProtos.Envelope generateRandomMessage(final UUID messageGuid) { private MessageProtos.Envelope generateRandomMessage(final UUID messageGuid) {
final long timestamp = serialTimestamp++;
return MessageProtos.Envelope.newBuilder() return MessageProtos.Envelope.newBuilder()
.setTimestamp(System.currentTimeMillis()) .setTimestamp(timestamp)
.setServerTimestamp(System.currentTimeMillis()) .setServerTimestamp(timestamp)
.setContent(ByteString.copyFromUtf8(RandomStringUtils.randomAlphanumeric(256))) .setContent(ByteString.copyFromUtf8(RandomStringUtils.randomAlphanumeric(256)))
.setType(MessageProtos.Envelope.Type.CIPHERTEXT) .setType(MessageProtos.Envelope.Type.CIPHERTEXT)
.setServerGuid(messageGuid.toString()) .setServerGuid(messageGuid.toString())