ingo@1127: package de.intevation.flys.wsplgen; ingo@1127: ingo@1970: import java.util.HashMap; ingo@1970: import java.util.Map; ingo@1970: import java.util.concurrent.Future; ingo@1970: import java.util.concurrent.ScheduledThreadPoolExecutor; ingo@1127: ingo@1127: import org.apache.log4j.Logger; ingo@1127: ingo@1970: import de.intevation.artifacts.CallContext; ingo@1970: ingo@1127: import de.intevation.flys.artifacts.model.WSPLGENJob; ingo@1127: ingo@1127: ingo@1970: /** ingo@1970: * The Scheduler is used to retrieve new WSPLGENJob. The incoming jobs are added ingo@1970: * to a ScheduledThreadPoolExecutor. This thread pool has a number of worker ingo@1970: * threads that processes the WSPLGENJobs. The number of worker threads can be ingo@1970: * set using a System property wsplgen.max.threads ; its default value is ingo@1970: * 1. ingo@1970: * ingo@1970: * @author Ingo Weinzierl ingo@1970: */ ingo@1970: public class Scheduler { ingo@1127: ingo@1970: private class FutureJob { ingo@1970: public Future future; ingo@1970: public WSPLGENJob job; ingo@1970: ingo@1970: public FutureJob(Future future, WSPLGENJob job) { ingo@1970: this.future = future; ingo@1970: this.job = job; ingo@1970: } ingo@1970: } ingo@1970: ingo@1970: public static final int MAX_WSPLGEN_PROCESSES = ingo@1970: Integer.getInteger("wsplgen.max.threads", 1); ingo@1127: ingo@1127: ingo@1970: protected ScheduledThreadPoolExecutor pool; ingo@1970: protected Map jobs; ingo@1127: ingo@1127: ingo@1127: private static Scheduler INSTANCE; ingo@1127: ingo@1127: private static final Logger logger = Logger.getLogger(Scheduler.class); ingo@1127: ingo@1127: ingo@1127: ingo@1127: private Scheduler() { ingo@1970: jobs = new HashMap(); ingo@1970: pool = new ScheduledThreadPoolExecutor(MAX_WSPLGEN_PROCESSES); ingo@1127: } ingo@1127: ingo@1127: ingo@1127: public static Scheduler getInstance() { ingo@1127: if (INSTANCE == null) { ingo@1127: logger.info("Create new WSPLGEN Scheduler..."); ingo@1127: ingo@1127: INSTANCE = new Scheduler(); ingo@1127: } ingo@1127: ingo@1127: return INSTANCE; ingo@1127: } ingo@1127: ingo@1127: ingo@1970: public void addJob(final WSPLGENJob job) { ingo@1970: synchronized (jobs) { ingo@1970: WSPLGENFuture f = new WSPLGENFuture(new WSPLGENCallable(this, job)); ingo@1970: pool.execute(f); ingo@1127: ingo@1970: jobs.put(job.getArtifact().identifier(), new FutureJob(f, job)); ingo@1127: ingo@1970: logger.info("New WSPLGEN job successfully added."); ingo@1127: } ingo@1127: } ingo@1127: ingo@1127: ingo@1970: /** ingo@1970: * Cancels a running (or queued) job. ingo@1970: * ingo@1970: * @param jobId The id of the job (which is the identifier of an Artifact). ingo@1970: */ ingo@1970: public void cancelJob(String jobId) { ingo@1970: logger.debug("Search job in queue: " + jobId); ingo@1127: ingo@1970: synchronized (jobs) { ingo@1970: FutureJob fj = jobs.get(jobId); ingo@1127: ingo@1970: if (fj != null) { ingo@1970: logger.info("Try to cancel job: " + jobId); ingo@1970: ingo@1970: fj.future.cancel(true); ingo@1970: ingo@1970: removeJob(jobId); ingo@1970: ingo@1970: fj.job.getCallContext().afterBackground( ingo@1970: CallContext.STORE); ingo@1970: ingo@1970: logger.info("Canceled job: " + jobId); ingo@1127: } ingo@1127: } ingo@1127: } ingo@1127: ingo@1127: ingo@1970: protected void removeJob(String id) { ingo@1970: synchronized (jobs) { ingo@1970: jobs.remove(id); ingo@1127: } ingo@1127: } ingo@1127: } ingo@1127: // vim:set ts=4 sw=4 si et sta sts=4 fenc=utf-8 :