private final IncomingMessageHandler incomingMessageHandler;
private final Context context;
+ private boolean hasCaughtUpWithOldMessages = false;
Manager(
SignalAccount account,
return LEGACY_LOCK::unlock;
}
};
- this.dependencies = new SignalDependencies(account.getSelfAddress(),
- serviceEnvironmentConfig,
+ this.dependencies = new SignalDependencies(serviceEnvironmentConfig,
userAgent,
credentialsProvider,
account.getSignalProtocolStore(),
avatarStore,
account.getProfileStore()::getProfileKey,
unidentifiedAccessHelper::getAccessFor,
- dependencies::getProfileService,
- dependencies::getMessageReceiver,
this::resolveSignalServiceAddress);
final GroupV2Helper groupV2Helper = new GroupV2Helper(profileHelper::getRecipientProfileKeyCredential,
this::getRecipientProfile,
this::resolveSignalServiceAddress);
this.context = new Context(account,
- dependencies.getAccountManager(),
- dependencies.getMessageReceiver(),
+ dependencies,
stickerPackStore,
sendHelper,
groupHelper,
final var signalWebSocket = dependencies.getSignalWebSocket();
signalWebSocket.connect();
- var hasCaughtUpWithOldMessages = false;
+ hasCaughtUpWithOldMessages = false;
while (!Thread.interrupted()) {
SignalServiceEnvelope envelope;
envelope = result.get();
} else {
// Received indicator that server queue is empty
- hasCaughtUpWithOldMessages = true;
-
handleQueuedActions(queuedActions);
queuedActions.clear();
+ hasCaughtUpWithOldMessages = true;
+ synchronized (this) {
+ this.notifyAll();
+ }
+
// Continue to wait another timeout for new messages
continue;
}
handleQueuedActions(queuedActions);
}
+ public boolean hasCaughtUpWithOldMessages() {
+ return hasCaughtUpWithOldMessages;
+ }
+
private void handleQueuedActions(final Collection<HandleAction> queuedActions) {
+ var interrupted = false;
for (var action : queuedActions) {
try {
action.execute(context);
} catch (Throwable e) {
- if (e instanceof AssertionError && e.getCause() instanceof InterruptedException) {
- Thread.currentThread().interrupt();
+ if ((e instanceof AssertionError || e instanceof RuntimeException)
+ && e.getCause() instanceof InterruptedException) {
+ interrupted = true;
+ continue;
}
logger.warn("Message action failed.", e);
}
}
+ if (interrupted) {
+ Thread.currentThread().interrupt();
+ }
}
public boolean isContactBlocked(final RecipientIdentifier.Single recipient) {
}
public SignalServiceAddress resolveSignalServiceAddress(SignalServiceAddress address) {
- if (address.matches(account.getSelfAddress())) {
- return account.getSelfAddress();
- }
-
return resolveSignalServiceAddress(resolveRecipient(address));
}