Asynchronous Worker in Android WorkManager

Question

Google recently announced new WorkManager architecture component. It makes it easy to schedule synchronous work by implementing doWork() in Worker class, but what if I want to do some asynchronous work in the background? For example, I want to make a network service call using Retrofit. I know I can make a synchronous network request, but it would block the thread and just feels wrong. Is there any solution for this or it is just not supported at the moment?

Krefetz

You mean MainThread is blocked or current thread is blocked? – Ethanethane

Worker thread will be blocked. – Krefetz

Rather than spawning a new thread, you can en-queue two Workers at same time? – Ethanethane

Please carefully read the question. I'm not spawning any new threads. – Krefetz

What I meant was, if you want to do something asynchronous, you need to spawn a thread so that it doesn't execute on the same thread. I am trying to understand your usecase. – Ethanethane

Well, internally Retrofit probably has a separate thread. – Krefetz

Answer 1

Per WorkManager docs:

By default, WorkManager runs its operations on a background thread. If you are already running on a background thread and have need for synchronous (blocking) calls to WorkManager, use synchronous() to access such methods.

Therefore, if you don't use synchronous(), you can safely perform sync network calls from doWork(). This is also a better approach from design perspective because callbacks are messy.

That said, if you really want to fire async jobs from doWork(), you'll need to pause the execution thread and resume it upon async job completion using wait/notify mechanism (or some other thread management mechanism e.g. Semaphore). Not something I would recommend in most cases.

As a side note, WorkManager is in very early alpha.

Polygamist

Answer 2

I used a countdownlatch and waited for this to reach 0, which will only happen once the asynchronous callback has updated it. See this code:

public WorkerResult doWork() {

        final WorkerResult[] result = {WorkerResult.RETRY};
        CountDownLatch countDownLatch = new CountDownLatch(1);
        FirebaseFirestore db = FirebaseFirestore.getInstance();

        db.collection("collection").whereEqualTo("this","that").get().addOnCompleteListener(task -> {
            if(task.isSuccessful()) {
                task.getResult().getDocuments().get(0).getReference().update("field", "value")
                        .addOnCompleteListener(task2 -> {
                            if (task2.isSuccessful()) {
                                result[0] = WorkerResult.SUCCESS;
                            } else {
                                result[0] = WorkerResult.RETRY;
                            }
                            countDownLatch.countDown();
                        });
            } else {
                result[0] = WorkerResult.RETRY;
                countDownLatch.countDown();
            }
        });

        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

        return result[0];

    }

Chihli

What happens when Constraint fails. Means Constraint For ideal state then work manager triggers . and after some time Phone out of Ideal state. – R

Answer 3

FYI there is now ListenableWorker, which is designed to be asynchronous.

Edit: Here are some snippets of example usage. I cut out big chunks of code that I think aren't illustrative, so there's a good chance there's a minor error or two here.

This is for a task that takes a String photoKey, retrieves metadata from a server, does some compression work, and then uploads the compressed photo. This happens off the main thread. Here's how we send the work request:

private void compressAndUploadFile(final String photoKey) {
    Data inputData = new Data.Builder()
            .putString(UploadWorker.ARG_PHOTO_KEY, photoKey)
            .build();
    Constraints constraints = new Constraints.Builder()
            .setRequiredNetworkType(NetworkType.CONNECTED)
            .build();
    OneTimeWorkRequest request = new OneTimeWorkRequest.Builder(UploadWorker.class)
            .setInputData(inputData)
            .setConstraints(constraints)
            .build();
    WorkManager.getInstance().enqueue(request);
}

And in UploadWorker:

public class UploadWorker extends ListenableWorker {
    private static final String TAG = "UploadWorker";
    public static final String ARG_PHOTO_KEY = "photo-key";

    private String mPhotoKey;

    /**
     * @param appContext   The application {@link Context}
     * @param workerParams Parameters to setup the internal state of this worker
     */
    public UploadWorker(@NonNull Context appContext, @NonNull WorkerParameters workerParams) {
        super(appContext, workerParams);
        mPhotoKey = workerParams.getInputData().getString(ARG_PHOTO_KEY);
    }

    @NonNull
    @Override
    public ListenableFuture<Payload> onStartWork() {
        SettableFuture<Payload> future = SettableFuture.create();
        Photo photo = getPhotoMetadataFromServer(mPhotoKey).addOnCompleteListener(task -> {
            if (!task.isSuccessful()) {
                Log.e(TAG, "Failed to retrieve photo metadata", task.getException());
                future.setException(task.getException());
                return;
            }
            MyPhotoType photo = task.getResult();
            File file = photo.getFile();
            Log.d(TAG, "Compressing " + photo);
            MyImageUtil.compressImage(file, MyConstants.photoUploadConfig).addOnCompleteListener(compressionTask -> {
                if (!compressionTask.isSuccessful()) {
                    Log.e(TAG, "Could not parse " + photo + " as an image.", compressionTask.getException());
                    future.set(new Payload(Result.FAILURE));
                    return;
                }
                byte[] imageData = compressionTask.getResult();
                Log.d(TAG, "Done compressing " + photo);
                UploadUtil.uploadToServer(photo, imageData);
                future.set(new Payload(Result.SUCCESS));
            });
        });
        return future;
    }
}

EDIT

Depending on the things you are using in your application, you can also extends RxWorker (if you are using RxJava) or CoroutineWorker (if you're using Coroutines). They both extend from ListenableWorker.

Comport

Can you please add an example on how to use this class? – Joris

@Joris I've added an example. – Comport

I can't use SettableFuture.create() in alpha-13, the class is restricted only to the same library group. – Tropopause

Indeed SettableFuture.create(); module is private only to the WorkManager library group. Cannot be used. – Joris

It's not actually private. #43657117 Just suppress the warning "@SuppressLint("RestrictedApi")" – Farnese

The task is executed on the main thread developer.android.com/reference/androidx/work/ListenableWorker. They say that The startWork() method is called on the main thread. Also I am unable to see any onStartWork in the class. Can you explain this? – Christianize

We can use ListenableWorker, you can see document here: developer.android.com/topic/libraries/architecture/workmanager/… – Luellaluelle

Answer 4

If you are talking about asynchronus job you can move your work into RxJava Observables / Singles.

There is a set of operators like .blockingGet() or .blockingFirst() which transforms Observable<T> into blocking T

Worker performs on background thread so do not worry about NetworkOnMainThreadException.

Basidium

can you answer this:#50580606 – Fennell

Using synchronous over async API when running from a background thread is not always good enough. For example, certain async APIs have some onProgress callbacks will be called on the main thread, heads up with that. – Joris

can ffmpeg command be executed using RxJava? as it is already async method having a callback – Fennell

Answer 5

With the power of coroutines, you can 'synchronise' the doWork() like this:

Suspend method for getting the location (asynchronously):

private suspend fun getLocation(): Location = suspendCoroutine { continuation ->
    val mFusedLocationClient = LocationServices.getFusedLocationProviderClient(appContext)
    mFusedLocationClient.lastLocation.addOnSuccessListener {
        continuation.resume(it)
    }.addOnFailureListener {
        continuation.resumeWithException(it)
    }
}

Call example in doWork():

override fun doWork(): Result {
    val loc = runBlocking {
        getLocation()
    }
    val latitude = loc.latitude
}

2021 update: You can now use CoroutineWorker, which has suspend doWork() method.

class MySuspendWorker(private val appContext: Context, workerParams: WorkerParameters) : CoroutineWorker(appContext, workerParams) {
    override suspend fun doWork(): Result {
        //do your async work
    }
}

Anemo

Very elegant solution 👍🏼 – Coltoncoltsfoot

Thanks @MaherAbuthraa, I've also updated the answer to use the built in suspend worker. – Anemo

Answer 6

I have used BlockingQueue, that simplifies threads synchronization and passing result between threads, you will need only one object

private var disposable = Disposables.disposed()

private val completable = Completable.fromAction { 
        //do some heavy computation
    }.subscribeOn(Schedulers.computation()) // you will do the work on background thread

override fun doWork(): Result {
    val result = LinkedBlockingQueue<Result>()

    disposable = completable.subscribe(
            { result.put(Result.SUCCESS) },
            { result.put(Result.RETRY) }
    )

    return try {
        result.take() //need to block this thread untill completable has finished
    } catch (e: InterruptedException) {
        Result.RETRY
    }
}

Also don't forget to release resources if your Worker has been stopped, this is the main advantage over .blockingGet() as now you can properly free cancel your Rx task.

override fun onStopped(cancelled: Boolean) {
    disposable.dispose()
}

Industrialize

Can you please add more code for same.This is quite abstract – Dishabille

Answer 7

This is late but this might help other people,

You can use CoroutineWorker and inside doWork(), use something called suspendCancellableCoroutine, its designed explicitly for this purpose.

Below is the code snippet:

class FileDownloader(private val appContext: Context, params: WorkerParameters) :
CoroutineWorker(appContext, params) {

   override suspend fun doWork(): Result {

       try {

          suspendCancellableCoroutine<Int> { cancellableContinuation ->

              // Here you can call your asynchronous callback based network

                override fun onComplete() {
                        cancellableContinuation.resumeWith(
                            kotlin.Result.success(100))
                }

                override fun onError(error: Error?) {

                        cancellableContinuation.resumeWithException(
                            error?.connectionException ?: Throwable()
                        )
                   
               }
               
     }

     }catch (e: Exception) {
           return Result.failure()
      }

  return Result.success()
}
}

Here, Coroutine will be stopped until you call cancellableContinuation.resumeWith.

Brigettebrigg

Answer 8

I would also prefer the approach that @TomH recommended. I was using it with Firebase Storage though. Using WorkManager together with CountDownlatch did the trick for me. Here a code snippet. Logs are done with Timber.

It returns the downloadUrl from Firebase as a String after the task is complete but before the worker returns success.

@NonNull
@Override
public Result doWork() {
    mFirebaseStorage = mFirebaseStorage.getInstance();
    mTriviaImageStorageReference = mFirebaseStorage.getReference().child("images");

    CountDownLatch countDown = new CountDownLatch(2);
    Uri imageUri = Uri.parse(getInputData().getString(KEY_IMAGE_URI));

    try {

    // get the image reference
    final StorageReference imageRef = mTriviaImageStorageReference.child(imageUri.getLastPathSegment());

    // upload the image to Firebase
    imageRef.putFile(imageUri).continueWithTask(new Continuation<UploadTask.TaskSnapshot, Task<Uri>>() {
        @Override
        public Task<Uri> then(@NonNull Task<UploadTask.TaskSnapshot> task) throws Exception {
            if (!task.isSuccessful()) {
                throw task.getException();
            }
            countDown.countDown();
            return imageRef.getDownloadUrl();
        }
    }).addOnCompleteListener(new OnCompleteListener<Uri>() {
        @Override
        public void onComplete(@NonNull Task<Uri> task) {
            if (task.isSuccessful()) {
                Timber.d("Image was successfully uploaded to Firebase");
                Uri downloadUri = task.getResult();
                String imageUrl = downloadUri.toString();

                Timber.d(("URl of the image is: " + imageUrl));

                mOutputData = new Data.Builder()
                        .putString(KEY_FIREBASE_IMAGE_URL, imageUrl)
                        .build();
                countDown.countDown();
            } else {
                Toast.makeText(getApplicationContext(), "upload failed", Toast.LENGTH_SHORT).show();
                countDown.countDown();
            }
        }
    });
    countDown.await();
    return Result.success(mOutputData);

    } catch (Throwable throwable) {
        Timber.e(throwable, "Error uploading image");
        return Result.failure();
    }
}

Overseer

I tried your way, but my doWork gets called multiple times, – Orlena

Answer 9

This sample could be useful for someone looking for firebase and work manager. it uses androidx.coccurrent so you would need to [install][1] it in your android project.

import android.content.Context;
import androidx.annotation.NonNull;
import androidx.work.ListenableWorker;
import androidx.work.WorkerParameters;
import androidx.concurrent.futures.CallbackToFutureAdapter;

import com.google.android.gms.tasks.OnFailureListener;
import com.google.android.gms.tasks.OnSuccessListener;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.firebase.firestore.FirebaseFirestore;

public class MessageWorker extends ListenableWorker
{
    // Define the parameter keys:
    public static final String MESSAGE_ID = "messageId";
    public static final String MESSAGE_STATUS = "messageStatus";
    public MessageWorker(@NonNull Context context, @NonNull WorkerParameters 
      workerParams) {
        super(context, workerParams);
         }

    @NonNull
    @Override
    public ListenableFuture<Result> startWork() {
        return CallbackToFutureAdapter.getFuture(completer -> {
            String messageId = getInputData().getString(MESSAGE_ID);
            String messageStatus = getInputData().getString(MESSAGE_STATUS);
            FirebaseFirestore.getInstance()
                    .collection("messages")
                    .document(messageId)
                    .update("status", messageStatus)
                    .addOnSuccessListener(new OnSuccessListener<Void>() {
                        @Override
                        public void onSuccess(Void unused) {
                            completer.set(Result.success());

                        }
                    })
                    .addOnFailureListener(new OnFailureListener() {
                        @Override
                        public void onFailure(@NonNull Exception e) {
                            completer.set(Result.retry());
                        }
                    });

            // This value is used only for debug purposes: it will be used
            // in toString() of returned future or error cases.
            return "startSomeAsyncStuff";
        });

    }


}


  [1]: https://developer.android.com/jetpack/androidx/releases/concurrent#1.0.0

Microphone