Constructor, which creates and registers the specified job. @param job job info @param context database context @param info input info (can be null) @param notify notify function (ignored if null) @throws QueryException query exception
(final QueryJobSpec job, final Context context, final InputInfo info,
final Consumer<QueryJobResult> notify)
| 46 | * @throws QueryException query exception |
| 47 | */ |
| 48 | public QueryJob(final QueryJobSpec job, final Context context, final InputInfo info, |
| 49 | final Consumer<QueryJobResult> notify) throws QueryException { |
| 50 | |
| 51 | this.job = job; |
| 52 | this.notify = notify; |
| 53 | jc().context = context; |
| 54 | |
| 55 | // check when job is to be started |
| 56 | final JobOptions opts = job.options; |
| 57 | final Item start = toTime(opts.get(JobOptions.START), info); |
| 58 | long delay = start == null ? 0 : toDelay(start, 0, info); |
| 59 | |
| 60 | // check when job is to be repeated |
| 61 | long interval = 0; |
| 62 | final String inter = opts.get(JobOptions.INTERVAL); |
| 63 | if(inter != null && !inter.isEmpty()) { |
| 64 | interval = new DTDur(token(inter), info).ms(info); |
| 65 | if(interval < 1000) throw JOBS_RANGE_X.get(info, inter); |
| 66 | while(delay < 0) delay += interval; |
| 67 | } |
| 68 | if(delay < 0) throw JOBS_RANGE_X.get(info, start); |
| 69 | |
| 70 | // check when job is to be stopped |
| 71 | final Item end = toTime(opts.get(JobOptions.END), info); |
| 72 | final long duration = end == null ? Long.MAX_VALUE : toDelay(end, delay, info); |
| 73 | if(duration <= delay) throw JOBS_RANGE_X.get(info, end); |
| 74 | |
| 75 | // check job results are to be cached |
| 76 | final boolean cache = opts.contains(JobOptions.CACHE) && opts.get(JobOptions.CACHE); |
| 77 | if(cache && interval > 0) throw JOBS_OPTIONS.get(info); |
| 78 | |
| 79 | // number of scheduled and active tasks must not exceed limit |
| 80 | final JobPool jobs = context.jobs; |
| 81 | if(jobs.tasks.size() + jobs.active.size() >= JobPool.MAX_REGISTERED) |
| 82 | throw JOBS_OVERFLOW1_X.get(info, JobPool.MAX_REGISTERED); |
| 83 | |
| 84 | synchronized(jobs.tasks) { |
| 85 | // custom job ID: check if it is invalid or has already been assigned |
| 86 | String id = opts.get(JobOptions.ID); |
| 87 | if(id != null) { |
| 88 | if(id.startsWith(JobContext.PREFIX)) throw JOBS_ID_INVALID_X.get(info, id); |
| 89 | if(jobs.tasks.containsKey(id) || jobs.active.containsKey(id) || |
| 90 | jobs.results.containsKey(id)) throw JOBS_ID_EXISTS_X.get(info, id); |
| 91 | jc().id(id); |
| 92 | } else { |
| 93 | id = jc().id(); |
| 94 | } |
| 95 | if(cache) { |
| 96 | // check if too many query results are cached |
| 97 | if(jobs.results.size() >= JobPool.MAX_CACHED) { |
| 98 | throw JOBS_OVERFLOW2_X.get(info, JobPool.MAX_CACHED); |
| 99 | } |
| 100 | jobs.results.put(id, result); |
| 101 | } |
| 102 | |
| 103 | // create and schedule job task |
| 104 | final QueryJobTask task = new QueryJobTask(this, delay, interval, duration); |
| 105 | jobs.tasks.put(id, task); |