src/jalview/ws2/actions/PollingTaskExecutor.java

   1 package jalview.ws2.actions;
   2
   3 import java.io.IOException;
   4 import java.util.Collections;
   5 import java.util.Map;
   6 import java.util.Objects;
   7 import java.util.WeakHashMap;
   8 import java.util.concurrent.CancellationException;
   9 import java.util.concurrent.CompletionException;
  10 import java.util.concurrent.Future;
  11 import java.util.concurrent.ScheduledExecutorService;
  12 import java.util.concurrent.TimeUnit;
  13
  14 import jalview.ws2.actions.api.TaskI;
  15
  16 public class PollingTaskExecutor
  17 {
  18   private static final Map<ScheduledExecutorService, PollingTaskExecutor> executorPool =
  19       Collections.synchronizedMap(new WeakHashMap<>());
  20
  21   public static PollingTaskExecutor fromPool(ScheduledExecutorService executor)
  22   {
  23     return executorPool.computeIfAbsent(executor, PollingTaskExecutor::new);
  24   }
  25
  26   private final ScheduledExecutorService executor;
  27
  28   public PollingTaskExecutor(ScheduledExecutorService executor)
  29   {
  30     this.executor = executor;
  31   }
  32
  33   public Future<?> submit(TaskI<?> task)
  34   {
  35     Objects.requireNonNull(task);
  36     return executor.scheduleWithFixedDelay(
  37         new TaskRunnable(task), 0, 2, TimeUnit.SECONDS);
  38   }
  39
  40   private static class TaskRunnable implements Runnable
  41   {
  42     private final TaskI<?> task;
  43
  44     private static final int STAGE_INIT = 0;
  45
  46     private static final int STAGE_POLLING = 2;
  47
  48     private static final int STAGE_FINISHED = 3;
  49
  50     private static final int STAGE_STOPPED = 4;
  51
  52     private int stage = STAGE_INIT;
  53
  54     private static final int MAX_POLL_RETRY = 5;
  55
  56     private int pollRetryCount = 0;
  57
  58     private TaskRunnable(TaskI<?> task)
  59     {
  60       this.task = task;
  61     }
  62
  63     @Override
  64     public void run()
  65     {
  66       if (task.getStatus().isDone())
  67       {
  68         stage = STAGE_STOPPED;
  69       }
  70       if (stage == STAGE_INIT)
  71       {
  72         try
  73         {
  74           task.init();
  75           stage = STAGE_POLLING;
  76         } catch (Exception e)
  77         {
  78           stage = STAGE_STOPPED;
  79           throw new CompletionException(e);
  80         }
  81       }
  82       try
  83       {
  84         if (stage == STAGE_POLLING && task.poll())
  85         {
  86           stage = STAGE_FINISHED;
  87         }
  88         if (stage == STAGE_FINISHED)
  89         {
  90           task.complete();
  91           stage = STAGE_STOPPED;
  92         }
  93       } catch (Exception e)
  94       {
  95         if (++pollRetryCount > MAX_POLL_RETRY || e instanceof RuntimeException)
  96         {
  97           task.cancel();
  98           stage = STAGE_STOPPED;
  99           throw new CompletionException(e);
 100         }
 101       }
 102       if (stage == STAGE_STOPPED)
 103       {
 104         throw new CancellationException();
 105       }
 106     }
 107   }
 108 }