What is the simplest way to run method asynchronously and being able to see its state?

Question

I would like to run one single long lasting operation and being able to see the following stages of it:

1) not yet ran

2) running

3) finished ok

4) finished with exception

I wrote the code below, which looks excessively complex. It uses three classes: Work, ThreadPoolExecutor, FutureTask<?>, from which Work is handwritten.

Simultaneously, work is partially duplicating FutureTask<?> functionality (exception storing, which is done in Future too, but is closed inside).

The question is: is there any few-line way to do the same from predefined classes from Java, Groovy, GPars, Apache etc?

The code:

public class AsyncRunAndTrackState {

   public static class Stub implements Runnable {
      @Override
      public void run() {
         try {
            Thread.sleep(1000);
         } catch (InterruptedException e) {
            e.printStackTrace();
         }
      }
   }

   public static class Work implements Runnable {

      private Exception exception;

      private boolean active;

      public synchronized Exception getException() {
         return exception;
      }

      public synchronized void setException(Exception exception) {
         this.exception = exception;
      }

      public synchronized boolean isActive() {
         return active;
      }

      public synchronized void setActive(boolean active) {
         this.active = active;
      }

      @Override
      public final void run() {

         setActive(true);
         setException(null);

         try {
            runImpl();
         }
         catch (Exception e) {
            setException(e);
         }
         finally {
            setActive(false);
         }

      }

      protected void runImpl() {
         System.out.println("Before");

         try {
            Thread.sleep(10000);
         } catch (InterruptedException e) {
            e.printStackTrace();
         }

         throw new RuntimeException("Some exception occurred");

         //System.out.println("After");
      }
   }

   static ThreadPoolExecutor executor = (ThreadPoolExecutor) Executors.newFixedThreadPool(1);

   static FutureTask<?> future;

   static Work work;

   public static void main(String[] args) {

      for(int i=0; i<10; ++i) {
         executor.submit(new Stub());
      }

      work = new Work();
      future = (FutureTask<?>) executor.submit(work);

      while(true) {

         System.out.println(String.format("future.done = %s, future.cancelled = %s", future.isDone(), future.isCancelled()));
         System.out.println(String.format("work.active = %s, work.exception = %s", work.isActive(), work.getException()));
         System.out.println();

         try {
            Thread.sleep(500);
         } catch (InterruptedException e) {
            e.printStackTrace();
         }

      }

   }
}

Answer 1

I typically use dataflow queues to notify about state changes from an asynchronous activity.

Answer 2

Maybe it would work if you subclass FutureTask. This would result in some code like this (if I understood everything right):

package experiment;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.FutureTask;

public class Work implements Runnable{

    @Override
    public void run() {
        try {
            Thread.sleep(10000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        throw new RuntimeException("Some exception occurred");
    }

    public static void main(String[] args){
        Work work = new Work();
        MyFutureTask<Object> future = new MyFutureTask<Object>(work, null);
        ExecutorService service = Executors.newCachedThreadPool();
        service.submit(future);

         while(true) {

             System.out.println(String.format("future.done = %s, future.cancelled = %s", future.isDone(), future.isCancelled()));
             System.out.println(String.format("work.active = %s, work.exception = %s", future.isActive(), future.retrieveExeption()));
             System.out.println();

             try {
                Thread.sleep(500);
             } catch (InterruptedException e) {
                e.printStackTrace();
             }

          }

    }
}

class MyFutureTask<A> extends FutureTask<A> {

    private Exception ex;

    @Override
    protected void done() {
        super.done();
        try {
            if (!isCancelled()) get();
        } catch (ExecutionException e) {
            // Exception occurred, deal with it
            ex = e;
        } catch (InterruptedException e) {
            // Shouldn't happen, we're invoked when computation is finished
            throw new AssertionError(e);
        }
    }

    public boolean isActive(){
        return !this.isDone() && !this.isCancelled();
    }

    @Override
    protected boolean runAndReset(){
        this.ex = null;
        return super.runAndReset();
    }

    @Override
    public void run(){
        this.ex = null;
        super.run();
    }

    public Exception retrieveExeption(){
        return ex;
    }

    public MyFutureTask(Runnable runnable, A result) {
        super(runnable, result);
    }

}

It is still a lot of code, but you could reuse MyFutureTask again.

This way of exception handling is described in How to catch exceptions in FutureTask