Merge pull request #354 from metaphore/background-polling
Use WorkManager API for Background Pollingpull/384/head
commit
9d8514d7c9
@ -1,17 +0,0 @@
|
|||||||
package org.thoughtcrime.securesms.jobmanager;
|
|
||||||
|
|
||||||
import android.content.BroadcastReceiver;
|
|
||||||
import android.content.Context;
|
|
||||||
import android.content.Intent;
|
|
||||||
|
|
||||||
import org.thoughtcrime.securesms.logging.Log;
|
|
||||||
|
|
||||||
public class BootReceiver extends BroadcastReceiver {
|
|
||||||
|
|
||||||
private static final String TAG = BootReceiver.class.getSimpleName();
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public void onReceive(Context context, Intent intent) {
|
|
||||||
Log.i(TAG, "Boot received. Application is created, kickstarting JobManager.");
|
|
||||||
}
|
|
||||||
}
|
|
@ -1,101 +0,0 @@
|
|||||||
package org.thoughtcrime.securesms.jobmanager.migration;
|
|
||||||
|
|
||||||
import android.content.Context;
|
|
||||||
import android.database.Cursor;
|
|
||||||
import android.database.sqlite.SQLiteDatabase;
|
|
||||||
import android.database.sqlite.SQLiteOpenHelper;
|
|
||||||
import androidx.annotation.NonNull;
|
|
||||||
import androidx.annotation.Nullable;
|
|
||||||
|
|
||||||
import org.thoughtcrime.securesms.jobmanager.Data;
|
|
||||||
import org.thoughtcrime.securesms.jobmanager.Job;
|
|
||||||
import org.thoughtcrime.securesms.jobmanager.impl.NetworkConstraint;
|
|
||||||
import org.thoughtcrime.securesms.jobmanager.persistence.ConstraintSpec;
|
|
||||||
import org.thoughtcrime.securesms.jobmanager.persistence.FullSpec;
|
|
||||||
import org.thoughtcrime.securesms.jobmanager.persistence.JobSpec;
|
|
||||||
import org.thoughtcrime.securesms.logging.Log;
|
|
||||||
|
|
||||||
import java.util.Collections;
|
|
||||||
import java.util.LinkedList;
|
|
||||||
import java.util.List;
|
|
||||||
import java.util.concurrent.TimeUnit;
|
|
||||||
|
|
||||||
final class WorkManagerDatabase extends SQLiteOpenHelper {
|
|
||||||
|
|
||||||
private static final String TAG = WorkManagerDatabase.class.getSimpleName();
|
|
||||||
|
|
||||||
static final String DB_NAME = "androidx.work.workdb";
|
|
||||||
|
|
||||||
WorkManagerDatabase(@NonNull Context context) {
|
|
||||||
super(context, DB_NAME, null, 5);
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public void onCreate(SQLiteDatabase db) {
|
|
||||||
throw new UnsupportedOperationException("We should never be creating this database, only migrating an existing one!");
|
|
||||||
}
|
|
||||||
|
|
||||||
@Override
|
|
||||||
public void onUpgrade(SQLiteDatabase db, int oldVersion, int newVersion) {
|
|
||||||
// There's a chance that a user who hasn't upgraded in > 6 months could hit this onUpgrade path,
|
|
||||||
// but we don't use any of the columns that were added in any migrations they could hit, so we
|
|
||||||
// can ignore this.
|
|
||||||
Log.w(TAG, "Hit onUpgrade path from " + oldVersion + " to " + newVersion);
|
|
||||||
}
|
|
||||||
|
|
||||||
@NonNull List<FullSpec> getAllJobs(@NonNull Data.Serializer dataSerializer) {
|
|
||||||
SQLiteDatabase db = getReadableDatabase();
|
|
||||||
String[] columns = new String[] { "id", "worker_class_name", "input", "required_network_type"};
|
|
||||||
List<FullSpec> fullSpecs = new LinkedList<>();
|
|
||||||
|
|
||||||
try (Cursor cursor = db.query("WorkSpec", columns, null, null, null, null, null)) {
|
|
||||||
while (cursor != null && cursor.moveToNext()) {
|
|
||||||
String factoryName = WorkManagerFactoryMappings.getFactoryKey(cursor.getString(cursor.getColumnIndexOrThrow("worker_class_name")));
|
|
||||||
|
|
||||||
if (factoryName != null) {
|
|
||||||
String id = cursor.getString(cursor.getColumnIndexOrThrow("id"));
|
|
||||||
byte[] data = cursor.getBlob(cursor.getColumnIndexOrThrow("input"));
|
|
||||||
|
|
||||||
List<ConstraintSpec> constraints = new LinkedList<>();
|
|
||||||
JobSpec jobSpec = new JobSpec(id,
|
|
||||||
factoryName,
|
|
||||||
getQueueKey(id),
|
|
||||||
System.currentTimeMillis(),
|
|
||||||
0,
|
|
||||||
0,
|
|
||||||
Job.Parameters.UNLIMITED,
|
|
||||||
TimeUnit.SECONDS.toMillis(30),
|
|
||||||
TimeUnit.DAYS.toMillis(1),
|
|
||||||
Job.Parameters.UNLIMITED,
|
|
||||||
dataSerializer.serialize(DataMigrator.convert(data)),
|
|
||||||
false);
|
|
||||||
|
|
||||||
|
|
||||||
|
|
||||||
if (cursor.getInt(cursor.getColumnIndexOrThrow("required_network_type")) != 0) {
|
|
||||||
constraints.add(new ConstraintSpec(id, NetworkConstraint.KEY));
|
|
||||||
}
|
|
||||||
|
|
||||||
fullSpecs.add(new FullSpec(jobSpec, constraints, Collections.emptyList()));
|
|
||||||
} else {
|
|
||||||
Log.w(TAG, "Failed to find a matching factory for worker class: " + factoryName);
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return fullSpecs;
|
|
||||||
}
|
|
||||||
|
|
||||||
private @Nullable String getQueueKey(@NonNull String jobId) {
|
|
||||||
String query = "work_spec_id = ?";
|
|
||||||
String[] args = new String[] { jobId };
|
|
||||||
|
|
||||||
try (Cursor cursor = getReadableDatabase().query("WorkName", null, query, args, null, null, null)) {
|
|
||||||
if (cursor != null && cursor.moveToFirst()) {
|
|
||||||
return cursor.getString(cursor.getColumnIndexOrThrow("name"));
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
return null;
|
|
||||||
}
|
|
||||||
}
|
|
@ -1,45 +0,0 @@
|
|||||||
package org.thoughtcrime.securesms.jobmanager.migration;
|
|
||||||
|
|
||||||
import android.annotation.SuppressLint;
|
|
||||||
import android.content.Context;
|
|
||||||
import androidx.annotation.NonNull;
|
|
||||||
import androidx.annotation.WorkerThread;
|
|
||||||
|
|
||||||
import org.thoughtcrime.securesms.jobmanager.Data;
|
|
||||||
import org.thoughtcrime.securesms.jobmanager.persistence.FullSpec;
|
|
||||||
import org.thoughtcrime.securesms.jobmanager.persistence.JobStorage;
|
|
||||||
import org.thoughtcrime.securesms.logging.Log;
|
|
||||||
|
|
||||||
import java.util.List;
|
|
||||||
|
|
||||||
public class WorkManagerMigrator {
|
|
||||||
|
|
||||||
private static final String TAG = Log.tag(WorkManagerMigrator.class);
|
|
||||||
|
|
||||||
@SuppressLint("DefaultLocale")
|
|
||||||
@WorkerThread
|
|
||||||
public static synchronized void migrate(@NonNull Context context,
|
|
||||||
@NonNull JobStorage jobStorage,
|
|
||||||
@NonNull Data.Serializer dataSerializer)
|
|
||||||
{
|
|
||||||
long startTime = System.currentTimeMillis();
|
|
||||||
Log.i(TAG, "Beginning WorkManager migration.");
|
|
||||||
|
|
||||||
WorkManagerDatabase database = new WorkManagerDatabase(context);
|
|
||||||
List<FullSpec> fullSpecs = database.getAllJobs(dataSerializer);
|
|
||||||
|
|
||||||
for (FullSpec fullSpec : fullSpecs) {
|
|
||||||
Log.i(TAG, String.format("Migrating job with key '%s' and %d constraint(s).", fullSpec.getJobSpec().getFactoryKey(), fullSpec.getConstraintSpecs().size()));
|
|
||||||
}
|
|
||||||
|
|
||||||
jobStorage.insertJobs(fullSpecs);
|
|
||||||
|
|
||||||
context.deleteDatabase(WorkManagerDatabase.DB_NAME);
|
|
||||||
Log.i(TAG, String.format("WorkManager migration finished. Migrated %d job(s) in %d ms.", fullSpecs.size(), System.currentTimeMillis() - startTime));
|
|
||||||
}
|
|
||||||
|
|
||||||
@WorkerThread
|
|
||||||
public static synchronized boolean needsMigration(@NonNull Context context) {
|
|
||||||
return context.getDatabasePath(WorkManagerDatabase.DB_NAME).exists();
|
|
||||||
}
|
|
||||||
}
|
|
@ -1,87 +0,0 @@
|
|||||||
package org.thoughtcrime.securesms.loki.api
|
|
||||||
|
|
||||||
import android.content.Context
|
|
||||||
import kotlinx.coroutines.awaitAll
|
|
||||||
import nl.komponents.kovenant.Promise
|
|
||||||
import nl.komponents.kovenant.all
|
|
||||||
import nl.komponents.kovenant.functional.map
|
|
||||||
import org.thoughtcrime.securesms.ApplicationContext
|
|
||||||
import org.thoughtcrime.securesms.database.DatabaseFactory
|
|
||||||
import org.thoughtcrime.securesms.dependencies.InjectableType
|
|
||||||
import org.thoughtcrime.securesms.jobmanager.Data
|
|
||||||
import org.thoughtcrime.securesms.jobmanager.Job
|
|
||||||
import org.thoughtcrime.securesms.jobmanager.impl.NetworkConstraint
|
|
||||||
import org.thoughtcrime.securesms.jobs.BaseJob
|
|
||||||
import org.thoughtcrime.securesms.jobs.PushContentReceiveJob
|
|
||||||
import org.thoughtcrime.securesms.jobs.RotateCertificateJob
|
|
||||||
import org.thoughtcrime.securesms.logging.Log
|
|
||||||
import org.thoughtcrime.securesms.util.TextSecurePreferences
|
|
||||||
import org.whispersystems.signalservice.api.SignalServiceAccountManager
|
|
||||||
import org.whispersystems.signalservice.api.messages.SignalServiceEnvelope
|
|
||||||
import org.whispersystems.signalservice.api.push.exceptions.PushNetworkException
|
|
||||||
import org.whispersystems.signalservice.loki.api.SnodeAPI
|
|
||||||
import java.io.IOException
|
|
||||||
import java.util.concurrent.TimeUnit
|
|
||||||
import javax.inject.Inject
|
|
||||||
|
|
||||||
class BackgroundPollJob private constructor(parameters: Parameters) : BaseJob(parameters) {
|
|
||||||
|
|
||||||
companion object {
|
|
||||||
const val KEY = "BackgroundPollJob"
|
|
||||||
}
|
|
||||||
|
|
||||||
constructor(context: Context) : this(Parameters.Builder()
|
|
||||||
.addConstraint(NetworkConstraint.KEY)
|
|
||||||
.setQueue(KEY)
|
|
||||||
.setLifespan(TimeUnit.DAYS.toMillis(1))
|
|
||||||
.setMaxAttempts(Parameters.UNLIMITED)
|
|
||||||
.build()) {
|
|
||||||
setContext(context)
|
|
||||||
}
|
|
||||||
|
|
||||||
override fun serialize(): Data {
|
|
||||||
return Data.EMPTY
|
|
||||||
}
|
|
||||||
|
|
||||||
override fun getFactoryKey(): String { return KEY }
|
|
||||||
|
|
||||||
public override fun onRun() {
|
|
||||||
try {
|
|
||||||
Log.d("Loki", "Performing background poll.")
|
|
||||||
val userPublicKey = TextSecurePreferences.getLocalNumber(context)
|
|
||||||
val promises = mutableListOf<Promise<Unit, Exception>>()
|
|
||||||
if (!TextSecurePreferences.isUsingFCM(context)) {
|
|
||||||
Log.d("Loki", "Not using FCM; polling for contacts and closed groups.")
|
|
||||||
val promise = SnodeAPI.shared.getMessages(userPublicKey).map { envelopes ->
|
|
||||||
envelopes.forEach {
|
|
||||||
PushContentReceiveJob(context).processEnvelope(SignalServiceEnvelope(it), false)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
promises.add(promise)
|
|
||||||
promises.addAll(ClosedGroupPoller.shared.pollOnce())
|
|
||||||
}
|
|
||||||
val openGroups = DatabaseFactory.getLokiThreadDatabase(context).getAllPublicChats().map { it.value }
|
|
||||||
for (openGroup in openGroups) {
|
|
||||||
val poller = PublicChatPoller(context, openGroup)
|
|
||||||
poller.stop()
|
|
||||||
promises.add(poller.pollForNewMessages())
|
|
||||||
}
|
|
||||||
all(promises).get()
|
|
||||||
} catch (exception: Exception) {
|
|
||||||
Log.d("Loki", "Background poll failed due to error: $exception.")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
public override fun onShouldRetry(e: Exception): Boolean {
|
|
||||||
return false
|
|
||||||
}
|
|
||||||
|
|
||||||
override fun onCanceled() { }
|
|
||||||
|
|
||||||
class Factory : Job.Factory<BackgroundPollJob> {
|
|
||||||
|
|
||||||
override fun create(parameters: Parameters, data: Data): BackgroundPollJob {
|
|
||||||
return BackgroundPollJob(parameters)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
}
|
|
@ -1,36 +0,0 @@
|
|||||||
package org.thoughtcrime.securesms.loki.api
|
|
||||||
|
|
||||||
import android.content.Context
|
|
||||||
import android.content.Intent
|
|
||||||
import nl.komponents.kovenant.functional.map
|
|
||||||
import org.thoughtcrime.securesms.ApplicationContext
|
|
||||||
import org.thoughtcrime.securesms.database.DatabaseFactory
|
|
||||||
import org.thoughtcrime.securesms.jobs.PushContentReceiveJob
|
|
||||||
import org.thoughtcrime.securesms.service.PersistentAlarmManagerListener
|
|
||||||
import org.thoughtcrime.securesms.util.TextSecurePreferences
|
|
||||||
import org.whispersystems.signalservice.api.messages.SignalServiceEnvelope
|
|
||||||
import org.whispersystems.signalservice.loki.api.SnodeAPI
|
|
||||||
import java.util.concurrent.TimeUnit
|
|
||||||
|
|
||||||
class BackgroundPollListener : PersistentAlarmManagerListener() {
|
|
||||||
|
|
||||||
companion object {
|
|
||||||
private val pollInterval = TimeUnit.MINUTES.toMillis(15)
|
|
||||||
|
|
||||||
@JvmStatic
|
|
||||||
fun schedule(context: Context) {
|
|
||||||
BackgroundPollListener().onReceive(context, Intent())
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
override fun getNextScheduledExecutionTime(context: Context): Long {
|
|
||||||
return TextSecurePreferences.getBackgroundPollTime(context)
|
|
||||||
}
|
|
||||||
|
|
||||||
override fun onAlarm(context: Context, scheduledTime: Long): Long {
|
|
||||||
ApplicationContext.getInstance(context).jobManager.add(BackgroundPollJob(context))
|
|
||||||
val nextTime = System.currentTimeMillis() + pollInterval
|
|
||||||
TextSecurePreferences.setBackgroundPollTime(context, nextTime)
|
|
||||||
return nextTime
|
|
||||||
}
|
|
||||||
}
|
|
@ -0,0 +1,113 @@
|
|||||||
|
package org.thoughtcrime.securesms.loki.api
|
||||||
|
|
||||||
|
import android.content.BroadcastReceiver
|
||||||
|
import android.content.Context
|
||||||
|
import android.content.Intent
|
||||||
|
import androidx.work.*
|
||||||
|
import nl.komponents.kovenant.Promise
|
||||||
|
import nl.komponents.kovenant.all
|
||||||
|
import nl.komponents.kovenant.functional.map
|
||||||
|
import org.thoughtcrime.securesms.database.DatabaseFactory
|
||||||
|
import org.thoughtcrime.securesms.jobs.PushContentReceiveJob
|
||||||
|
import org.thoughtcrime.securesms.logging.Log
|
||||||
|
import org.thoughtcrime.securesms.util.TextSecurePreferences
|
||||||
|
import org.whispersystems.signalservice.api.messages.SignalServiceEnvelope
|
||||||
|
import org.whispersystems.signalservice.loki.api.SnodeAPI
|
||||||
|
import java.util.concurrent.TimeUnit
|
||||||
|
|
||||||
|
class BackgroundPollWorker(val context: Context, params: WorkerParameters) : Worker(context, params) {
|
||||||
|
|
||||||
|
companion object {
|
||||||
|
const val TAG = "BackgroundPollWorker"
|
||||||
|
|
||||||
|
private const val RETRY_ATTEMPTS = 3
|
||||||
|
|
||||||
|
@JvmStatic
|
||||||
|
fun scheduleInstant(context: Context) {
|
||||||
|
val workRequest = OneTimeWorkRequestBuilder<BackgroundPollWorker>()
|
||||||
|
.setConstraints(
|
||||||
|
Constraints.Builder()
|
||||||
|
.setRequiredNetworkType(NetworkType.CONNECTED)
|
||||||
|
.build()
|
||||||
|
)
|
||||||
|
.build()
|
||||||
|
|
||||||
|
WorkManager
|
||||||
|
.getInstance(context)
|
||||||
|
.enqueue(workRequest)
|
||||||
|
}
|
||||||
|
|
||||||
|
@JvmStatic
|
||||||
|
fun schedulePeriodic(context: Context) {
|
||||||
|
Log.v(TAG, "Scheduling periodic work.")
|
||||||
|
val workRequest = PeriodicWorkRequestBuilder<BackgroundPollWorker>(15, TimeUnit.MINUTES)
|
||||||
|
.setConstraints(
|
||||||
|
Constraints.Builder()
|
||||||
|
.setRequiredNetworkType(NetworkType.CONNECTED)
|
||||||
|
.build()
|
||||||
|
)
|
||||||
|
.build()
|
||||||
|
|
||||||
|
WorkManager
|
||||||
|
.getInstance(context)
|
||||||
|
.enqueueUniquePeriodicWork(
|
||||||
|
TAG,
|
||||||
|
ExistingPeriodicWorkPolicy.KEEP,
|
||||||
|
workRequest
|
||||||
|
)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
override fun doWork(): Result {
|
||||||
|
if (TextSecurePreferences.getLocalNumber(context) == null) {
|
||||||
|
Log.v(TAG, "Background poll is canceled due to the Session user is not set up yet.")
|
||||||
|
return Result.failure()
|
||||||
|
}
|
||||||
|
|
||||||
|
try {
|
||||||
|
Log.v(TAG, "Performing background poll.")
|
||||||
|
val promises = mutableListOf<Promise<Unit, Exception>>()
|
||||||
|
|
||||||
|
// Private chats
|
||||||
|
val userPublicKey = TextSecurePreferences.getLocalNumber(context)
|
||||||
|
val privateChatsPromise = SnodeAPI.shared.getMessages(userPublicKey).map { envelopes ->
|
||||||
|
envelopes.forEach {
|
||||||
|
PushContentReceiveJob(context).processEnvelope(SignalServiceEnvelope(it), false)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
promises.add(privateChatsPromise)
|
||||||
|
|
||||||
|
// Closed groups
|
||||||
|
val sskDatabase = DatabaseFactory.getSSKDatabase(context)
|
||||||
|
ClosedGroupPoller.configureIfNeeded(context, sskDatabase)
|
||||||
|
promises.addAll(ClosedGroupPoller.shared.pollOnce())
|
||||||
|
|
||||||
|
// Open Groups
|
||||||
|
val openGroups = DatabaseFactory.getLokiThreadDatabase(context).getAllPublicChats().map { it.value }
|
||||||
|
for (openGroup in openGroups) {
|
||||||
|
val poller = PublicChatPoller(context, openGroup)
|
||||||
|
promises.add(poller.pollForNewMessages())
|
||||||
|
}
|
||||||
|
|
||||||
|
// Wait till all the promises get resolved
|
||||||
|
all(promises).get()
|
||||||
|
|
||||||
|
return Result.success()
|
||||||
|
} catch (exception: Exception) {
|
||||||
|
Log.v(TAG, "Background poll failed due to error: ${exception.message}.", exception)
|
||||||
|
|
||||||
|
return if (runAttemptCount < RETRY_ATTEMPTS) Result.retry() else Result.failure()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
class BootBroadcastReceiver: BroadcastReceiver() {
|
||||||
|
|
||||||
|
override fun onReceive(context: Context, intent: Intent) {
|
||||||
|
if (intent.action == Intent.ACTION_BOOT_COMPLETED) {
|
||||||
|
Log.v(TAG, "Boot broadcast caught.")
|
||||||
|
BackgroundPollWorker.scheduleInstant(context)
|
||||||
|
BackgroundPollWorker.schedulePeriodic(context)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
Loading…
Reference in New Issue