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

Write synchronously to the cache cluster.

This commit is contained in:
Jon Chambers 2020-06-17 11:51:47 -04:00 committed by Jon Chambers
parent c745fe7778
commit 7454e55693
8 changed files with 24 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
service/src/main/java/org/whispersystems/textsecuregcm/limits/LockingRateLimiter.java
View File

@ -47,7 +47,7 @@ public class LockingRateLimiter extends RateLimiter {
final String lockName = getLockName(key); final String lockName = getLockName(key);
jedis.del(lockName); jedis.del(lockName);
cacheCluster.useWriteCluster(connection -> connection.async().del(lockName)); cacheCluster.useWriteCluster(connection -> connection.sync().del(lockName));
} }
} }
@ -59,7 +59,7 @@ public class LockingRateLimiter extends RateLimiter {
if (acquiredLock) { if (acquiredLock) {
// TODO Restore the NX flag when the cluster becomes the primary source of truth // TODO Restore the NX flag when the cluster becomes the primary source of truth
cacheCluster.useWriteCluster(connection -> connection.async().set(lockName, "L", SetArgs.Builder.ex(10))); cacheCluster.useWriteCluster(connection -> connection.sync().set(lockName, "L", SetArgs.Builder.ex(10)));
} }
return acquiredLock; return acquiredLock;

4
service/src/main/java/org/whispersystems/textsecuregcm/limits/RateLimiter.java
View File

@ -96,7 +96,7 @@ public class RateLimiter {
final String bucketName = getBucketName(key); final String bucketName = getBucketName(key);
jedis.del(bucketName); jedis.del(bucketName);
cacheCluster.useWriteCluster(connection -> connection.async().del(bucketName)); cacheCluster.useWriteCluster(connection -> connection.sync().del(bucketName));
} }
} }
@ -107,7 +107,7 @@ public class RateLimiter {
final int level = (int) Math.ceil((bucketSize / leakRatePerMillis) / 1000); final int level = (int) Math.ceil((bucketSize / leakRatePerMillis) / 1000);
jedis.setex(bucketName, level, serialized); jedis.setex(bucketName, level, serialized);
cacheCluster.useWriteCluster(connection -> connection.async().setex(bucketName, level, serialized)); cacheCluster.useWriteCluster(connection -> connection.sync().setex(bucketName, level, serialized));
} catch (JsonProcessingException e) { } catch (JsonProcessingException e) {
throw new IllegalArgumentException(e); throw new IllegalArgumentException(e);
} }

7
service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountsManager.java
View File

@ -23,6 +23,7 @@ import com.codahale.metrics.Timer;
import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper; import com.fasterxml.jackson.databind.ObjectMapper;
import io.lettuce.core.cluster.api.async.RedisAdvancedClusterAsyncCommands; import io.lettuce.core.cluster.api.async.RedisAdvancedClusterAsyncCommands;
import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.whispersystems.textsecuregcm.auth.AmbiguousIdentifier; import org.whispersystems.textsecuregcm.auth.AmbiguousIdentifier;
@ -162,10 +163,10 @@ public class AccountsManager {
jedis.set(accountEntityKey, accountJson); jedis.set(accountEntityKey, accountJson);
cacheCluster.useWriteCluster(connection -> { cacheCluster.useWriteCluster(connection -> {
RedisAdvancedClusterAsyncCommands<String, String> asyncCommands = connection.async(); final RedisAdvancedClusterCommands<String, String> commands = connection.sync();
asyncCommands.set(accountMapKey, account.getUuid().toString()); commands.set(accountMapKey, account.getUuid().toString());
asyncCommands.set(accountEntityKey, accountJson); commands.set(accountEntityKey, accountJson);
}); });
} catch (JsonProcessingException e) { } catch (JsonProcessingException e) {
throw new IllegalStateException(e); throw new IllegalStateException(e);

4
service/src/main/java/org/whispersystems/textsecuregcm/storage/ActiveUserCounter.java
View File

@ -66,7 +66,7 @@ public class ActiveUserCounter extends AccountDatabaseCrawlerListener {
public void onCrawlStart() { public void onCrawlStart() {
try (Jedis jedis = jedisPool.getWriteResource()) { try (Jedis jedis = jedisPool.getWriteResource()) {
jedis.del(TALLY_KEY); jedis.del(TALLY_KEY);
cacheCluster.useWriteCluster(connection -> connection.async().del(TALLY_KEY)); cacheCluster.useWriteCluster(connection -> connection.sync().del(TALLY_KEY));
} }
} }
@ -186,7 +186,7 @@ public class ActiveUserCounter extends AccountDatabaseCrawlerListener {
final String tallyJson = mapper.writeValueAsString(activeUserTally); final String tallyJson = mapper.writeValueAsString(activeUserTally);
jedis.set(TALLY_KEY, tallyJson); jedis.set(TALLY_KEY, tallyJson);
cacheCluster.useWriteCluster(connection -> connection.async().set(TALLY_KEY, tallyJson)); cacheCluster.useWriteCluster(connection -> connection.sync().set(TALLY_KEY, tallyJson));
} catch (JsonProcessingException e) { } catch (JsonProcessingException e) {
throw new IllegalArgumentException(e); throw new IllegalArgumentException(e);
} catch (IOException e) { } catch (IOException e) {

4
service/src/main/java/org/whispersystems/textsecuregcm/storage/PendingAccountsManager.java
View File

@ -78,7 +78,7 @@ public class PendingAccountsManager {
final String verificationCodeJson = mapper.writeValueAsString(code); final String verificationCodeJson = mapper.writeValueAsString(code);
jedis.set(key, verificationCodeJson); jedis.set(key, verificationCodeJson);
cacheCluster.useWriteCluster(connection -> connection.async().set(key, verificationCodeJson)); cacheCluster.useWriteCluster(connection -> connection.sync().set(key, verificationCodeJson));
} catch (JsonProcessingException e) { } catch (JsonProcessingException e) {
throw new IllegalArgumentException(e); throw new IllegalArgumentException(e);
} }
@ -104,7 +104,7 @@ public class PendingAccountsManager {
final String key = CACHE_PREFIX + number; final String key = CACHE_PREFIX + number;
jedis.del(key); jedis.del(key);
cacheCluster.useWriteCluster(connection -> connection.async().del(key)); cacheCluster.useWriteCluster(connection -> connection.sync().del(key));
} }
} }
} }

4
service/src/main/java/org/whispersystems/textsecuregcm/storage/PendingDevicesManager.java
View File

@ -77,7 +77,7 @@ public class PendingDevicesManager {
final String verificationCodeJson = mapper.writeValueAsString(code); final String verificationCodeJson = mapper.writeValueAsString(code);
jedis.set(key, verificationCodeJson); jedis.set(key, verificationCodeJson);
cacheCluster.useWriteCluster(connection -> connection.async().set(key, verificationCodeJson)); cacheCluster.useWriteCluster(connection -> connection.sync().set(key, verificationCodeJson));
} catch (JsonProcessingException e) { } catch (JsonProcessingException e) {
throw new IllegalArgumentException(e); throw new IllegalArgumentException(e);
} }
@ -103,7 +103,7 @@ public class PendingDevicesManager {
final String key = CACHE_PREFIX + number; final String key = CACHE_PREFIX + number;
jedis.del(key); jedis.del(key);
cacheCluster.useWriteCluster(connection -> connection.async().del(key)); cacheCluster.useWriteCluster(connection -> connection.sync().del(key));
} }
} }

4
service/src/main/java/org/whispersystems/textsecuregcm/storage/ProfilesManager.java
View File

@ -63,7 +63,7 @@ public class ProfilesManager {
final String profileJson = mapper.writeValueAsString(profile); final String profileJson = mapper.writeValueAsString(profile);
jedis.hset(key, profile.getVersion(), profileJson); jedis.hset(key, profile.getVersion(), profileJson);
cacheCluster.useWriteCluster(connection -> connection.async().hset(key, profile.getVersion(), profileJson)); cacheCluster.useWriteCluster(connection -> connection.sync().hset(key, profile.getVersion(), profileJson));
} catch (JsonProcessingException e) { } catch (JsonProcessingException e) {
throw new IllegalArgumentException(e); throw new IllegalArgumentException(e);
} }
@ -92,7 +92,7 @@ public class ProfilesManager {
final String key = CACHE_PREFIX + uuid.toString(); final String key = CACHE_PREFIX + uuid.toString();
jedis.del(key); jedis.del(key);
cacheCluster.useWriteCluster(connection -> connection.async().del(key)); cacheCluster.useWriteCluster(connection -> connection.sync().del(key));
} }
} }
} }

16
service/src/main/java/org/whispersystems/textsecuregcm/storage/UsernamesManager.java
View File

@ -3,7 +3,7 @@ package org.whispersystems.textsecuregcm.storage;
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 io.lettuce.core.cluster.api.async.RedisAdvancedClusterAsyncCommands; import io.lettuce.core.cluster.api.sync.RedisAdvancedClusterCommands;
import org.slf4j.Logger; import org.slf4j.Logger;
import org.slf4j.LoggerFactory; import org.slf4j.LoggerFactory;
import org.whispersystems.textsecuregcm.experiment.Experiment; import org.whispersystems.textsecuregcm.experiment.Experiment;
@ -129,11 +129,11 @@ public class UsernamesManager {
jedis.set(usernameMapKey, uuid.toString()); jedis.set(usernameMapKey, uuid.toString());
cacheCluster.useWriteCluster(connection -> { cacheCluster.useWriteCluster(connection -> {
final RedisAdvancedClusterAsyncCommands<String, String> asyncCommands = connection.async(); final RedisAdvancedClusterCommands<String, String> commands = connection.sync();
maybeOldUsername.ifPresent(oldUsername -> asyncCommands.del(getUsernameMapKey(oldUsername))); maybeOldUsername.ifPresent(oldUsername -> commands.del(getUsernameMapKey(oldUsername)));
asyncCommands.set(uuidMapKey, username); commands.set(uuidMapKey, username);
asyncCommands.set(usernameMapKey, uuid.toString()); commands.set(usernameMapKey, uuid.toString());
}); });
} catch (JedisException e) { } catch (JedisException e) {
if (required) throw e; if (required) throw e;
@ -187,10 +187,10 @@ public class UsernamesManager {
jedis.del(uuidMapKey); jedis.del(uuidMapKey);
cacheCluster.useWriteCluster(connection -> { cacheCluster.useWriteCluster(connection -> {
final RedisAdvancedClusterAsyncCommands<String, String> asyncCommands = connection.async(); final RedisAdvancedClusterCommands<String, String> commands = connection.sync();
asyncCommands.del(usernameMapKey); commands.del(usernameMapKey);
asyncCommands.del(uuidMapKey); commands.del(uuidMapKey);
}); });
}); });
} }