nmode's Git Repositories - signal-cli/blob - lib/src/main/java/org/asamk/signal/manager/internal/JobExecutor.java

   1 package org.asamk.signal.manager.internal;
   2
   3 import org.asamk.signal.manager.helper.Context;
   4 import org.asamk.signal.manager.jobs.Job;
   5 import org.slf4j.Logger;
   6 import org.slf4j.LoggerFactory;
   7
   8 import java.util.ArrayDeque;
   9 import java.util.Queue;
  10 import java.util.concurrent.ExecutorService;
  11 import java.util.concurrent.Executors;
  12
  13 public class JobExecutor implements AutoCloseable {
  14
  15     private static final Logger logger = LoggerFactory.getLogger(JobExecutor.class);
  16     private final Context context;
  17     private final ExecutorService executorService;
  18     private Job running;
  19     private final Queue<Job> queue = new ArrayDeque<>();
  20
  21     public JobExecutor(final Context context) {
  22         this.context = context;
  23         this.executorService = Executors.newCachedThreadPool();
  24     }
  25
  26     public void enqueueJob(Job job) {
  27         if (executorService.isShutdown()) {
  28             logger.debug("Not enqueuing {} job, shutting down", job.getClass().getSimpleName());
  29             return;
  30         }
  31
  32         synchronized (queue) {
  33             logger.trace("Enqueuing {} job", job.getClass().getSimpleName());
  34             queue.add(job);
  35         }
  36
  37         runNextJob();
  38     }
  39
  40     private void runNextJob() {
  41         Job job;
  42         synchronized (queue) {
  43             if (running != null) {
  44                 return;
  45             }
  46             job = queue.poll();
  47             running = job;
  48         }
  49
  50         if (job == null) {
  51             synchronized (this) {
  52                 this.notifyAll();
  53             }
  54             return;
  55         }
  56         logger.debug("Running {} job", job.getClass().getSimpleName());
  57         executorService.execute(() -> {
  58             try {
  59                 job.run(context);
  60             } catch (Throwable e) {
  61                 logger.warn("Job {} failed", job.getClass().getSimpleName(), e);
  62             } finally {
  63                 synchronized (queue) {
  64                     running = null;
  65                 }
  66                 runNextJob();
  67             }
  68         });
  69     }
  70
  71     @Override
  72     public void close() {
  73         synchronized (queue) {
  74             if (queue.isEmpty()) {
  75                 executorService.close();
  76                 return;
  77             }
  78         }
  79         synchronized (this) {
  80             try {
  81                 this.wait();
  82             } catch (InterruptedException ignored) {
  83             }
  84         }
  85         executorService.close();
  86     }
  87 }