Android Network Job Queue

Question

I've just finished working on a multi-threaded network queue and was wondering where I could improve. I'm a little rusty when it comes to waits and such.

The idea is to queue up network actions whilst waiting for a form of network internet connectivity. I also added the ability to prioritize which network actions should be performed first. I also intend to add functionality allowing a NetworkJob to specify the type of connection (For those wanting to be connected to WiFi).

With that being said, my main questions are:

1. Can you see any glaring issues in the synchronized blocks and waits?
2. Can you see anywhere I could improve the code for legibility? I'm not totally sure why, but it does seem like quite a mess to me. (Though I might be being over the top.)
3. Are there any features that I've missed?

NetworkQueue.java

import android.content.BroadcastReceiver;
import android.content.Context;
import android.content.Intent;
import android.content.IntentFilter;
import android.net.ConnectivityManager;
import android.net.NetworkInfo;
import android.util.Log;
import java.util.PriorityQueue;
/**
 * Created by Tom on 14/09/2015.
 */
public final class NetworkQueue extends Thread {
 private Context context;
 private final PriorityQueue<NetworkJob> actions;
 private volatile ConnectionType connection = ConnectionType.NONE;
 private final Object networkLock = new Object();
 private final Object actionsLock = new Object();
 private volatile boolean stop = false;
 public NetworkQueue(Context context) {
 this.context = context;
 this.actions = new PriorityQueue<>(5, new NetworkJobComparator());
 context.registerReceiver(new NetworkListener(),
 new IntentFilter(ConnectivityManager.CONNECTIVITY_ACTION));
 }
 @Override
 public void run() {
 try {
 while (!stop) {
 synchronized (actionsLock) {
 if (actions.isEmpty()) {
 Log.v("NetworkQueue", "Awaiting new job");
 actionsLock.wait();
 Log.v("NetworkQueue", "Received new job");
 }
 }
 if (connection == ConnectionType.NONE) {
 synchronized (networkLock) {
 Log.v("NetworkQueue", "Awaiting internet connection!");
 networkLock.wait();
 Log.v("NetworkQueue", "Internet connection obtained!");
 }
 }
 synchronized (actionsLock) {
 NetworkJob job = actions.poll();
 if (job.perform()) {
 continue;
 }
 actions.add(job);
 }
 }
 } catch (InterruptedException e) {
 e.printStackTrace();
 }
 }
 public void addJob(NetworkJob job) {
 synchronized (actionsLock) {
 if (this.actions.contains(job)) {
 Log.v("Network Queue", "Waiting for internet connection, "
 + job.priority() + " priority job: " + job.getClass().getSimpleName() + " already in queue.");
 return;
 }
 this.actions.add(job);
 synchronized (actionsLock) {
 this.actionsLock.notify();
 }
 }
 }
 public void end() {
 this.stop = false;
 }
 private class NetworkListener extends BroadcastReceiver {
 ConnectivityManager conn = (ConnectivityManager)
 context.getSystemService(Context.CONNECTIVITY_SERVICE);
 @Override
 public void onReceive(Context context, Intent intent) {
 NetworkInfo networkInfo = conn.getActiveNetworkInfo();
 if (networkInfo == null) {
 NetworkQueue.this.connection = ConnectionType.NONE;
 return;
 }
 if (networkInfo.getType() == ConnectivityManager.TYPE_WIFI) {
 NetworkQueue.this.connection = ConnectionType.WIFI;
 synchronized (networkLock) {
 networkLock.notifyAll();
 }
 return;
 }
 NetworkQueue.this.connection = ConnectionType.ANY;
 synchronized (networkLock) {
 networkLock.notifyAll();
 }
 }
 }
}

NetworkPriority.java

/**
 * Created by thomas on 16/09/15.
 */
public enum NetworkPriority {
 /**
 * Used when the user is sending data to the server (Responses)
 */
 HIGH,
 /**
 * Used when the user is requesting a specific resource (Usually more important)
 */
 MEDIUM,
 /**
 * Used for general requests
 */
 LOW
}

NetworkJobComparator.java

import java.util.Comparator;
/**
 * Created by thomas on 16/09/15.
 */
public class NetworkJobComparator implements Comparator<NetworkJob> {
 @Override
 public int compare(NetworkJob lhs, NetworkJob rhs) {
 return lhs.priority().compareTo(rhs.priority());
 }
}

NetworkJob.java

public interface NetworkJob {
 /**
 * The action requiring network connectivity
 * @return true only if the network action has succeeded.
 */
 boolean perform();
 NetworkPriority priority();
}

Answer 1

There are three significant items that stand out to me:

1. don't mix memory model strategies - volatile, and synchronized (and I see you removed the concurrent-based imports) all exist in your code. Typically, you only need one of them (which one it is depends on your circumstances)
2. don't override Thread, implement Runnable instead.
3. don't log in synchronized blocks

In addition, there are some smaller items:

1. you double-nest the synchronization on actionLock in the method addJob(...).
2. your end() method sets stop to false, but it was already false, so your end() does nothing, actually..... ;p
3. you don't have enough functions - your existing functions do too much.

On the up-side:

• your enums are good
• variable names are good
• generally neat, and readable code.

So, having said the above, I would recommend your re-introduction of the java.util.concurrent.* classes in to your code, and ignore the synchronization and volatile aspects. A Reentrant lock and condition would be good for waiting for network availability.

public final class NetworkQueue implements Runnable {
 private final Context context;
 private final PriorityBlockingQueue<NetworkJob> actions;
 private final AtomicBoolean running = new AtomicBoolean(true);
 private final ReentrantLock networkLock = new ReentrantLock();
 private final Condition networkUp = networklock.newCondition();
 private ConnectionType connection = ConnectionType.NONE;
 public NetworkQueue(Context context) {
 this.context = context;
 this.actions = new PriorityBlockingQueue<>(5, new NetworkJobComparator());
 context.registerReceiver(new NetworkListener(),
 new IntentFilter(ConnectivityManager.CONNECTIVITY_ACTION));
 }
 private final void setNetwork(ConnectionType net) {
 networkLock.lock();
 try {
 connection = net;
 if (connection != Connection.NONE) {
 networkUp.signal();
 }
 } finally {
 networkLock.unlock();
 }
 }
 private ConnectionType waitNetworkUp() {
 networkLock.lock();
 try {
 while (connection != Connection.NONE) {
 networkUp.await();
 }
 return connection;
 } finally {
 networkLock.unlock();
 }
 }
 @Override
 public void run() {
 try {
 while (running.get()) {
 ConnectionType current = waitNetworkUp();
 NetworkJob job = actions.take();
 if (!job.perform()) {
 actions.add(job);
 }
 }
 } catch (......) {....}
 }
 public void end() {
 running.set(false);
 }
 ....
}

Note how you can use the blocking nature of the actions queue to handle most of the logic in your code.

• Thomas Nairn: Perfect, this is exactly what I was looking for, and actually what I knew - Just hadn't thought of splitting it up the way you did. Nice work! I was actually trying to avoid the PriorityBlockingQueue as I didn't want the logic to rely on the collection, but the way you have done it is really elegant. And I just learnt about ReentrantLock conditions. Thanks!
• Thomas Nairn: Ahh, I remember why I didn't use PriorityBlockingQueue now! I didn't want the application to block on taking a job as - The program will wait until a network is up and then block on getting a job, if the internet goes down whilst it is waiting for a job and then a job is added, it could actually initiate a job with no internet available
• Thomas Nairn: I guess you could use peek() but then the queue will actually retrieve it which is a little inefficient. I solved this by simply adding another condition on your network lock, works great!