frontend/src/yukihane/inqubus/manager/TaskManage.java

   1 package yukihane.inqubus.manager;
   2
   3 import java.io.File;
   4 import java.util.HashMap;
   5 import java.util.Map;
   6 import java.util.concurrent.ExecutorService;
   7 import java.util.concurrent.Executors;
   8 import java.util.concurrent.Future;
   9 import java.util.logging.Level;
  10 import java.util.logging.Logger;
  11 import saccubus.worker.WorkerListener;
  12 import saccubus.worker.impl.MessageReportable;
  13 import saccubus.worker.impl.PercentageReportable;
  14 import saccubus.worker.impl.convert.Convert;
  15 import saccubus.worker.impl.convert.ConvertProgress;
  16 import saccubus.worker.impl.convert.ConvertResult;
  17 import saccubus.worker.impl.download.Download;
  18 import saccubus.worker.impl.download.DownloadProgress;
  19 import saccubus.worker.impl.download.DownloadResult;
  20 import saccubus.worker.profile.ConvertProfile;
  21 import saccubus.worker.profile.DownloadProfile;
  22
  23 /**
  24  *
  25  * @author yuki
  26  */
  27 public class TaskManage {
  28
  29     private static final Logger logger = Logger.getLogger(TaskManage.class.getName());
  30     private final ExecutorService downloadExecutorService;
  31     private final ExecutorService convertExecutorService;
  32     private final Map<Integer, ManageTarget<DownloadResult>> downloadTargets = new HashMap<>();
  33     private final Map<Integer, ManageTarget<ConvertResult>> convertTargets = new HashMap<>();
  34     private final TaskManageListener clientListener;
  35     private final int waitDownload;
  36
  37     public TaskManage() {
  38         this(1, 30, 1, null);
  39     }
  40
  41     public TaskManage(int maxDownload, int maxConvert) {
  42         this(maxDownload, 30, maxConvert, null);
  43     }
  44
  45     public TaskManage(int maxDownload, int waitDownload, int maxConvert, TaskManageListener listener) {
  46         downloadExecutorService = Executors.newFixedThreadPool(maxDownload);
  47         convertExecutorService = Executors.newFixedThreadPool(maxConvert);
  48         this.waitDownload = waitDownload;
  49         this.clientListener = listener;
  50     }
  51
  52     public synchronized boolean add(RequestProcess request) {
  53         final DownloadProfile dp = request.getDownloadProfile();
  54         final ConvertProfile cp = request.getConvertProfile();
  55         if (dp != null && (dp.getVideoProfile().isDownload() || dp.getCommentProfile().isDownload())) {
  56             // ダウンロードするものがあればまずダウンロード処理
  57             final Download task = new Download(dp, request.getVideoId(),
  58                     new DownloadListener(request.getRowId()), waitDownload);
  59             final Future<DownloadResult> future = downloadExecutorService.submit(task);
  60             downloadTargets.put(request.getRowId(), new ManageTarget<>(request, future));
  61             return true;
  62
  63         } else if (cp != null && cp.isConvert()) {
  64             final Convert task = new Convert(cp, dp.getVideoProfile().getLocalFile(), dp.getCommentProfile().
  65                     getLocalFile(), new ConvertListener(request.getRowId()));
  66             final Future<ConvertResult> future = convertExecutorService.submit(task);
  67             convertTargets.put(request.getRowId(), new ManageTarget<>(request, future));
  68             return true;
  69         }
  70         return false;
  71     }
  72
  73     public synchronized boolean cancel(int rowId) {
  74         // FIXME 実行前にキャンセルした場合にはcancelledイベントが飛ばないのでMapからリクエストを削除できない
  75         final ManageTarget<DownloadResult> down = downloadTargets.get(rowId);
  76         if (down != null) {
  77             return down.getFuture().cancel(true);
  78         }
  79         final ManageTarget<ConvertResult> conv = convertTargets.get(rowId);
  80         if (conv != null) {
  81             return conv.getFuture().cancel(true);
  82         }
  83
  84         return false;
  85     }
  86
  87     private class DownloadListener extends TaskManageInnerListener<DownloadResult, DownloadProgress> {
  88
  89         private DownloadListener(int rowId) {
  90             super(rowId);
  91         }
  92
  93         @Override
  94         public void done(DownloadResult result) {
  95             super.done(result);
  96             synchronized (TaskManage.this) {
  97                 final ManageTarget<DownloadResult> mt = removeRequest(getRowId());
  98                 final RequestProcess request = mt.getRequest();
  99                 if (request.getConvertProfile().isConvert()) {
 100                     final DownloadProfile dp = request.getDownloadProfile();
 101                     final File video = (dp.getVideoProfile().isDownload()) ? result.getDownloadVideo() : dp.
 102                             getVideoProfile().getLocalFile();
 103                     final File comment = (dp.getCommentProfile().isDownload()) ? result.getDownloadComment() : dp.
 104                             getCommentProfile().getLocalFile();
 105                     final ConvertProfile cp = request.getConvertProfile();
 106                     final Convert task = new Convert(cp, video, comment, new ConvertListener(getRowId()));
 107                     final Future<ConvertResult> future = convertExecutorService.submit(task);
 108                     convertTargets.put(request.getRowId(), new ManageTarget<>(request, future));
 109                 }
 110             }
 111         }
 112
 113         @Override
 114         protected TaskKind getKind() {
 115             return TaskKind.DOWNLOAD;
 116         }
 117
 118         @Override
 119         protected ManageTarget<DownloadResult> removeRequest(int rowId) {
 120             return downloadTargets.remove(rowId);
 121         }
 122     }
 123
 124     private class ConvertListener extends TaskManageInnerListener<ConvertResult, ConvertProgress> {
 125
 126         private ConvertListener(int rowId) {
 127             super(rowId);
 128         }
 129
 130         @Override
 131         protected TaskKind getKind() {
 132             return TaskKind.CONVERT;
 133         }
 134
 135         @Override
 136         public void done(ConvertResult result) {
 137             super.done(result);
 138             synchronized (TaskManage.this) {
 139                 removeRequest(getRowId());
 140             }
 141         }
 142
 143         @Override
 144         protected ManageTarget<ConvertResult> removeRequest(int rowId) {
 145             return convertTargets.remove(rowId);
 146         }
 147     }
 148
 149     abstract class TaskManageInnerListener<T, V extends PercentageReportable & MessageReportable> implements WorkerListener<T, V> {
 150
 151         private final int rowId;
 152
 153         protected TaskManageInnerListener(int rowId) {
 154             this.rowId = rowId;
 155         }
 156
 157         protected int getRowId() {
 158             return rowId;
 159         }
 160
 161         private void notify(TaskStatus status) {
 162             notify(status, -1.0, "");
 163         }
 164
 165         private void notify(TaskStatus status, double percentage, String message) {
 166             if (getListener() == null) {
 167                 return;
 168             }
 169             getListener().process(rowId, getKind(), status, percentage, message);
 170         }
 171
 172         private TaskManageListener getListener() {
 173             return clientListener;
 174         }
 175
 176         protected abstract TaskKind getKind();
 177
 178         @Override
 179         public void process(V progress) {
 180             logger.log(Level.FINEST, "process: {0}", progress);
 181             notify(TaskStatus.DOING, progress.getPercentage(), progress.getMessage());
 182         }
 183
 184         @Override
 185         public final void cancelled() {
 186             logger.log(Level.FINE, "cancelled: {0}", toString());
 187             synchronized (TaskManage.this) {
 188                 removeRequest(rowId);
 189             }
 190             notify(TaskStatus.CANCELLED);
 191         }
 192
 193         /**
 194          * この処理をオーバライドしてキューからリクエストを削除する必要があります.
 195          * @param result 処理結果.
 196          */
 197         @Override
 198         public void done(T result) {
 199             logger.log(Level.FINE, "done: {0}", result);
 200             notify(TaskStatus.DONE);
 201         }
 202
 203         @Override
 204         public final void error(Throwable th) {
 205             logger.log(Level.SEVERE, "error", th);
 206             synchronized (TaskManage.this) {
 207                 removeRequest(rowId);
 208             }
 209             notify(TaskStatus.ERROR, 0.0, th.getMessage());
 210         }
 211
 212         protected abstract ManageTarget<T> removeRequest(int rowId);
 213     }
 214
 215     class ManageTarget<T> {
 216
 217         private final RequestProcess request;
 218         private final Future<T> future;
 219
 220         ManageTarget(RequestProcess request, Future<T> future) {
 221             this.request = request;
 222             this.future = future;
 223         }
 224
 225         Future<T> getFuture() {
 226             return future;
 227         }
 228
 229         RequestProcess getRequest() {
 230             return request;
 231         }
 232     }
 233 }