Further improve threading

This commit is contained in:
gmbnt 2020-03-24 11:31:01 +11:00
parent c8d5abfe73
commit 6246fea83b
7 changed files with 46 additions and 36 deletions

View File

@ -666,7 +666,7 @@
<action android:name="android.intent.action.BOOT_COMPLETED" /> <action android:name="android.intent.action.BOOT_COMPLETED" />
</intent-filter> </intent-filter>
</receiver> </receiver>
<receiver android:name="org.thoughtcrime.securesms.loki.redesign.messaging.BackgroundPublicChatPollWorker"> <receiver android:name="org.thoughtcrime.securesms.loki.redesign.messaging.BackgroundOpenGroupPollWorker">
<intent-filter> <intent-filter>
<action android:name="android.intent.action.BOOT_COMPLETED" /> <action android:name="android.intent.action.BOOT_COMPLETED" />
</intent-filter> </intent-filter>

View File

@ -64,7 +64,7 @@ import org.thoughtcrime.securesms.loki.LokiPublicChatManager;
import org.thoughtcrime.securesms.loki.MultiDeviceUtilities; import org.thoughtcrime.securesms.loki.MultiDeviceUtilities;
import org.thoughtcrime.securesms.loki.redesign.activities.HomeActivity; import org.thoughtcrime.securesms.loki.redesign.activities.HomeActivity;
import org.thoughtcrime.securesms.loki.redesign.messaging.BackgroundPollWorker; import org.thoughtcrime.securesms.loki.redesign.messaging.BackgroundPollWorker;
import org.thoughtcrime.securesms.loki.redesign.messaging.BackgroundPublicChatPollWorker; import org.thoughtcrime.securesms.loki.redesign.messaging.BackgroundOpenGroupPollWorker;
import org.thoughtcrime.securesms.loki.redesign.messaging.LokiAPIDatabase; import org.thoughtcrime.securesms.loki.redesign.messaging.LokiAPIDatabase;
import org.thoughtcrime.securesms.loki.redesign.messaging.LokiRSSFeedPoller; import org.thoughtcrime.securesms.loki.redesign.messaging.LokiRSSFeedPoller;
import org.thoughtcrime.securesms.loki.redesign.messaging.LokiUserDatabase; import org.thoughtcrime.securesms.loki.redesign.messaging.LokiUserDatabase;
@ -355,7 +355,7 @@ public class ApplicationContext extends MultiDexApplication implements Dependenc
LocalBackupListener.schedule(this); LocalBackupListener.schedule(this);
RotateSenderCertificateListener.schedule(this); RotateSenderCertificateListener.schedule(this);
BackgroundPollWorker.schedule(this); // Session BackgroundPollWorker.schedule(this); // Session
BackgroundPublicChatPollWorker.schedule(this); // Session BackgroundOpenGroupPollWorker.schedule(this); // Session
if (BuildConfig.PLAY_STORE_DISABLED) { if (BuildConfig.PLAY_STORE_DISABLED) {
UpdateApkRefreshListener.schedule(this); UpdateApkRefreshListener.schedule(this);

View File

@ -7,32 +7,32 @@ import org.thoughtcrime.securesms.service.PersistentAlarmManagerListener
import org.thoughtcrime.securesms.util.TextSecurePreferences import org.thoughtcrime.securesms.util.TextSecurePreferences
import java.util.concurrent.TimeUnit import java.util.concurrent.TimeUnit
class BackgroundPublicChatPollWorker : PersistentAlarmManagerListener() { class BackgroundOpenGroupPollWorker : PersistentAlarmManagerListener() {
companion object { companion object {
private val pollInterval = TimeUnit.MINUTES.toMillis(4) private val pollInterval = TimeUnit.MINUTES.toMillis(4)
@JvmStatic @JvmStatic
fun schedule(context: Context) { fun schedule(context: Context) {
BackgroundPublicChatPollWorker().onReceive(context, Intent()) BackgroundOpenGroupPollWorker().onReceive(context, Intent())
} }
} }
override fun getNextScheduledExecutionTime(context: Context): Long { override fun getNextScheduledExecutionTime(context: Context): Long {
return TextSecurePreferences.getPublicChatBackgroundPollTime(context) return TextSecurePreferences.getOpenGroupBackgroundPollTime(context)
} }
override fun onAlarm(context: Context, scheduledTime: Long): Long { override fun onAlarm(context: Context, scheduledTime: Long): Long {
if (scheduledTime != 0L) { if (scheduledTime != 0L) {
val publicChats = DatabaseFactory.getLokiThreadDatabase(context).getAllPublicChats().map { it.value } val openGroups = DatabaseFactory.getLokiThreadDatabase(context).getAllPublicChats().map { it.value }
for (publicChat in publicChats) { for (openGroup in openGroups) {
val poller = LokiPublicChatPoller(context, publicChat) val poller = LokiPublicChatPoller(context, openGroup)
poller.stop() poller.stop()
poller.pollForNewMessages() poller.pollForNewMessages()
} }
} }
val nextTime = System.currentTimeMillis() + pollInterval val nextTime = System.currentTimeMillis() + pollInterval
TextSecurePreferences.setPublicChatBackgroundPollTime(context, nextTime) TextSecurePreferences.setOpenGroupBackgroundPollTime(context, nextTime)
return nextTime return nextTime
} }
} }

View File

@ -32,7 +32,9 @@ class BackgroundPollWorker : PersistentAlarmManagerListener() {
val userHexEncodedPublicKey = TextSecurePreferences.getLocalNumber(context) val userHexEncodedPublicKey = TextSecurePreferences.getLocalNumber(context)
val lokiAPIDatabase = DatabaseFactory.getLokiAPIDatabase(context) val lokiAPIDatabase = DatabaseFactory.getLokiAPIDatabase(context)
try { try {
LokiAPI(userHexEncodedPublicKey, lokiAPIDatabase, (context.applicationContext as ApplicationContext).broadcaster).getMessages().map { messages -> val applicationContext = context.applicationContext as ApplicationContext
val broadcaster = applicationContext.broadcaster
LokiAPI(userHexEncodedPublicKey, lokiAPIDatabase, broadcaster).getMessages().map { messages ->
messages.forEach { messages.forEach {
PushContentReceiveJob(context).processEnvelope(SignalServiceEnvelope(it)) PushContentReceiveJob(context).processEnvelope(SignalServiceEnvelope(it))
} }

View File

@ -23,7 +23,6 @@ import org.whispersystems.signalservice.api.messages.multidevice.SentTranscriptM
import org.whispersystems.signalservice.api.push.SignalServiceAddress import org.whispersystems.signalservice.api.push.SignalServiceAddress
import org.whispersystems.signalservice.loki.api.* import org.whispersystems.signalservice.loki.api.*
import org.whispersystems.signalservice.loki.messaging.LokiThreadFriendRequestStatus import org.whispersystems.signalservice.loki.messaging.LokiThreadFriendRequestStatus
import org.whispersystems.signalservice.loki.utilities.successBackground
import java.security.MessageDigest import java.security.MessageDigest
import java.util.* import java.util.*
@ -70,6 +69,7 @@ class LokiPublicChatPoller(private val context: Context, private val group: Loki
} }
private val pollForDisplayNamesTask = object : Runnable { private val pollForDisplayNamesTask = object : Runnable {
override fun run() { override fun run() {
pollForDisplayNames() pollForDisplayNames()
handler.postDelayed(this, pollForDisplayNamesInterval) handler.postDelayed(this, pollForDisplayNamesInterval)
@ -217,6 +217,7 @@ class LokiPublicChatPoller(private val context: Context, private val group: Loki
val userPrivateKey = IdentityKeyUtil.getIdentityKeyPair(context).privateKey.serialize() val userPrivateKey = IdentityKeyUtil.getIdentityKeyPair(context).privateKey.serialize()
val database = DatabaseFactory.getLokiAPIDatabase(context) val database = DatabaseFactory.getLokiAPIDatabase(context)
LokiFileServerAPI.configure(false, userHexEncodedPublicKey, userPrivateKey, database) LokiFileServerAPI.configure(false, userHexEncodedPublicKey, userPrivateKey, database)
// Kovenant propagates a context to chained promises, so LokiPublicChatAPI.sharedContext should be used for all of the below
LokiDeviceLinkUtilities.getAllLinkedDeviceHexEncodedPublicKeys(userHexEncodedPublicKey).bind(LokiPublicChatAPI.sharedContext) { devices -> LokiDeviceLinkUtilities.getAllLinkedDeviceHexEncodedPublicKeys(userHexEncodedPublicKey).bind(LokiPublicChatAPI.sharedContext) { devices ->
userDevices = devices userDevices = devices
api.getMessages(group.channel, group.server) api.getMessages(group.channel, group.server)
@ -235,15 +236,16 @@ class LokiPublicChatPoller(private val context: Context, private val group: Loki
} }
} }
Promise.of(messages) Promise.of(messages)
}.successBackground { }.success {
val newDisplayNameUpdatees = uniqueDevices.mapNotNull { val newDisplayNameUpdatees = uniqueDevices.mapNotNull {
// This will return null if the current device is a master device // This will return null if the current device is a master device
LokiDeviceLinkUtilities.getMasterHexEncodedPublicKey(it).get() LokiDeviceLinkUtilities.getMasterHexEncodedPublicKey(it).get()
}.toSet() }.toSet()
// Fetch the display names of the master devices // Fetch the display names of the master devices
displayNameUpdatees = displayNameUpdatees.union(newDisplayNameUpdatees) displayNameUpdatees = displayNameUpdatees.union(newDisplayNameUpdatees)
}.successBackground { messages -> }.success { messages ->
// Process messages in the background // Process messages in the background
Thread {
messages.forEach { message -> messages.forEach { message ->
if (userDevices.contains(message.hexEncodedPublicKey)) { if (userDevices.contains(message.hexEncodedPublicKey)) {
processOutgoingMessage(message) processOutgoingMessage(message)
@ -251,6 +253,7 @@ class LokiPublicChatPoller(private val context: Context, private val group: Loki
processIncomingMessage(message) processIncomingMessage(message)
} }
} }
}.start()
}.fail { }.fail {
Log.d("Loki", "Failed to get messages for group chat with ID: ${group.channel} on server: ${group.server}.") Log.d("Loki", "Failed to get messages for group chat with ID: ${group.channel} on server: ${group.server}.")
} }
@ -260,11 +263,13 @@ class LokiPublicChatPoller(private val context: Context, private val group: Loki
if (displayNameUpdatees.isEmpty()) { return } if (displayNameUpdatees.isEmpty()) { return }
val hexEncodedPublicKeys = displayNameUpdatees val hexEncodedPublicKeys = displayNameUpdatees
displayNameUpdatees = setOf() displayNameUpdatees = setOf()
api.getDisplayNames(hexEncodedPublicKeys, group.server).successBackground { mapping -> api.getDisplayNames(hexEncodedPublicKeys, group.server).success { mapping ->
Thread {
for (pair in mapping.entries) { for (pair in mapping.entries) {
val senderDisplayName = "${pair.value} (...${pair.key.takeLast(8)})" val senderDisplayName = "${pair.value} (...${pair.key.takeLast(8)})"
DatabaseFactory.getLokiUserDatabase(context).setServerDisplayName(group.id, pair.key, senderDisplayName) DatabaseFactory.getLokiUserDatabase(context).setServerDisplayName(group.id, pair.key, senderDisplayName)
} }
}.start()
}.fail { }.fail {
displayNameUpdatees = displayNameUpdatees.union(hexEncodedPublicKeys) displayNameUpdatees = displayNameUpdatees.union(hexEncodedPublicKeys)
} }
@ -272,6 +277,7 @@ class LokiPublicChatPoller(private val context: Context, private val group: Loki
private fun pollForDeletedMessages() { private fun pollForDeletedMessages() {
api.getDeletedMessageServerIDs(group.channel, group.server).success { deletedMessageServerIDs -> api.getDeletedMessageServerIDs(group.channel, group.server).success { deletedMessageServerIDs ->
Thread {
val lokiMessageDatabase = DatabaseFactory.getLokiMessageDatabase(context) val lokiMessageDatabase = DatabaseFactory.getLokiMessageDatabase(context)
val deletedMessageIDs = deletedMessageServerIDs.mapNotNull { lokiMessageDatabase.getMessageID(it) } val deletedMessageIDs = deletedMessageServerIDs.mapNotNull { lokiMessageDatabase.getMessageID(it) }
val smsMessageDatabase = DatabaseFactory.getSmsDatabase(context) val smsMessageDatabase = DatabaseFactory.getSmsDatabase(context)
@ -280,6 +286,7 @@ class LokiPublicChatPoller(private val context: Context, private val group: Loki
smsMessageDatabase.deleteMessage(it) smsMessageDatabase.deleteMessage(it)
mmsMessageDatabase.delete(it) mmsMessageDatabase.delete(it)
} }
}.start()
}.fail { }.fail {
Log.d("Loki", "Failed to get deleted messages for group chat with ID: ${group.channel} on server: ${group.server}.") Log.d("Loki", "Failed to get deleted messages for group chat with ID: ${group.channel} on server: ${group.server}.")
} }

View File

@ -25,6 +25,7 @@ class LokiRSSFeedPoller(private val context: Context, private val feed: LokiRSSF
private var hasStarted = false private var hasStarted = false
private val task = object : Runnable { private val task = object : Runnable {
override fun run() { override fun run() {
poll() poll()
handler.postDelayed(this, interval) handler.postDelayed(this, interval)
@ -69,7 +70,7 @@ class LokiRSSFeedPoller(private val context: Context, private val feed: LokiRSSF
PushDecryptJob(context).handleTextMessage(x3, x2, Optional.absent(), Optional.absent()) PushDecryptJob(context).handleTextMessage(x3, x2, Optional.absent(), Optional.absent())
} }
}.fail { exception -> }.fail { exception ->
Log.d("Loki", "Couldn't update RSS feed with ID: $feed.id. $exception") Log.d("Loki", "Couldn't update RSS feed with ID: $feed.id due to exception: $exception.")
} }
} }
} }

View File

@ -1185,11 +1185,11 @@ public class TextSecurePreferences {
setLongPreference(context, "background_poll_time", backgroundPollTime); setLongPreference(context, "background_poll_time", backgroundPollTime);
} }
public static long getPublicChatBackgroundPollTime(Context context) { public static long getOpenGroupBackgroundPollTime(Context context) {
return getLongPreference(context, "public_chat_background_poll_time", 0L); return getLongPreference(context, "public_chat_background_poll_time", 0L);
} }
public static void setPublicChatBackgroundPollTime(Context context, long backgroundPollTime) { public static void setOpenGroupBackgroundPollTime(Context context, long backgroundPollTime) {
setLongPreference(context, "public_chat_background_poll_time", backgroundPollTime); setLongPreference(context, "public_chat_background_poll_time", backgroundPollTime);
} }