Queue like system to check existence of email

Question

I have a table in my database containing email and email_state amongst other values.

Email is the email of the contact, and email_state is the existence of the email. It can contains 3 values: null (not checked), EXISTS or NOTEXISTS.

When the application (a PlayFramework Java application) starts, I call the start() method from the code below. Everytime I add a new contact (either a one time, or from a CSV import), I also call start.

The start method checks if there is no checked emails in the database, and if so, add them in the queue list. Then it checks if the queue is empty, if yes stop right here. Then it checks if there is already a processus working, if yes, it stops right here.

The code is right below, and I was wondering if it was correct or if I missed anything. The aim is to check for all the emails in the database, in the most efficient way, and manage app shutdown/restart, concurrent creations (two users uploading massive CSV), etc.

package in.twolead.asyncs;

import in.twolead.enums.EmailState;

import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.URLEncoder;
import java.util.List;
import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.TimeUnit;

import org.apache.http.HttpResponse;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;

import play.libs.Akka;
import scala.concurrent.duration.Duration;

import com.avaje.ebean.Ebean;
import com.avaje.ebean.SqlRow;
import com.avaje.ebean.SqlUpdate;

public class EmailVerifier implements Runnable {
    private static EmailVerifier instance = null;

    public static EmailVerifier getInstance() {
        if (instance == null) {
            instance = new EmailVerifier();
        }

        return instance;
    }

    private boolean running = false;
    private ConcurrentLinkedQueue<String> emails;
    private SqlUpdate update;

    private EmailVerifier() {
        this.emails = new ConcurrentLinkedQueue<String>();
        this.update = Ebean.createSqlUpdate("UPDATE `contacts` SET `email_state` = :state WHERE `email` = :email");
    }

    public boolean isRunning() {
        return running;
    }

    public boolean isEmpty() {
        return this.emails.isEmpty();
    }

    public void start() {
        List<SqlRow> rows = Ebean.createSqlQuery("SELECT `email` FROM `contacts` WHERE `email_state` IS NULL GROUP BY `email`").findList();
        for (SqlRow row : rows) {
            this.emails.add(row.getString("email"));
        }

        if (isEmpty()) return;
        if (isRunning()) return;

        Akka.system().scheduler().scheduleOnce(
            Duration.create(10, TimeUnit.MILLISECONDS),
            this,
            Akka.system().dispatcher()
        );
    }

    @Override
    public void run() {
        running = true;

        while (this.emails.peek() != null) {
            String email = this.emails.poll();

            // Request the server
            this.update.setParameter("email", email);
            this.update.setParameter("state", (isEmailExists(email) ? EmailState.EXISTS : EmailState.NOTEXISTS));
            this.update.execute();

            try {
                Thread.sleep(500);
            } catch (InterruptedException e) {
                running = false;
                return;
            }
        }

        running = false;
    }

    private boolean isEmailExists(String email) {
        CloseableHttpClient httpclient = HttpClients.createSystem();

        try {
            HttpGet get = new HttpGet("http://api.voilanorbert.com/v2/verify/" + URLEncoder.encode(email, "UTF-8"));
            get.addHeader("TOKEN", "secure_token");

            HttpResponse response = httpclient.execute(get);
            return (response.getStatusLine().getStatusCode() == 204);
        }
        catch (UnsupportedEncodingException e1) {}
        catch (IOException e) {}

        return false;
    }
}

(note: VoilaNorbert is a service that I run. The currently used API is not yet released to the public)

Answer 1

There are numerous problems with your code, but it might work in most cases at least.

1. Singleton isn't really a pattern one should use. Objects scope is best managed by some sort of external mechanism. Do you use Spring or Guice for dependency injection? Why do you even need your bean to be a singleton?
2. Your class does way too many things. It is all in one: a scheduler, a runnable and a service that retrieves data. You should split out those responsibilities to different classes.
3. The running shouldn't be really neccessary if your code was designed ina different way, however you should make it volatile - you are writing to it and accessing it from different threads.
4. Why does isEmpty() method check (this.emails.peek() == null);. this.emails.isEmpty() works as well and is more descriptive :)
5. Why putting your Thread to sleep? No need for that.
6. You have mentioned in your SO question that the "list" will be modified while running. Is this really the case? Will the "start" method be invoked while the Thread is still running?
7. Since your request method verifies emails existence, it should be named accordingly for readability's sake.

I imagine the way you could rewrite your code is to have a separate Runnable class, separate scheduler and separate service. Email existence verifier can be a separate collaborator as well if it is used elsewhere.

Possible solution

What you could do is make a Runnable with simple List like so (skipped import and some unimportant parts):

/** This class will be injected into your Runnable */
public class EmailExistanceService {

    private SqlUpdate update;

    public EmailExistanceService() {
        this.update = Ebean.createSqlUpdate("UPDATE `contacts` SET `email_state` = :state WHERE `email` = :email");
    }

    /* Synchronized since many threads may access it. You can skip the synchronsied if i.e. you create one instance of this service for each runnable */
    public synchronized void updateEmail(String email, boolean emailExists) {
        this.update.setParameter("email", email);
        this.update.setParameter("state", emailExists ? EmailState.EXISTS : EmailState.NOTEXISTS));
        this.update.execute();
    }
}

Your runnable can have a simple task of verifying emails. It doesn't have to use a queue. You can create a new Runnable each time you need it and do not worry about using queues etc.

public class EmailVerifier implements Runnable {

    private final Collection<String> emailsToProcess;
    private EmailExistanceService emailExistanceService;

    public EmailVerifier(Collection<String> emailsToProcess, EmailExistanceService emailExistanceService) {
        this.emailsToProcess = emailsToProcess;
        this.emailExistanceService = emailExistanceService;
    }

    @Override
    public void run() {
        for (String email : emailsToProcess) {
            emailExistanceService.updateEmail(email, doesEmailExists(email));
        }
    }

    private boolean doesEmailExists(String email) {
        // email verifying code

        return false;
    }
}

Your scheduler will need to create the EmailExistanceService or have it injected from somewhere. Each time you need to go through emails just run the scheduleEmailVerification method.

public class EmailVerifyingScheduler {

    private EmailExistanceService emailExistanceService;

    public void scheduleEmailVerification() {
        List<SqlRow> rows = Ebean.createSqlQuery("SELECT `email` FROM `contacts` WHERE `email_state` IS NULL GROUP BY `email`").findList();
        ArrayList emails = new ArrayList<>(rows.size());
        for (SqlRow row : rows) {
            this.emails.add(row.getString("email"));
        }

        EmailVerifier emailVerifier = new EmailVerifier(emails, emailExistanceService);
        Akka.system().scheduler().scheduleOnce(
            Duration.create(10, TimeUnit.MILLISECONDS),
            emailVerifier,
            Akka.system().dispatcher()
        );
    }

    public void setEmailExistanceService(EmailExistanceService emailExistanceService) {
        this.emailExistanceService = emailExistanceService;
    }
}

For scheduling with only one thread you can simply create a one-thread executor:

public class EmailVerifyingScheduler { 
    private ExecutorService executorService = Executors.newFixedThreadPool(1);

    (...) // rest of code

    public void scheduleEmailVerification() {
        (...)
        executorService.execute(emailVerifier); // instead of Akka's scheduler
    }
}

You can probably achieve this in Akka too, but the above will work.