Update libsignal-service-java

lib/build.gradle.kts

@@ -14,7 +14,7 @@ repositories {
 }
 
 dependencies {
-    api("com.github.turasa:signal-service-java:2.15.3_unofficial_24")
+    api("com.github.turasa:signal-service-java:2.15.3_unofficial_25")
     implementation("com.google.protobuf:protobuf-javalite:3.10.0")
     implementation("org.bouncycastle:bcprov-jdk15on:1.69")
     implementation("org.slf4j:slf4j-api:1.7.30")

lib/src/main/java/org/asamk/signal/manager/Manager.java

@@ -65,14 +65,12 @@ import org.signal.libsignal.metadata.ProtocolLegacyMessageException;
 import org.signal.libsignal.metadata.ProtocolNoSessionException;
 import org.signal.libsignal.metadata.ProtocolUntrustedIdentityException;
 import org.signal.libsignal.metadata.SelfSendException;
-import org.signal.libsignal.metadata.certificate.CertificateValidator;
 import org.signal.storageservice.protos.groups.GroupChange;
 import org.signal.storageservice.protos.groups.local.DecryptedGroup;
 import org.signal.zkgroup.InvalidInputException;
 import org.signal.zkgroup.VerificationFailedException;
 import org.signal.zkgroup.groups.GroupMasterKey;
 import org.signal.zkgroup.groups.GroupSecretParams;
-import org.signal.zkgroup.profiles.ClientZkProfileOperations;
 import org.signal.zkgroup.profiles.ProfileKey;
 import org.signal.zkgroup.profiles.ProfileKeyCredential;
 import org.slf4j.Logger;
@@ -86,19 +84,12 @@ import org.whispersystems.libsignal.state.PreKeyRecord;
 import org.whispersystems.libsignal.state.SignedPreKeyRecord;
 import org.whispersystems.libsignal.util.Pair;
 import org.whispersystems.libsignal.util.guava.Optional;
-import org.whispersystems.signalservice.api.SignalServiceAccountManager;
+import org.whispersystems.signalservice.api.InvalidMessageStructureException;
-import org.whispersystems.signalservice.api.SignalServiceMessagePipe;
-import org.whispersystems.signalservice.api.SignalServiceMessageReceiver;
-import org.whispersystems.signalservice.api.SignalServiceMessageSender;
 import org.whispersystems.signalservice.api.SignalSessionLock;
 import org.whispersystems.signalservice.api.crypto.ContentHint;
-import org.whispersystems.signalservice.api.crypto.SignalServiceCipher;
 import org.whispersystems.signalservice.api.crypto.UntrustedIdentityException;
-import org.whispersystems.signalservice.api.groupsv2.ClientZkOperations;
 import org.whispersystems.signalservice.api.groupsv2.GroupLinkNotActiveException;
-import org.whispersystems.signalservice.api.groupsv2.GroupsV2Api;
 import org.whispersystems.signalservice.api.groupsv2.GroupsV2AuthorizationString;
-import org.whispersystems.signalservice.api.groupsv2.GroupsV2Operations;
 import org.whispersystems.signalservice.api.messages.SendMessageResult;
 import org.whispersystems.signalservice.api.messages.SignalServiceAttachment;
 import org.whispersystems.signalservice.api.messages.SignalServiceAttachmentPointer;
@@ -133,9 +124,8 @@ import org.whispersystems.signalservice.api.push.exceptions.UnregisteredUserExce
 import org.whispersystems.signalservice.api.util.DeviceNameUtil;
 import org.whispersystems.signalservice.api.util.InvalidNumberException;
 import org.whispersystems.signalservice.api.util.PhoneNumberFormatter;
-import org.whispersystems.signalservice.api.util.SleepTimer;
-import org.whispersystems.signalservice.api.util.UptimeSleepTimer;
 import org.whispersystems.signalservice.api.util.UuidUtil;
+import org.whispersystems.signalservice.api.websocket.WebSocketUnavailableException;
 import org.whispersystems.signalservice.internal.contacts.crypto.Quote;
 import org.whispersystems.signalservice.internal.contacts.crypto.UnauthenticatedQuoteException;
 import org.whispersystems.signalservice.internal.contacts.crypto.UnauthenticatedResponseException;
@@ -182,23 +172,13 @@ public class Manager implements Closeable {
 
     private final static Logger logger = LoggerFactory.getLogger(Manager.class);
 
-    private final CertificateValidator certificateValidator;
-
     private final ServiceEnvironmentConfig serviceEnvironmentConfig;
-    private final String userAgent;
+    private final SignalDependencies dependencies;
 
     private SignalAccount account;
-    private final SignalServiceAccountManager accountManager;
-    private final GroupsV2Api groupsV2Api;
-    private final GroupsV2Operations groupsV2Operations;
-    private final SignalServiceMessageReceiver messageReceiver;
-    private final ClientZkProfileOperations clientZkProfileOperations;
 
     private final ExecutorService executor = Executors.newCachedThreadPool();
 
-    private SignalServiceMessagePipe messagePipe = null;
-    private SignalServiceMessagePipe unidentifiedMessagePipe = null;
-
     private final UnidentifiedAccessHelper unidentifiedAccessHelper;
     private final ProfileHelper profileHelper;
     private final GroupV2Helper groupV2Helper;
@@ -224,42 +204,19 @@ public class Manager implements Closeable {
     ) {
         this.account = account;
         this.serviceEnvironmentConfig = serviceEnvironmentConfig;
-        this.certificateValidator = new CertificateValidator(serviceEnvironmentConfig.getUnidentifiedSenderTrustRoot());
-        this.userAgent = userAgent;
-        this.groupsV2Operations = capabilities.isGv2() ? new GroupsV2Operations(ClientZkOperations.create(
-                serviceEnvironmentConfig.getSignalServiceConfiguration())) : null;
-        final SleepTimer timer = new UptimeSleepTimer();
-        this.accountManager = new SignalServiceAccountManager(serviceEnvironmentConfig.getSignalServiceConfiguration(),
-                new DynamicCredentialsProvider(account.getUuid(),
-                        account.getUsername(),
-                        account.getPassword(),
-                        account.getDeviceId()),
-                userAgent,
-                groupsV2Operations,
-                ServiceConfig.AUTOMATIC_NETWORK_RETRY,
-                timer);
-        this.groupsV2Api = accountManager.getGroupsV2Api();
-        final var keyBackupService = accountManager.getKeyBackupService(ServiceConfig.getIasKeyStore(),
-                serviceEnvironmentConfig.getKeyBackupConfig().getEnclaveName(),
-                serviceEnvironmentConfig.getKeyBackupConfig().getServiceId(),
-                serviceEnvironmentConfig.getKeyBackupConfig().getMrenclave(),
-                10);
 
-        this.pinHelper = new PinHelper(keyBackupService);
+        final var credentialsProvider = new DynamicCredentialsProvider(account.getUuid(),
-        this.clientZkProfileOperations = capabilities.isGv2()
-                ? ClientZkOperations.create(serviceEnvironmentConfig.getSignalServiceConfiguration())
-                .getProfileOperations()
-                : null;
-        this.messageReceiver = new SignalServiceMessageReceiver(serviceEnvironmentConfig.getSignalServiceConfiguration(),
-                account.getUuid(),
                 account.getUsername(),
                 account.getPassword(),
-                account.getDeviceId(),
+                account.getDeviceId());
+        this.dependencies = new SignalDependencies(account.getSelfAddress(),
+                serviceEnvironmentConfig,
                 userAgent,
-                null,
+                credentialsProvider,
-                timer,
+                account.getSignalProtocolStore(),
-                clientZkProfileOperations,
+                executor,
-                ServiceConfig.AUTOMATIC_NETWORK_RETRY);
+                sessionLock);
+        this.pinHelper = new PinHelper(dependencies.getKeyBackupService());
 
         this.unidentifiedAccessHelper = new UnidentifiedAccessHelper(account::getProfileKey,
                 account.getProfileStore()::getProfileKey,
@@ -267,14 +224,14 @@ public class Manager implements Closeable {
                 this::getSenderCertificate);
         this.profileHelper = new ProfileHelper(account.getProfileStore()::getProfileKey,
                 unidentifiedAccessHelper::getAccessFor,
-                unidentified -> unidentified ? getOrCreateUnidentifiedMessagePipe() : getOrCreateMessagePipe(),
+                dependencies::getProfileService,
-                () -> messageReceiver,
+                dependencies::getMessageReceiver,
                 this::resolveSignalServiceAddress);
         this.groupV2Helper = new GroupV2Helper(this::getRecipientProfileKeyCredential,
                 this::getRecipientProfile,
                 account::getSelfRecipientId,
-                groupsV2Operations,
+                dependencies.getGroupsV2Operations(),
-                groupsV2Api,
+                dependencies.getGroupsV2Api(),
                 this::getGroupAuthForToday,
                 this::resolveSignalServiceAddress);
         this.avatarStore = new AvatarStore(pathConfig.getAvatarsPath());
@@ -350,11 +307,11 @@ public class Manager implements Closeable {
                         days);
             }
         }
-        if (accountManager.getPreKeysCount() < ServiceConfig.PREKEY_MINIMUM_COUNT) {
+        if (dependencies.getAccountManager().getPreKeysCount() < ServiceConfig.PREKEY_MINIMUM_COUNT) {
             refreshPreKeys();
         }
         if (account.getUuid() == null) {
-            account.setUuid(accountManager.getOwnUuid());
+            account.setUuid(dependencies.getAccountManager().getOwnUuid());
         }
         updateAccountAttributes();
     }
@@ -376,17 +333,18 @@ public class Manager implements Closeable {
     }
 
     public void updateAccountAttributes() throws IOException {
-        accountManager.setAccountAttributes(account.getEncryptedDeviceName(),
+        dependencies.getAccountManager()
-                null,
+                .setAccountAttributes(account.getEncryptedDeviceName(),
-                account.getLocalRegistrationId(),
+                        null,
-                true,
+                        account.getLocalRegistrationId(),
-                // set legacy pin only if no KBS master key is set
+                        true,
-                account.getPinMasterKey() == null ? account.getRegistrationLockPin() : null,
+                        // set legacy pin only if no KBS master key is set
-                account.getPinMasterKey() == null ? null : account.getPinMasterKey().deriveRegistrationLock(),
+                        account.getPinMasterKey() == null ? account.getRegistrationLockPin() : null,
-                account.getSelfUnidentifiedAccessKey(),
+                        account.getPinMasterKey() == null ? null : account.getPinMasterKey().deriveRegistrationLock(),
-                account.isUnrestrictedUnidentifiedAccess(),
+                        account.getSelfUnidentifiedAccessKey(),
-                capabilities,
+                        account.isUnrestrictedUnidentifiedAccess(),
-                account.isDiscoverableByPhoneNumber());
+                        capabilities,
+                        account.isDiscoverableByPhoneNumber());
     }
 
     /**
@@ -418,13 +376,14 @@ public class Manager implements Closeable {
         try (final var streamDetails = avatar == null
                 ? avatarStore.retrieveProfileAvatar(getSelfAddress())
                 : avatar.isPresent() ? Utils.createStreamDetailsFromFile(avatar.get()) : null) {
-            accountManager.setVersionedProfile(account.getUuid(),
+            dependencies.getAccountManager()
-                    account.getProfileKey(),
+                    .setVersionedProfile(account.getUuid(),
-                    newProfile.getInternalServiceName(),
+                            account.getProfileKey(),
-                    newProfile.getAbout() == null ? "" : newProfile.getAbout(),
+                            newProfile.getInternalServiceName(),
-                    newProfile.getAboutEmoji() == null ? "" : newProfile.getAboutEmoji(),
+                            newProfile.getAbout() == null ? "" : newProfile.getAbout(),
-                    Optional.absent(),
+                            newProfile.getAboutEmoji() == null ? "" : newProfile.getAboutEmoji(),
-                    streamDetails);
+                            Optional.absent(),
+                            streamDetails);
         }
 
         if (avatar != null) {
@@ -447,19 +406,19 @@ public class Manager implements Closeable {
         // When setting an empty GCM id, the Signal-Server also sets the fetchesMessages property to false.
         // If this is the master device, other users can't send messages to this number anymore.
         // If this is a linked device, other users can still send messages, but this device doesn't receive them anymore.
-        accountManager.setGcmId(Optional.absent());
+        dependencies.getAccountManager().setGcmId(Optional.absent());
 
         account.setRegistered(false);
     }
 
     public void deleteAccount() throws IOException {
-        accountManager.deleteAccount();
+        dependencies.getAccountManager().deleteAccount();
 
         account.setRegistered(false);
     }
 
     public List<Device> getLinkedDevices() throws IOException {
-        var devices = accountManager.getDevices();
+        var devices = dependencies.getAccountManager().getDevices();
         account.setMultiDevice(devices.size() > 1);
         var identityKey = account.getIdentityKeyPair().getPrivateKey();
         return devices.stream().map(d -> {
@@ -476,8 +435,8 @@ public class Manager implements Closeable {
     }
 
     public void removeLinkedDevices(int deviceId) throws IOException {
-        accountManager.removeDevice(deviceId);
+        dependencies.getAccountManager().removeDevice(deviceId);
-        var devices = accountManager.getDevices();
+        var devices = dependencies.getAccountManager().getDevices();
         account.setMultiDevice(devices.size() > 1);
     }
 
@@ -489,13 +448,14 @@ public class Manager implements Closeable {
 
     private void addDevice(String deviceIdentifier, ECPublicKey deviceKey) throws IOException, InvalidKeyException {
         var identityKeyPair = getIdentityKeyPair();
-        var verificationCode = accountManager.getNewDeviceVerificationCode();
+        var verificationCode = dependencies.getAccountManager().getNewDeviceVerificationCode();
 
-        accountManager.addDevice(deviceIdentifier,
+        dependencies.getAccountManager()
-                deviceKey,
+                .addDevice(deviceIdentifier,
-                identityKeyPair,
+                        deviceKey,
-                Optional.of(account.getProfileKey().serialize()),
+                        identityKeyPair,
-                verificationCode);
+                        Optional.of(account.getProfileKey().serialize()),
+                        verificationCode);
         account.setMultiDevice(true);
     }
 
@@ -513,7 +473,7 @@ public class Manager implements Closeable {
             account.setRegistrationLockPin(pin.get(), masterKey);
         } else {
             // Remove legacy registration lock
-            accountManager.removeRegistrationLockV1();
+            dependencies.getAccountManager().removeRegistrationLockV1();
 
             // Remove KBS Pin
             pinHelper.removeRegistrationLockPin();
@@ -527,7 +487,7 @@ public class Manager implements Closeable {
         final var identityKeyPair = getIdentityKeyPair();
         var signedPreKeyRecord = generateSignedPreKey(identityKeyPair);
 
-        accountManager.setPreKeys(identityKeyPair.getPublicKey(), signedPreKeyRecord, oneTimePreKeys);
+        dependencies.getAccountManager().setPreKeys(identityKeyPair.getPublicKey(), signedPreKeyRecord, oneTimePreKeys);
     }
 
     private List<PreKeyRecord> generatePreKeys() {
@@ -548,39 +508,6 @@ public class Manager implements Closeable {
         return record;
     }
 
-    private SignalServiceMessagePipe getOrCreateMessagePipe() {
-        if (messagePipe == null) {
-            messagePipe = messageReceiver.createMessagePipe();
-        }
-        return messagePipe;
-    }
-
-    private SignalServiceMessagePipe getOrCreateUnidentifiedMessagePipe() {
-        if (unidentifiedMessagePipe == null) {
-            unidentifiedMessagePipe = messageReceiver.createUnidentifiedMessagePipe();
-        }
-        return unidentifiedMessagePipe;
-    }
-
-    private SignalServiceMessageSender createMessageSender() {
-        return new SignalServiceMessageSender(serviceEnvironmentConfig.getSignalServiceConfiguration(),
-                account.getUuid(),
-                account.getUsername(),
-                account.getPassword(),
-                account.getDeviceId(),
-                account.getSignalProtocolStore(),
-                sessionLock,
-                userAgent,
-                account.isMultiDevice(),
-                Optional.fromNullable(messagePipe),
-                Optional.fromNullable(unidentifiedMessagePipe),
-                Optional.absent(),
-                clientZkProfileOperations,
-                executor,
-                ServiceConfig.MAX_ENVELOPE_SIZE,
-                ServiceConfig.AUTOMATIC_NETWORK_RETRY);
-    }
-
     public Profile getRecipientProfile(
             RecipientId recipientId
     ) {
@@ -1180,14 +1107,15 @@ public class Manager implements Closeable {
     ) throws IOException {
         final var today = currentTimeDays();
         // Returns credentials for the next 7 days
-        final var credentials = groupsV2Api.getCredentials(today);
+        final var credentials = dependencies.getGroupsV2Api().getCredentials(today);
         // TODO cache credentials until they expire
         var authCredentialResponse = credentials.get(today);
         try {
-            return groupsV2Api.getGroupsV2AuthorizationString(account.getUuid(),
+            return dependencies.getGroupsV2Api()
-                    today,
+                    .getGroupsV2AuthorizationString(account.getUuid(),
-                    groupSecretParams,
+                            today,
-                    authCredentialResponse);
+                            groupSecretParams,
+                            authCredentialResponse);
         } catch (VerificationFailedException e) {
             throw new IOException(e);
         }
@@ -1264,9 +1192,10 @@ public class Manager implements Closeable {
                 List.of(messageId),
                 System.currentTimeMillis());
 
-        createMessageSender().sendReceipt(remoteAddress,
+        dependencies.getMessageSender()
-                unidentifiedAccessHelper.getAccessFor(resolveRecipient(remoteAddress)),
+                .sendReceipt(remoteAddress,
-                receiptMessage);
+                        unidentifiedAccessHelper.getAccessFor(resolveRecipient(remoteAddress)),
+                        receiptMessage);
     }
 
     public Pair<Long, List<SendMessageResult>> sendMessage(
@@ -1277,7 +1206,7 @@ public class Manager implements Closeable {
             var attachmentStreams = AttachmentUtils.getSignalServiceAttachments(attachments);
 
             // Upload attachments here, so we only upload once even for multiple recipients
-            var messageSender = createMessageSender();
+            var messageSender = dependencies.getMessageSender();
             var attachmentPointers = new ArrayList<SignalServiceAttachment>(attachmentStreams.size());
             for (var attachment : attachmentStreams) {
                 if (attachment.isStream()) {
@@ -1351,11 +1280,6 @@ public class Manager implements Closeable {
         }
     }
 
-    public String getContactName(String number) throws InvalidNumberException {
-        var contact = account.getContactStore().getContact(canonicalizeAndResolveRecipient(number));
-        return contact == null || contact.getName() == null ? "" : contact.getName();
-    }
-
     public void setContactName(String number, String name) throws InvalidNumberException, NotMasterDeviceException {
         if (!account.isMasterDevice()) {
             throw new NotMasterDeviceException();
@@ -1442,7 +1366,7 @@ public class Manager implements Closeable {
     public String uploadStickerPack(File path) throws IOException, StickerPackInvalidException {
         var manifest = StickerUtils.getSignalServiceStickerManifestUpload(path);
 
-        var messageSender = createMessageSender();
+        var messageSender = dependencies.getMessageSender();
 
         var packKey = KeyUtils.createStickerUploadKey();
         var packIdString = messageSender.uploadStickerManifest(manifest, packKey);
@@ -1536,9 +1460,9 @@ public class Manager implements Closeable {
         byte[] certificate;
         try {
             if (account.isPhoneNumberShared()) {
-                certificate = accountManager.getSenderCertificate();
+                certificate = dependencies.getAccountManager().getSenderCertificate();
             } else {
-                certificate = accountManager.getSenderCertificateForPhoneNumberPrivacy();
+                certificate = dependencies.getAccountManager().getSenderCertificateForPhoneNumberPrivacy();
             }
         } catch (IOException e) {
             logger.warn("Failed to get sender certificate, ignoring: {}", e.getMessage());
@@ -1549,7 +1473,7 @@ public class Manager implements Closeable {
     }
 
     private void sendSyncMessage(SignalServiceSyncMessage message) throws IOException, UntrustedIdentityException {
-        var messageSender = createMessageSender();
+        var messageSender = dependencies.getMessageSender();
         messageSender.sendSyncMessage(message, unidentifiedAccessHelper.getAccessForSync());
     }
 
@@ -1604,9 +1528,10 @@ public class Manager implements Closeable {
 
     private Map<String, UUID> getRegisteredUsers(final Set<String> numbers) throws IOException {
         try {
-            return accountManager.getRegisteredUsers(ServiceConfig.getIasKeyStore(),
+            return dependencies.getAccountManager()
-                    numbers,
+                    .getRegisteredUsers(ServiceConfig.getIasKeyStore(),
-                    serviceEnvironmentConfig.getCdsMrenclave());
+                            numbers,
+                            serviceEnvironmentConfig.getCdsMrenclave());
         } catch (Quote.InvalidQuoteFormatException | UnauthenticatedQuoteException | SignatureException | UnauthenticatedResponseException | InvalidKeyException e) {
             throw new IOException(e);
         }
@@ -1623,7 +1548,7 @@ public class Manager implements Closeable {
     ) throws IOException, UntrustedIdentityException {
         final var timestamp = System.currentTimeMillis();
         var message = new SignalServiceTypingMessage(action.toSignalService(), timestamp, Optional.absent());
-        var messageSender = createMessageSender();
+        var messageSender = dependencies.getMessageSender();
         for (var recipientId : recipientIds) {
             final var address = resolveSignalServiceAddress(recipientId);
             messageSender.sendTyping(address, unidentifiedAccessHelper.getAccessFor(recipientId), message);
@@ -1638,7 +1563,7 @@ public class Manager implements Closeable {
         final var message = new SignalServiceTypingMessage(action.toSignalService(),
                 timestamp,
                 Optional.of(groupId.serialize()));
-        final var messageSender = createMessageSender();
+        final var messageSender = dependencies.getMessageSender();
         final var recipientIdList = new ArrayList<>(g.getMembersWithout(account.getSelfRecipientId()));
         final var addresses = recipientIdList.stream()
                 .map(this::resolveSignalServiceAddress)
@@ -1651,14 +1576,13 @@ public class Manager implements Closeable {
     ) throws IOException {
         final var timestamp = System.currentTimeMillis();
         messageBuilder.withTimestamp(timestamp);
-        getOrCreateMessagePipe();
+
-        getOrCreateUnidentifiedMessagePipe();
         SignalServiceDataMessage message = null;
         try {
             message = messageBuilder.build();
             if (message.getGroupContext().isPresent()) {
                 try {
-                    var messageSender = createMessageSender();
+                    var messageSender = dependencies.getMessageSender();
                     final var isRecipientUpdate = false;
                     final var recipientIdList = new ArrayList<>(recipientIds);
                     final var addresses = recipientIdList.stream()
@@ -1668,7 +1592,9 @@ public class Manager implements Closeable {
                             unidentifiedAccessHelper.getAccessFor(recipientIdList),
                             isRecipientUpdate,
                             ContentHint.DEFAULT,
-                            message);
+                            message,
+                            sendResult -> logger.trace("Partial message send result: {}", sendResult.isSuccess()),
+                            () -> false);
 
                     for (var r : result) {
                         if (r.getIdentityFailure() != null) {
@@ -1712,8 +1638,6 @@ public class Manager implements Closeable {
     ) throws IOException {
         final var timestamp = System.currentTimeMillis();
         messageBuilder.withTimestamp(timestamp);
-        getOrCreateMessagePipe();
-        getOrCreateUnidentifiedMessagePipe();
         final var recipientId = account.getSelfRecipientId();
 
         final var contact = account.getContactStore().getContact(recipientId);
@@ -1726,7 +1650,7 @@ public class Manager implements Closeable {
     }
 
     private SendMessageResult sendSelfMessage(SignalServiceDataMessage message) throws IOException {
-        var messageSender = createMessageSender();
+        var messageSender = dependencies.getMessageSender();
 
         var recipientId = account.getSelfRecipientId();
 
@@ -1741,12 +1665,7 @@ public class Manager implements Closeable {
         var syncMessage = SignalServiceSyncMessage.forSentTranscript(transcript);
 
         try {
-            var startTime = System.currentTimeMillis();
+            return messageSender.sendSyncMessage(syncMessage, unidentifiedAccess);
-            messageSender.sendSyncMessage(syncMessage, unidentifiedAccess);
-            return SendMessageResult.success(recipient,
-                    unidentifiedAccess.isPresent(),
-                    false,
-                    System.currentTimeMillis() - startTime);
         } catch (UntrustedIdentityException e) {
             return SendMessageResult.identityFailure(recipient, e.getIdentityKey());
         }
@@ -1755,7 +1674,7 @@ public class Manager implements Closeable {
     private SendMessageResult sendMessage(
             RecipientId recipientId, SignalServiceDataMessage message
     ) throws IOException {
-        var messageSender = createMessageSender();
+        var messageSender = dependencies.getMessageSender();
 
         final var address = resolveSignalServiceAddress(recipientId);
         try {
@@ -1777,7 +1696,7 @@ public class Manager implements Closeable {
     }
 
     private SendMessageResult sendNullMessage(RecipientId recipientId) throws IOException {
-        var messageSender = createMessageSender();
+        var messageSender = dependencies.getMessageSender();
 
         final var address = resolveSignalServiceAddress(recipientId);
         try {
@@ -1793,12 +1712,8 @@ public class Manager implements Closeable {
         }
     }
 
-    private SignalServiceContent decryptMessage(SignalServiceEnvelope envelope) throws InvalidMetadataMessageException, ProtocolInvalidMessageException, ProtocolDuplicateMessageException, ProtocolLegacyMessageException, ProtocolInvalidKeyIdException, InvalidMetadataVersionException, ProtocolInvalidVersionException, ProtocolNoSessionException, ProtocolInvalidKeyException, SelfSendException, UnsupportedDataMessageException, ProtocolUntrustedIdentityException {
+    private SignalServiceContent decryptMessage(SignalServiceEnvelope envelope) throws InvalidMetadataMessageException, ProtocolInvalidMessageException, ProtocolDuplicateMessageException, ProtocolLegacyMessageException, ProtocolInvalidKeyIdException, InvalidMetadataVersionException, ProtocolInvalidVersionException, ProtocolNoSessionException, ProtocolInvalidKeyException, SelfSendException, UnsupportedDataMessageException, ProtocolUntrustedIdentityException, InvalidMessageStructureException {
-        var cipher = new SignalServiceCipher(account.getSelfAddress(),
+        return dependencies.getCipher().decrypt(envelope);
-                account.getSignalProtocolStore(),
-                sessionLock,
-                certificateValidator);
-        return cipher.decrypt(envelope);
     }
 
     private void handleEndSession(RecipientId recipientId) {
@@ -2082,7 +1997,8 @@ public class Manager implements Closeable {
 
         Set<HandleAction> queuedActions = null;
 
-        final var messagePipe = getOrCreateMessagePipe();
+        final var signalWebSocket = dependencies.getSignalWebSocket();
+        signalWebSocket.connect();
 
         var hasCaughtUpWithOldMessages = false;
 
@@ -2094,7 +2010,7 @@ public class Manager implements Closeable {
             account.setLastReceiveTimestamp(System.currentTimeMillis());
             logger.debug("Checking for new message from server");
             try {
-                var result = messagePipe.readOrEmpty(timeout, unit, envelope1 -> {
+                var result = signalWebSocket.readOrEmpty(unit.toMillis(timeout), envelope1 -> {
                     final var recipientId = envelope1.hasSource()
                             ? resolveRecipient(envelope1.getSourceIdentifier())
                             : null;
@@ -2132,6 +2048,10 @@ public class Manager implements Closeable {
                 } else {
                     throw e;
                 }
+            } catch (WebSocketUnavailableException e) {
+                logger.debug("Pipe unexpectedly unavailable, connecting");
+                signalWebSocket.connect();
+                continue;
             } catch (TimeoutException e) {
                 if (returnOnTimeout) return;
                 continue;
@@ -2602,12 +2522,11 @@ public class Manager implements Closeable {
     private void retrieveGroupV2Avatar(
             GroupSecretParams groupSecretParams, String cdnKey, OutputStream outputStream
     ) throws IOException {
-        var groupOperations = groupsV2Operations.forGroup(groupSecretParams);
+        var groupOperations = dependencies.getGroupsV2Operations().forGroup(groupSecretParams);
 
         var tmpFile = IOUtils.createTempFile();
-        try (InputStream input = messageReceiver.retrieveGroupsV2ProfileAvatar(cdnKey,
+        try (InputStream input = dependencies.getMessageReceiver()
-                tmpFile,
+                .retrieveGroupsV2ProfileAvatar(cdnKey, tmpFile, ServiceConfig.AVATAR_DOWNLOAD_FAILSAFE_MAX_SIZE)) {
-                ServiceConfig.AVATAR_DOWNLOAD_FAILSAFE_MAX_SIZE)) {
             var encryptedData = IOUtils.readFully(input);
 
             var decryptedData = groupOperations.decryptAvatar(encryptedData);
@@ -2627,10 +2546,11 @@ public class Manager implements Closeable {
             String avatarPath, ProfileKey profileKey, OutputStream outputStream
     ) throws IOException {
         var tmpFile = IOUtils.createTempFile();
-        try (var input = messageReceiver.retrieveProfileAvatar(avatarPath,
+        try (var input = dependencies.getMessageReceiver()
-                tmpFile,
+                .retrieveProfileAvatar(avatarPath,
-                profileKey,
+                        tmpFile,
-                ServiceConfig.AVATAR_DOWNLOAD_FAILSAFE_MAX_SIZE)) {
+                        profileKey,
+                        ServiceConfig.AVATAR_DOWNLOAD_FAILSAFE_MAX_SIZE)) {
             // Use larger buffer size to prevent AssertionError: Need: 12272 but only have: 8192 ...
             IOUtils.copyStream(input, outputStream, (int) ServiceConfig.AVATAR_DOWNLOAD_FAILSAFE_MAX_SIZE);
         } finally {
@@ -2678,7 +2598,8 @@ public class Manager implements Closeable {
     private InputStream retrieveAttachmentAsStream(
             SignalServiceAttachmentPointer pointer, File tmpFile
     ) throws IOException, InvalidMessageException, MissingConfigurationException {
-        return messageReceiver.retrieveAttachment(pointer, tmpFile, ServiceConfig.MAX_ATTACHMENT_SIZE);
+        return dependencies.getMessageReceiver()
+                .retrieveAttachment(pointer, tmpFile, ServiceConfig.MAX_ATTACHMENT_SIZE);
     }
 
     void sendGroups() throws IOException, UntrustedIdentityException {
@@ -2979,7 +2900,10 @@ public class Manager implements Closeable {
     }
 
     private void enqueueJob(Job job) {
-        var context = new Context(account, accountManager, messageReceiver, stickerPackStore);
+        var context = new Context(account,
+                dependencies.getAccountManager(),
+                dependencies.getMessageReceiver(),
+                stickerPackStore);
         job.run(context);
     }
 
@@ -2991,15 +2915,7 @@ public class Manager implements Closeable {
     void close(boolean closeAccount) throws IOException {
         executor.shutdown();
 
-        if (messagePipe != null) {
+        dependencies.getSignalWebSocket().disconnect();
-            messagePipe.shutdown();
-            messagePipe = null;
-        }
-
-        if (unidentifiedMessagePipe != null) {
-            unidentifiedMessagePipe.shutdown();
-            unidentifiedMessagePipe = null;
-        }
 
         if (closeAccount && account != null) {
             account.close();

lib/src/main/java/org/asamk/signal/manager/ProvisioningManager.java

@@ -31,8 +31,6 @@ import org.whispersystems.signalservice.api.groupsv2.GroupsV2Operations;
 import org.whispersystems.signalservice.api.push.SignalServiceAddress;
 import org.whispersystems.signalservice.api.push.exceptions.AuthorizationFailedException;
 import org.whispersystems.signalservice.api.util.DeviceNameUtil;
-import org.whispersystems.signalservice.api.util.SleepTimer;
-import org.whispersystems.signalservice.api.util.UptimeSleepTimer;
 import org.whispersystems.signalservice.internal.util.DynamicCredentialsProvider;
 
 import java.io.File;
@@ -61,7 +59,6 @@ public class ProvisioningManager {
         tempIdentityKey = KeyUtils.generateIdentityKeyPair();
         registrationId = KeyHelper.generateRegistrationId(false);
         password = KeyUtils.createPassword();
-        final SleepTimer timer = new UptimeSleepTimer();
         GroupsV2Operations groupsV2Operations;
         try {
             groupsV2Operations = new GroupsV2Operations(ClientZkOperations.create(serviceEnvironmentConfig.getSignalServiceConfiguration()));
@@ -72,8 +69,7 @@ public class ProvisioningManager {
                 new DynamicCredentialsProvider(null, null, password, SignalServiceAddress.DEFAULT_DEVICE_ID),
                 userAgent,
                 groupsV2Operations,
-                ServiceConfig.AUTOMATIC_NETWORK_RETRY,
+                ServiceConfig.AUTOMATIC_NETWORK_RETRY);
-                timer);
     }
 
     public static ProvisioningManager init(
@@ -162,7 +158,7 @@ public class ProvisioningManager {
         }
     }
 
-    private boolean canRelinkExistingAccount(final String number) throws UserAlreadyExists, IOException {
+    private boolean canRelinkExistingAccount(final String number) throws IOException {
         final SignalAccount signalAccount;
         try {
             signalAccount = SignalAccount.load(pathConfig.getDataPath(), number, false);

lib/src/main/java/org/asamk/signal/manager/RegistrationManager.java

@@ -33,8 +33,6 @@ import org.whispersystems.signalservice.api.groupsv2.ClientZkOperations;
 import org.whispersystems.signalservice.api.groupsv2.GroupsV2Operations;
 import org.whispersystems.signalservice.api.kbs.MasterKey;
 import org.whispersystems.signalservice.api.push.SignalServiceAddress;
-import org.whispersystems.signalservice.api.util.SleepTimer;
-import org.whispersystems.signalservice.api.util.UptimeSleepTimer;
 import org.whispersystems.signalservice.api.util.UuidUtil;
 import org.whispersystems.signalservice.internal.push.LockedException;
 import org.whispersystems.signalservice.internal.push.VerifyAccountResponse;
@@ -68,7 +66,6 @@ public class RegistrationManager implements Closeable {
         this.serviceEnvironmentConfig = serviceEnvironmentConfig;
         this.userAgent = userAgent;
 
-        final SleepTimer timer = new UptimeSleepTimer();
         GroupsV2Operations groupsV2Operations;
         try {
             groupsV2Operations = new GroupsV2Operations(ClientZkOperations.create(serviceEnvironmentConfig.getSignalServiceConfiguration()));
@@ -81,8 +78,7 @@ public class RegistrationManager implements Closeable {
                         null, account.getUsername(), account.getPassword(), SignalServiceAddress.DEFAULT_DEVICE_ID),
                 userAgent,
                 groupsV2Operations,
-                ServiceConfig.AUTOMATIC_NETWORK_RETRY,
+                ServiceConfig.AUTOMATIC_NETWORK_RETRY);
-                timer);
         final var keyBackupService = accountManager.getKeyBackupService(ServiceConfig.getIasKeyStore(),
                 serviceEnvironmentConfig.getKeyBackupConfig().getEnclaveName(),
                 serviceEnvironmentConfig.getKeyBackupConfig().getServiceId(),

lib/src/main/java/org/asamk/signal/manager/SignalDependencies.java

@@ -0,0 +1,150 @@
+package org.asamk.signal.manager;
+
+import org.asamk.signal.manager.config.ServiceConfig;
+import org.asamk.signal.manager.config.ServiceEnvironmentConfig;
+import org.signal.libsignal.metadata.certificate.CertificateValidator;
+import org.signal.zkgroup.profiles.ClientZkProfileOperations;
+import org.whispersystems.libsignal.util.guava.Optional;
+import org.whispersystems.signalservice.api.KeyBackupService;
+import org.whispersystems.signalservice.api.SignalServiceAccountManager;
+import org.whispersystems.signalservice.api.SignalServiceDataStore;
+import org.whispersystems.signalservice.api.SignalServiceMessageReceiver;
+import org.whispersystems.signalservice.api.SignalServiceMessageSender;
+import org.whispersystems.signalservice.api.SignalSessionLock;
+import org.whispersystems.signalservice.api.SignalWebSocket;
+import org.whispersystems.signalservice.api.crypto.SignalServiceCipher;
+import org.whispersystems.signalservice.api.groupsv2.ClientZkOperations;
+import org.whispersystems.signalservice.api.groupsv2.GroupsV2Api;
+import org.whispersystems.signalservice.api.groupsv2.GroupsV2Operations;
+import org.whispersystems.signalservice.api.push.SignalServiceAddress;
+import org.whispersystems.signalservice.api.services.ProfileService;
+import org.whispersystems.signalservice.api.util.SleepTimer;
+import org.whispersystems.signalservice.api.util.UptimeSleepTimer;
+import org.whispersystems.signalservice.api.websocket.WebSocketFactory;
+import org.whispersystems.signalservice.internal.util.DynamicCredentialsProvider;
+import org.whispersystems.signalservice.internal.websocket.WebSocketConnection;
+
+import java.util.concurrent.ExecutorService;
+
+import static org.asamk.signal.manager.config.ServiceConfig.capabilities;
+
+public class SignalDependencies {
+
+    private final SignalServiceAccountManager accountManager;
+    private final GroupsV2Api groupsV2Api;
+    private final GroupsV2Operations groupsV2Operations;
+
+    private final SignalWebSocket signalWebSocket;
+    private final SignalServiceMessageReceiver messageReceiver;
+    private final SignalServiceMessageSender messageSender;
+
+    private final KeyBackupService keyBackupService;
+    private final ProfileService profileService;
+    private final SignalServiceCipher cipher;
+
+    public SignalDependencies(
+            final SignalServiceAddress selfAddress,
+            final ServiceEnvironmentConfig serviceEnvironmentConfig,
+            final String userAgent,
+            final DynamicCredentialsProvider credentialsProvider,
+            final SignalServiceDataStore dataStore,
+            final ExecutorService executor,
+            final SignalSessionLock sessionLock
+    ) {
+        this.groupsV2Operations = capabilities.isGv2() ? new GroupsV2Operations(ClientZkOperations.create(
+                serviceEnvironmentConfig.getSignalServiceConfiguration())) : null;
+        final SleepTimer timer = new UptimeSleepTimer();
+        this.accountManager = new SignalServiceAccountManager(serviceEnvironmentConfig.getSignalServiceConfiguration(),
+                credentialsProvider,
+                userAgent,
+                groupsV2Operations,
+                ServiceConfig.AUTOMATIC_NETWORK_RETRY);
+        this.groupsV2Api = accountManager.getGroupsV2Api();
+        this.keyBackupService = accountManager.getKeyBackupService(ServiceConfig.getIasKeyStore(),
+                serviceEnvironmentConfig.getKeyBackupConfig().getEnclaveName(),
+                serviceEnvironmentConfig.getKeyBackupConfig().getServiceId(),
+                serviceEnvironmentConfig.getKeyBackupConfig().getMrenclave(),
+                10);
+        final ClientZkProfileOperations clientZkProfileOperations = capabilities.isGv2() ? ClientZkOperations.create(
+                serviceEnvironmentConfig.getSignalServiceConfiguration()).getProfileOperations() : null;
+        this.messageReceiver = new SignalServiceMessageReceiver(serviceEnvironmentConfig.getSignalServiceConfiguration(),
+                credentialsProvider,
+                userAgent,
+                clientZkProfileOperations,
+                ServiceConfig.AUTOMATIC_NETWORK_RETRY);
+
+        final var healthMonitor = new SignalWebSocketHealthMonitor(timer);
+        final WebSocketFactory webSocketFactory = new WebSocketFactory() {
+            @Override
+            public WebSocketConnection createWebSocket() {
+                return new WebSocketConnection("normal",
+                        serviceEnvironmentConfig.getSignalServiceConfiguration(),
+                        Optional.of(credentialsProvider),
+                        userAgent,
+                        healthMonitor);
+            }
+
+            @Override
+            public WebSocketConnection createUnidentifiedWebSocket() {
+                return new WebSocketConnection("unidentified",
+                        serviceEnvironmentConfig.getSignalServiceConfiguration(),
+                        Optional.absent(),
+                        userAgent,
+                        healthMonitor);
+            }
+        };
+        this.signalWebSocket = new SignalWebSocket(webSocketFactory);
+        healthMonitor.monitor(signalWebSocket);
+        this.profileService = new ProfileService(clientZkProfileOperations, messageReceiver, signalWebSocket);
+
+        final var certificateValidator = new CertificateValidator(serviceEnvironmentConfig.getUnidentifiedSenderTrustRoot());
+        this.cipher = new SignalServiceCipher(selfAddress, dataStore, sessionLock, certificateValidator);
+        this.messageSender = new SignalServiceMessageSender(serviceEnvironmentConfig.getSignalServiceConfiguration(),
+                credentialsProvider,
+                dataStore,
+                sessionLock,
+                userAgent,
+                signalWebSocket,
+                Optional.absent(),
+                clientZkProfileOperations,
+                executor,
+                ServiceConfig.MAX_ENVELOPE_SIZE,
+                ServiceConfig.AUTOMATIC_NETWORK_RETRY);
+    }
+
+    public SignalServiceAccountManager getAccountManager() {
+        return accountManager;
+    }
+
+    public GroupsV2Api getGroupsV2Api() {
+        return groupsV2Api;
+    }
+
+    public GroupsV2Operations getGroupsV2Operations() {
+        return groupsV2Operations;
+    }
+
+    public SignalWebSocket getSignalWebSocket() {
+        return signalWebSocket;
+    }
+
+    public SignalServiceMessageReceiver getMessageReceiver() {
+        return messageReceiver;
+    }
+
+    public SignalServiceMessageSender getMessageSender() {
+        return messageSender;
+    }
+
+    public KeyBackupService getKeyBackupService() {
+        return keyBackupService;
+    }
+
+    public ProfileService getProfileService() {
+        return profileService;
+    }
+
+    public SignalServiceCipher getCipher() {
+        return cipher;
+    }
+}

lib/src/main/java/org/asamk/signal/manager/SignalWebSocketHealthMonitor.java

@@ -0,0 +1,200 @@
+package org.asamk.signal.manager;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.whispersystems.libsignal.util.guava.Preconditions;
+import org.whispersystems.signalservice.api.SignalWebSocket;
+import org.whispersystems.signalservice.api.util.SleepTimer;
+import org.whispersystems.signalservice.api.websocket.HealthMonitor;
+import org.whispersystems.signalservice.api.websocket.WebSocketConnectionState;
+import org.whispersystems.signalservice.internal.websocket.WebSocketConnection;
+
+import java.util.Arrays;
+import java.util.concurrent.TimeUnit;
+
+import io.reactivex.rxjava3.schedulers.Schedulers;
+
+/**
+ * Monitors the health of the identified and unidentified WebSockets. If either one appears to be
+ * unhealthy, will trigger restarting both.
+ * <p>
+ * The monitor is also responsible for sending heartbeats/keep-alive messages to prevent
+ * timeouts.
+ */
+public final class SignalWebSocketHealthMonitor implements HealthMonitor {
+
+    private final static Logger logger = LoggerFactory.getLogger(SignalWebSocketHealthMonitor.class);
+
+    private static final long KEEP_ALIVE_SEND_CADENCE = TimeUnit.SECONDS.toMillis(WebSocketConnection.KEEPALIVE_TIMEOUT_SECONDS);
+    private static final long MAX_TIME_SINCE_SUCCESSFUL_KEEP_ALIVE = KEEP_ALIVE_SEND_CADENCE * 3;
+
+    private SignalWebSocket signalWebSocket;
+    private final SleepTimer sleepTimer;
+
+    private volatile KeepAliveSender keepAliveSender;
+
+    private final HealthState identified = new HealthState();
+    private final HealthState unidentified = new HealthState();
+
+    public SignalWebSocketHealthMonitor(SleepTimer sleepTimer) {
+        this.sleepTimer = sleepTimer;
+    }
+
+    public void monitor(SignalWebSocket signalWebSocket) {
+        Preconditions.checkNotNull(signalWebSocket);
+        Preconditions.checkArgument(this.signalWebSocket == null, "monitor can only be called once");
+
+        this.signalWebSocket = signalWebSocket;
+
+        //noinspection ResultOfMethodCallIgnored
+        signalWebSocket.getWebSocketState()
+                .subscribeOn(Schedulers.computation())
+                .observeOn(Schedulers.computation())
+                .distinctUntilChanged()
+                .subscribe(s -> onStateChange(s, identified));
+
+        //noinspection ResultOfMethodCallIgnored
+        signalWebSocket.getUnidentifiedWebSocketState()
+                .subscribeOn(Schedulers.computation())
+                .observeOn(Schedulers.computation())
+                .distinctUntilChanged()
+                .subscribe(s -> onStateChange(s, unidentified));
+    }
+
+    private synchronized void onStateChange(WebSocketConnectionState connectionState, HealthState healthState) {
+        switch (connectionState) {
+            case CONNECTED:
+                logger.debug("WebSocket is now connected");
+                break;
+            case AUTHENTICATION_FAILED:
+                logger.debug("WebSocket authentication failed");
+                break;
+            case FAILED:
+                logger.debug("WebSocket connection failed");
+                break;
+        }
+
+        healthState.needsKeepAlive = connectionState == WebSocketConnectionState.CONNECTED;
+
+        if (keepAliveSender == null && isKeepAliveNecessary()) {
+            keepAliveSender = new KeepAliveSender();
+            keepAliveSender.start();
+        } else if (keepAliveSender != null && !isKeepAliveNecessary()) {
+            keepAliveSender.shutdown();
+            keepAliveSender = null;
+        }
+    }
+
+    @Override
+    public void onKeepAliveResponse(long sentTimestamp, boolean isIdentifiedWebSocket) {
+        if (isIdentifiedWebSocket) {
+            identified.lastKeepAliveReceived = System.currentTimeMillis();
+        } else {
+            unidentified.lastKeepAliveReceived = System.currentTimeMillis();
+        }
+    }
+
+    @Override
+    public void onMessageError(int status, boolean isIdentifiedWebSocket) {
+        if (status == 409) {
+            HealthState healthState = (isIdentifiedWebSocket ? identified : unidentified);
+            if (healthState.mismatchErrorTracker.addSample(System.currentTimeMillis())) {
+                logger.warn("Received too many mismatch device errors, forcing new websockets.");
+                signalWebSocket.forceNewWebSockets();
+            }
+        }
+    }
+
+    private boolean isKeepAliveNecessary() {
+        return identified.needsKeepAlive || unidentified.needsKeepAlive;
+    }
+
+    private static class HealthState {
+
+        private final HttpErrorTracker mismatchErrorTracker = new HttpErrorTracker(5, TimeUnit.MINUTES.toMillis(1));
+
+        private volatile boolean needsKeepAlive;
+        private volatile long lastKeepAliveReceived;
+    }
+
+    /**
+     * Sends periodic heartbeats/keep-alives over both WebSockets to prevent connection timeouts. If
+     * either WebSocket fails 3 times to get a return heartbeat both are forced to be recreated.
+     */
+    private class KeepAliveSender extends Thread {
+
+        private volatile boolean shouldKeepRunning = true;
+
+        public void run() {
+            identified.lastKeepAliveReceived = System.currentTimeMillis();
+            unidentified.lastKeepAliveReceived = System.currentTimeMillis();
+
+            while (shouldKeepRunning && isKeepAliveNecessary()) {
+                try {
+                    sleepTimer.sleep(KEEP_ALIVE_SEND_CADENCE);
+
+                    if (shouldKeepRunning && isKeepAliveNecessary()) {
+                        long keepAliveRequiredSinceTime = System.currentTimeMillis()
+                                - MAX_TIME_SINCE_SUCCESSFUL_KEEP_ALIVE;
+
+                        if (identified.lastKeepAliveReceived < keepAliveRequiredSinceTime
+                                || unidentified.lastKeepAliveReceived < keepAliveRequiredSinceTime) {
+                            logger.warn("Missed keep alives, identified last: "
+                                    + identified.lastKeepAliveReceived
+                                    + " unidentified last: "
+                                    + unidentified.lastKeepAliveReceived
+                                    + " needed by: "
+                                    + keepAliveRequiredSinceTime);
+                            signalWebSocket.forceNewWebSockets();
+                        } else {
+                            signalWebSocket.sendKeepAlive();
+                        }
+                    }
+                } catch (Throwable e) {
+                    logger.warn("Error occured in KeepAliveSender, ignoring ...", e);
+                }
+            }
+        }
+
+        public void shutdown() {
+            shouldKeepRunning = false;
+        }
+    }
+
+    private final static class HttpErrorTracker {
+
+        private final long[] timestamps;
+        private final long errorTimeRange;
+
+        public HttpErrorTracker(int samples, long errorTimeRange) {
+            this.timestamps = new long[samples];
+            this.errorTimeRange = errorTimeRange;
+        }
+
+        public synchronized boolean addSample(long now) {
+            long errorsMustBeAfter = now - errorTimeRange;
+            int count = 1;
+            int minIndex = 0;
+
+            for (int i = 0; i < timestamps.length; i++) {
+                if (timestamps[i] < errorsMustBeAfter) {
+                    timestamps[i] = 0;
+                } else if (timestamps[i] != 0) {
+                    count++;
+                }
+
+                if (timestamps[i] < timestamps[minIndex]) {
+                    minIndex = i;
+                }
+            }
+
+            timestamps[minIndex] = now;
+
+            if (count >= timestamps.length) {
+                Arrays.fill(timestamps, 0);
+                return true;
+            }
+            return false;
+        }
+    }
+}

lib/src/main/java/org/asamk/signal/manager/config/ServiceConfig.java

@@ -34,7 +34,12 @@ public class ServiceConfig {
         } catch (Throwable ignored) {
             zkGroupAvailable = false;
         }
-        capabilities = new AccountAttributes.Capabilities(false, zkGroupAvailable, false, zkGroupAvailable, false);
+        capabilities = new AccountAttributes.Capabilities(false,
+                zkGroupAvailable,
+                false,
+                zkGroupAvailable,
+                false,
+                false);
 
         try {
             TrustStore contactTrustStore = new IasTrustStore();

lib/src/main/java/org/asamk/signal/manager/helper/GroupV2Helper.java

@@ -207,7 +207,7 @@ public class GroupV2Helper {
         var change = name != null ? groupOperations.createModifyGroupTitle(name) : GroupChange.Actions.newBuilder();
 
         if (description != null) {
-            change.setModifyDescription(groupOperations.createModifyGroupDescription(description));
+            change.setModifyDescription(groupOperations.createModifyGroupDescriptionAction(description));
         }
 
         if (avatarFile != null) {

lib/src/main/java/org/asamk/signal/manager/helper/MessagePipeProvider.java

@@ -1,8 +0,0 @@
-package org.asamk.signal.manager.helper;
-
-import org.whispersystems.signalservice.api.SignalServiceMessagePipe;
-
-public interface MessagePipeProvider {
-
-    SignalServiceMessagePipe getMessagePipe(boolean unidentified);
-}

lib/src/main/java/org/asamk/signal/manager/helper/ProfileHelper.java

@@ -9,14 +9,12 @@ import org.whispersystems.signalservice.api.profiles.SignalServiceProfile;
 import org.whispersystems.signalservice.api.push.SignalServiceAddress;
 import org.whispersystems.signalservice.api.push.exceptions.NotFoundException;
 import org.whispersystems.signalservice.api.push.exceptions.PushNetworkException;
-import org.whispersystems.signalservice.internal.util.concurrent.CascadingFuture;
+import org.whispersystems.signalservice.api.services.ProfileService;
-import org.whispersystems.signalservice.internal.util.concurrent.ListenableFuture;
+import org.whispersystems.signalservice.internal.ServiceResponse;
 
 import java.io.IOException;
-import java.util.Arrays;
+
-import java.util.concurrent.ExecutionException;
+import io.reactivex.rxjava3.core.Single;
-import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 
 public final class ProfileHelper {
 
@@ -24,7 +22,7 @@ public final class ProfileHelper {
 
     private final UnidentifiedAccessProvider unidentifiedAccessProvider;
 
-    private final MessagePipeProvider messagePipeProvider;
+    private final ProfileServiceProvider profileServiceProvider;
 
     private final MessageReceiverProvider messageReceiverProvider;
 
@@ -33,13 +31,13 @@ public final class ProfileHelper {
     public ProfileHelper(
             final ProfileKeyProvider profileKeyProvider,
             final UnidentifiedAccessProvider unidentifiedAccessProvider,
-            final MessagePipeProvider messagePipeProvider,
+            final ProfileServiceProvider profileServiceProvider,
             final MessageReceiverProvider messageReceiverProvider,
             final SignalServiceAddressResolver addressResolver
     ) {
         this.profileKeyProvider = profileKeyProvider;
         this.unidentifiedAccessProvider = unidentifiedAccessProvider;
-        this.messagePipeProvider = messagePipeProvider;
+        this.profileServiceProvider = profileServiceProvider;
         this.messageReceiverProvider = messageReceiverProvider;
         this.addressResolver = addressResolver;
     }
@@ -48,8 +46,8 @@ public final class ProfileHelper {
             RecipientId recipientId, SignalServiceProfile.RequestType requestType
     ) throws IOException {
         try {
-            return retrieveProfile(recipientId, requestType).get(10, TimeUnit.SECONDS);
+            return retrieveProfile(recipientId, requestType).blockingGet();
-        } catch (ExecutionException e) {
+        } catch (RuntimeException e) {
             if (e.getCause() instanceof PushNetworkException) {
                 throw (PushNetworkException) e.getCause();
             } else if (e.getCause() instanceof NotFoundException) {
@@ -57,79 +55,55 @@ public final class ProfileHelper {
             } else {
                 throw new IOException(e);
             }
-        } catch (InterruptedException | TimeoutException e) {
-            throw new PushNetworkException(e);
         }
     }
 
-    public ListenableFuture<ProfileAndCredential> retrieveProfile(
+    public SignalServiceProfile retrieveProfileSync(String username) throws IOException {
+        return messageReceiverProvider.getMessageReceiver().retrieveProfileByUsername(username, Optional.absent());
+    }
+
+    public Single<ProfileAndCredential> retrieveProfile(
             RecipientId recipientId, SignalServiceProfile.RequestType requestType
-    ) {
+    ) throws IOException {
         var unidentifiedAccess = getUnidentifiedAccess(recipientId);
         var profileKey = Optional.fromNullable(profileKeyProvider.getProfileKey(recipientId));
 
         final var address = addressResolver.resolveSignalServiceAddress(recipientId);
-        if (unidentifiedAccess.isPresent()) {
+        return retrieveProfile(address, profileKey, unidentifiedAccess, requestType);
-            return new CascadingFuture<>(Arrays.asList(() -> getPipeRetrievalFuture(address,
-                    profileKey,
-                    unidentifiedAccess,
-                    requestType),
-                    () -> getSocketRetrievalFuture(address, profileKey, unidentifiedAccess, requestType),
-                    () -> getPipeRetrievalFuture(address, profileKey, Optional.absent(), requestType),
-                    () -> getSocketRetrievalFuture(address, profileKey, Optional.absent(), requestType)),
-                    e -> !(e instanceof NotFoundException));
-        } else {
-            return new CascadingFuture<>(Arrays.asList(() -> getPipeRetrievalFuture(address,
-                    profileKey,
-                    Optional.absent(),
-                    requestType), () -> getSocketRetrievalFuture(address, profileKey, Optional.absent(), requestType)),
-                    e -> !(e instanceof NotFoundException));
-        }
     }
 
-    private ListenableFuture<ProfileAndCredential> getPipeRetrievalFuture(
+    private Single<ProfileAndCredential> retrieveProfile(
             SignalServiceAddress address,
             Optional<ProfileKey> profileKey,
             Optional<UnidentifiedAccess> unidentifiedAccess,
             SignalServiceProfile.RequestType requestType
     ) throws IOException {
-        var unidentifiedPipe = messagePipeProvider.getMessagePipe(true);
+        var profileService = profileServiceProvider.getProfileService();
-        var pipe = unidentifiedPipe != null && unidentifiedAccess.isPresent()
-                ? unidentifiedPipe
-                : messagePipeProvider.getMessagePipe(false);
-        if (pipe != null) {
-            try {
-                return pipe.getProfile(address, profileKey, unidentifiedAccess, requestType);
-            } catch (NoClassDefFoundError e) {
-                // Native zkgroup lib not available for ProfileKey
-                if (!address.getNumber().isPresent()) {
-                    throw new NotFoundException("Can't request profile without number");
-                }
-                var addressWithoutUuid = new SignalServiceAddress(Optional.absent(), address.getNumber());
-                return pipe.getProfile(addressWithoutUuid, profileKey, unidentifiedAccess, requestType);
-            }
-        }
 
-        throw new IOException("No pipe available!");
+        Single<ServiceResponse<ProfileAndCredential>> responseSingle;
-    }
-
-    private ListenableFuture<ProfileAndCredential> getSocketRetrievalFuture(
-            SignalServiceAddress address,
-            Optional<ProfileKey> profileKey,
-            Optional<UnidentifiedAccess> unidentifiedAccess,
-            SignalServiceProfile.RequestType requestType
-    ) throws NotFoundException {
-        var receiver = messageReceiverProvider.getMessageReceiver();
         try {
-            return receiver.retrieveProfile(address, profileKey, unidentifiedAccess, requestType);
+            responseSingle = profileService.getProfile(address, profileKey, unidentifiedAccess, requestType);
         } catch (NoClassDefFoundError e) {
             // Native zkgroup lib not available for ProfileKey
             if (!address.getNumber().isPresent()) {
                 throw new NotFoundException("Can't request profile without number");
             }
             var addressWithoutUuid = new SignalServiceAddress(Optional.absent(), address.getNumber());
-            return receiver.retrieveProfile(addressWithoutUuid, profileKey, unidentifiedAccess, requestType);
+            responseSingle = profileService.getProfile(addressWithoutUuid, profileKey, unidentifiedAccess, requestType);
         }
+
+        return responseSingle.map(pair -> {
+            var processor = new ProfileService.ProfileResponseProcessor(pair);
+            if (processor.hasResult()) {
+                return processor.getResult();
+            } else if (processor.notFound()) {
+                throw new NotFoundException("Profile not found");
+            } else {
+                throw pair.getExecutionError()
+                        .or(pair.getApplicationError())
+                        .or(new IOException("Unknown error while retrieving profile"));
+            }
+        });
     }
 
     private Optional<UnidentifiedAccess> getUnidentifiedAccess(RecipientId recipientId) {

lib/src/main/java/org/asamk/signal/manager/helper/ProfileServiceProvider.java

@@ -0,0 +1,8 @@
+package org.asamk.signal.manager.helper;
+
+import org.whispersystems.signalservice.api.services.ProfileService;
+
+public interface ProfileServiceProvider {
+
+    ProfileService getProfileService();
+}

lib/src/main/java/org/asamk/signal/manager/storage/SignalAccount.java

@@ -168,7 +168,11 @@ public class SignalAccount implements Closeable {
                 recipientStore::resolveRecipient,
                 identityKey,
                 registrationId);
-        signalProtocolStore = new SignalProtocolStore(preKeyStore, signedPreKeyStore, sessionStore, identityKeyStore);
+        signalProtocolStore = new SignalProtocolStore(preKeyStore,
+                signedPreKeyStore,
+                sessionStore,
+                identityKeyStore,
+                this::isMultiDevice);
 
         messageCache = new MessageCache(getMessageCachePath(dataPath, username));
     }

lib/src/main/java/org/asamk/signal/manager/storage/protocol/SignalProtocolStore.java

@@ -12,7 +12,7 @@ import org.whispersystems.libsignal.state.PreKeyStore;
 import org.whispersystems.libsignal.state.SessionRecord;
 import org.whispersystems.libsignal.state.SignedPreKeyRecord;
 import org.whispersystems.libsignal.state.SignedPreKeyStore;
-import org.whispersystems.signalservice.api.SignalServiceProtocolStore;
+import org.whispersystems.signalservice.api.SignalServiceDataStore;
 import org.whispersystems.signalservice.api.SignalServiceSessionStore;
 import org.whispersystems.signalservice.api.push.DistributionId;
 
@@ -20,24 +20,28 @@ import java.util.Collection;
 import java.util.List;
 import java.util.Set;
 import java.util.UUID;
+import java.util.function.Supplier;
 
-public class SignalProtocolStore implements SignalServiceProtocolStore {
+public class SignalProtocolStore implements SignalServiceDataStore {
 
     private final PreKeyStore preKeyStore;
     private final SignedPreKeyStore signedPreKeyStore;
     private final SignalServiceSessionStore sessionStore;
     private final IdentityKeyStore identityKeyStore;
+    private final Supplier<Boolean> isMultiDevice;
 
     public SignalProtocolStore(
             final PreKeyStore preKeyStore,
             final SignedPreKeyStore signedPreKeyStore,
             final SignalServiceSessionStore sessionStore,
-            final IdentityKeyStore identityKeyStore
+            final IdentityKeyStore identityKeyStore,
+            final Supplier<Boolean> isMultiDevice
     ) {
         this.preKeyStore = preKeyStore;
         this.signedPreKeyStore = signedPreKeyStore;
         this.sessionStore = sessionStore;
         this.identityKeyStore = identityKeyStore;
+        this.isMultiDevice = isMultiDevice;
     }
 
     @Override
@@ -177,9 +181,12 @@ public class SignalProtocolStore implements SignalServiceProtocolStore {
     }
 
     @Override
-    public void clearSenderKeySharedWith(
+    public void clearSenderKeySharedWith(final Collection<SignalProtocolAddress> addresses) {
-            final DistributionId distributionId, final Collection<SignalProtocolAddress> addresses
-    ) {
         // TODO
     }
+
+    @Override
+    public boolean isMultiDevice() {
+        return isMultiDevice.get();
+    }
 }