Multi-threaded Command Processor

Question

I have written a class that is intended to execute a unit of work to be run in a separate thread. The intended use case for this class is running business logic off of the user interface thread so as not to lock it up while waiting for long running processes to finish.

Here is the class:

import javafx.beans.property.ObjectProperty;
import javafx.beans.property.ReadOnlyObjectProperty;
import javafx.beans.property.SimpleObjectProperty;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Map;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList;

public abstract class Command<V> implements Callable<V> {
    private static final Logger logger = LoggerFactory.getLogger(Command.class);
    private final Map<State, List<CommandListener<V>>> stateListeners = new ConcurrentHashMap<>();
    private final ObjectProperty<State> state = new SimpleObjectProperty<>();
    private final ObjectProperty<V> value = new SimpleObjectProperty<>();
    private final ObjectProperty<Exception> exception = new SimpleObjectProperty<>();
    private Thread thread;

    public Command() {
        reset();
        state.addListener((observable, oldValue, newValue) -> {
            logger.trace(this.getClass().getSimpleName() + " changed state to " + newValue);
        });
    }

    public boolean isReady() {
        return state.get().equals(State.READY);
    }

    public boolean isRunning() {
        return state.get().equals(State.RUNNING);
    }

    public boolean isDone() {
        switch (state.get()) {
            case SUCCEEDED:
            case FAILED:
            case CANCELLED:
                return true;
            default:
                return false;
        }
    }

    public boolean isSucceeded() {
        return state.get().equals(State.SUCCEEDED);
    }

    public boolean isFailed() {
        return state.get().equals(State.FAILED);
    }

    public boolean isCancelled() {
        return state.get().equals(State.CANCELLED);
    }

    public State getState() {
        return state.get();
    }

    public ReadOnlyObjectProperty<State> stateProperty() {
        return state;
    }

    public V getValue() {
        return value.get();
    }

    public ReadOnlyObjectProperty<V> valueProperty() {
        return value;
    }

    public Exception getException() {
        return exception.get();
    }

    public ReadOnlyObjectProperty<Exception> exceptionProperty() {
        return exception;
    }

    public void addEventListener(State state, CommandListener<V> listener) {
        if (stateListeners.get(state) == null) {
            stateListeners.put(state, new CopyOnWriteArrayList<>());
        }

        stateListeners.get(state).add(listener);
    }

    public void removeEventListener(State state, CommandListener<V> listener) {
        if (stateListeners.get(state) == null) {
            return;
        }

        stateListeners.get(state).remove(listener);

        if (stateListeners.get(state).isEmpty()) {
            stateListeners.remove(state);
        }
    }

    private void fireEvent(State state) {
        List<CommandListener<V>> listeners = stateListeners.get(state);

        if (listeners == null || listeners.isEmpty()) {
            return;
        }

        for (CommandListener<V> listener : listeners) {
            if (isCancelled()) {
                break;
            }

            listener.stateChanged(this);
        }
    }

    public void start() {
        thread = new Thread(() -> {
            try {

                V v = call();

                if (isCancelled()) {
                    return;
                }

                value.set(v);
                state.set(State.SUCCEEDED);
                fireEvent(State.SUCCEEDED);

            } catch (Exception e) {

                if (isCancelled()) {
                    return;
                }

                exception.set(e);
                state.set(State.FAILED);
                fireEvent(State.FAILED);
            }
        });
        thread.setName(getClass().getSimpleName() + " Thread");
        if (isCancelled()) {
            return;
        }

        state.set(State.RUNNING);
        fireEvent(State.RUNNING);
        thread.start();
    }

    public void cancel() {
        if (thread == null) {
            return;
        }

        thread.interrupt();
        state.set(State.CANCELLED);
        fireEvent(State.CANCELLED);
    }

    public void reset() {
        state.set(State.READY);
        value.set(null);
        exception.set(null);
        thread = null;
    }

    public void restart() {
        reset();
        start();
    }

    public enum State {
        READY,
        RUNNING,
        SUCCEEDED,
        CANCELLED,
        FAILED
    }
}

Instances are created from a user interface thread (Swing, JavaFX, etc.), and start() is invoked on the same thread. Receiving the results of the command are handled through the addEventHandler(CommandListener<V>) and similar methods.

Here is an example listener used with the above class:

private class SignInSucceededHandler implements CommandListener<User> {
    @Override
    public void stateChanged(Command<User> command) {
        Platform.runLater(() -> {

            loginView.setBusy(false);
            User user = command.getValue();
            logger.info("Successful login for user [" + user.getProfile() + "]");

        });
    }
}

The listener is registered on the user interface thread like so:

command.addEventListener(Command.State.SUCCEEDED, new SignInSucceededHandler());

Notice that the command's return value is retrieved on the JavaFX thread (Platform.runLater() executes the provided Runnable on the JavaFX Application Thread).

Given the typical use case, is it necessary to make this class thread-safe? If so, is it simply a matter of adding the synchronized keywords to the various getter/property methods?

Answer 1

Getting state

public State getState() {
    return state.get();
}

Since you already have a getState() method, you can consider using it instead of state.get().

Comparing State enum (part one)

Enums are safe for comparison by ==, so most of your operations on equals() can be replaced accordingly.

Comparing State enum (part two)

If you are comparing against a set of enum values, such as in your isDone() method, you can also consider using EnumSet.contains() instead of the switch statement:

Set<State> DONE_STATES = EnumSet.of(State.SUCCEEDED, State.FAILED, State.CANCELLED);

public boolean isDone() {
    return DONE_STATES.contains(getState());
}

Answer 2

1. In addEventListener, the code could be simplified to the following:

   stateListeners.computeIfAbsent(state,key-> new CopyOnWriteArrayList<>()).add(listener);
   

2. Avoid using Threads directly. Consider an ExecutorService. You can allow callers to pass their ExecutorService.

   public Command(ExecutorService executor){
    ...
   }
   

3. reset() is an overridable method that you call from the constructor. That's BAD, make it private or final