content: package act.job;
  
  /*-
   * #%L
   * ACT Framework
   * %%
   * Copyright (C) 2014 - 2017 ActFramework
   * %%
   * Licensed under the Apache License, Version 2.0 (the "License");
   * you may not use this file except in compliance with the License.
   * You may obtain a copy of the License at
   * 
   *      http://www.apache.org/licenses/LICENSE-2.0
   * 
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   * #L%
   */
  
  import static java.util.concurrent.TimeUnit.MILLISECONDS;
  
  import act.app.ActionContext;
  import act.cli.*;
  import act.cli.meta.CommandMethodMetaInfo;
  import act.util.CsvView;
  import act.util.JsonView;
  import act.util.*;
  import act.ws.*;
  import com.alibaba.fastjson.JSONObject;
  import org.osgl.$;
  import org.osgl.exception.UnexpectedException;
  import org.osgl.http.H;
  import org.osgl.mvc.annotation.GetAction;
  import org.osgl.mvc.annotation.ResponseContentType;
  import org.osgl.util.C;
  import org.osgl.util.S;
  
  import java.lang.annotation.Annotation;
  import java.lang.reflect.Method;
  import java.util.Comparator;
  import java.util.List;
  import java.util.concurrent.ScheduledThreadPoolExecutor;
  import javax.inject.Inject;
  
  /**
   * Provide admin service to act {@link JobManager}
   */
  @SuppressWarnings("unused")
  public class JobAdmin {
  
      private static Comparator<Job> _UNIQ_JOB_FILTER = new Comparator<Job>() {
          @Override
          public int compare(Job o1, Job o2) {
              return o1.id().compareTo(o2.id());
          }
      };
  
      /**
       * List all jobs in the job manager
       * @return a list of {@link Job jobs}
       */
      @Command(value = "act.job.list,act.job,act.jobs", help = "List jobs")
      @PropertySpec(Job.BRIEF_VIEW)
      @TableView
      public List<Job> listJobs(@Optional(lead = "-q", help = "search string") final String q, JobManager jobManager) {
          C.List<Job> jobs = jobManager.jobs().append(jobManager.virtualJobs()).unique(_UNIQ_JOB_FILTER);
          if (S.notBlank(q)) {
              jobs = jobs.filter(new $.Predicate<Job>() {
                  @Override
                  public boolean test(Job job) {
                      String jobStr = job.toString();
                      return jobStr.contains(q) || jobStr.matches(q);
                  }
              });
          }
          return jobs;
      }
  
      @Command(value = "act.job.show", help = "Show job details")
      @JsonView
      @PropertySpec(Job.DETAIL_VIEW)
      public Job getJob(@Required("specify job id") final String id, JobManager jobManager) {
          return jobManager.jobById(id);
      }
  
      @Command(value = "act.job.progress", help = "Show job progress")
      public int getJobProgress(@Required("specify job id") final String id, JobManager jobManager) {
          Job job = jobManager.jobById(id);
          return null == job ? -1 : job.getProgressInPercent();
      }
  
      @Command(name = "act.job.cancel", help = "Cancel a job")
      public void cancel(@Required("specify job id") String id, JobManager jobManager) {
          jobManager.cancel(id);
      }
  
      @Command(value = "act.job.scheduler", help = "Show Job manager scheduler status")
      public String getSchedulerStatus(JobManager jobManager) {
          ScheduledThreadPoolExecutor executor = jobManager.executor();
          JSONObject json = new JSONObject();
          json.put("is terminating", executor.isTerminating());
          json.put("is terminated", executor.isTerminated());
          json.put("is shutdown", executor.isShutdown());
          json.put("# of runnable in the queue", executor.getQueue().size());
          json.put("active count", executor.getActiveCount());
          json.put("# of completed tasks", executor.getActiveCount());
          json.put("core pool size", executor.getCorePoolSize());
          json.put("pool size", executor.getPoolSize());
          return json.toJSONString();
      }
  
      @GetAction("jobs/{jobId}/result")
      @Command(value = "act.job.result", help = "Retrieve async job result")
      public Object getAsyncResult(@Required("specify job id") String jobId, JobManager jobManager, ActContext context) {
          Object result = jobManager.cachedResult(jobId);
          if (null == result) {
              return null;
          }
          Object meta = jobManager.cachedMeta(jobId);
          if (null == meta) {
              return result;
          }
          if (meta instanceof CommandMethodMetaInfo) {
              CommandMethodMetaInfo methodMetaInfo = $.cast(meta);
              PropertySpec.MetaInfo filter = methodMetaInfo.propertySpec();
              methodMetaInfo.view().print(result, filter, (CliContext) context);
              return null;
          }
          if (meta instanceof Method) {
              Method method = $.cast(meta);
              PropertySpec spec = method.getAnnotation(PropertySpec.class);
              if (null != spec) {
                  PropertySpec.MetaInfo metaInfo = new PropertySpec.MetaInfo();
                  for (String s : spec.value()) {
                      metaInfo.onValue(s);
                  }
                  for (String s : spec.http()) {
                      metaInfo.onHttp(s);
                  }
                  PropertySpec.currentSpec.set(metaInfo);
              }
              if (isAnnotationPresented(JsonView.class, method)) {
                  context.accept(H.Format.JSON);
              } else if (isAnnotationPresented(CsvView.class, method)) {
                  context.accept(H.Format.CSS);
              } else {
                  ResponseContentType rct = getAnnotation(ResponseContentType.class, method);
                  if (null != rct) {
                      context.accept(rct.value().format());
                  }
              }
              String attachmentName = (String) jobManager.cachedPayload(jobId, "attachmentName");
              if (null != attachmentName && context instanceof ActionContext) {
                  ((ActionContext) context).downloadFileName(attachmentName);
              }
              return result;
          }
          throw new UnexpectedException("Oops! how come? The cached async handler meta is type of " + meta.getClass());
      }
  
      private boolean isAnnotationPresented(Class<? extends Annotation> annotationClass, Method method) {
          return null != getAnnotation(annotationClass, method);
      }
  
      private <T extends Annotation> T getAnnotation(Class<T> annotationClass, Method method) {
          T anno = ReflectedInvokerHelper.getAnnotation(annotationClass, method);
          if (null != anno) {
              return anno;
          }
          return method.getClass().getAnnotation(annotationClass);
      }
  
      @GetAction("jobs/{id}/progress")
      public ProgressGauge jobProgress(String id, JobManager jobManager) {
          Job job = jobManager.jobById(id);
          if (null == job) {
              return SimpleProgressGauge.NULL;
          }
          ProgressGauge gauge = job.progress();
          return null == gauge ? SimpleProgressGauge.NULL : gauge;
      }
  
      @WsEndpoint("/~/ws/jobs/{jobId}/progress")
      public static class WsProgress implements WebSocketConnectionListener {
          @Inject
          private WebSocketConnectionManager connectionManager;
          @Inject
          private JobManager jobManager;
  
          @Override
          public void onConnect(final WebSocketContext context) {
              String jobId = context.actionContext().paramVal("jobId");
              Job job = null;
              if (null != jobId) {
                  job = jobManager.jobById(jobId);
              }
              if (null == job) {
                  jobManager.delay(new Runnable() {
                      @Override
                      public void run() {
                          context.connection().close();
                      }
                  }, 10, MILLISECONDS);
                  return;
              }
              String tag = SimpleProgressGauge.wsJobProgressTag(jobId);
              connectionManager.subscribe(context.session(), tag);
              job.progress().addListener(new ProgressGauge.Listener() {
                  @Override
                  public void onUpdate(ProgressGauge progressGauge) {
                      if (progressGauge.isDone()) {
                          jobManager.delay(new Runnable() {
                              @Override
                              public void run() {
                                  context.connection().close();
                              }
                          }, 10, MILLISECONDS);
                      }
                  }
              });
          }
  
          @Override
          public void onClose(WebSocketContext context) {
          }
      }
  
  }
  