001
014
015 package com.liferay.portal.scheduler.quartz;
016
017 import com.liferay.portal.kernel.bean.BeanReference;
018 import com.liferay.portal.kernel.bean.ClassLoaderBeanHandler;
019 import com.liferay.portal.kernel.dao.db.DB;
020 import com.liferay.portal.kernel.dao.db.DBFactoryUtil;
021 import com.liferay.portal.kernel.json.JSONFactoryUtil;
022 import com.liferay.portal.kernel.log.Log;
023 import com.liferay.portal.kernel.log.LogFactoryUtil;
024 import com.liferay.portal.kernel.messaging.Destination;
025 import com.liferay.portal.kernel.messaging.InvokerMessageListener;
026 import com.liferay.portal.kernel.messaging.Message;
027 import com.liferay.portal.kernel.messaging.MessageBus;
028 import com.liferay.portal.kernel.messaging.MessageBusUtil;
029 import com.liferay.portal.kernel.messaging.MessageListener;
030 import com.liferay.portal.kernel.portlet.PortletClassLoaderUtil;
031 import com.liferay.portal.kernel.scheduler.IntervalTrigger;
032 import com.liferay.portal.kernel.scheduler.JobState;
033 import com.liferay.portal.kernel.scheduler.JobStateSerializeUtil;
034 import com.liferay.portal.kernel.scheduler.SchedulerEngine;
035 import com.liferay.portal.kernel.scheduler.SchedulerEngineHelperUtil;
036 import com.liferay.portal.kernel.scheduler.SchedulerException;
037 import com.liferay.portal.kernel.scheduler.StorageType;
038 import com.liferay.portal.kernel.scheduler.TriggerFactoryUtil;
039 import com.liferay.portal.kernel.scheduler.TriggerState;
040 import com.liferay.portal.kernel.scheduler.TriggerType;
041 import com.liferay.portal.kernel.scheduler.messaging.SchedulerEventMessageListenerWrapper;
042 import com.liferay.portal.kernel.scheduler.messaging.SchedulerResponse;
043 import com.liferay.portal.kernel.util.CharPool;
044 import com.liferay.portal.kernel.util.ClassLoaderPool;
045 import com.liferay.portal.kernel.util.ProxyUtil;
046 import com.liferay.portal.kernel.util.ServerDetector;
047 import com.liferay.portal.kernel.util.StringPool;
048 import com.liferay.portal.kernel.util.Time;
049 import com.liferay.portal.kernel.util.Validator;
050 import com.liferay.portal.scheduler.job.MessageSenderJob;
051 import com.liferay.portal.security.pacl.PACLClassLoaderUtil;
052 import com.liferay.portal.service.QuartzLocalService;
053 import com.liferay.portal.util.PropsUtil;
054 import com.liferay.portal.util.PropsValues;
055
056 import java.text.ParseException;
057
058 import java.util.ArrayList;
059 import java.util.Collections;
060 import java.util.Date;
061 import java.util.List;
062 import java.util.Map;
063 import java.util.Properties;
064 import java.util.Set;
065
066 import org.quartz.CronScheduleBuilder;
067 import org.quartz.CronTrigger;
068 import org.quartz.JobBuilder;
069 import org.quartz.JobDataMap;
070 import org.quartz.JobDetail;
071 import org.quartz.JobKey;
072 import org.quartz.ObjectAlreadyExistsException;
073 import org.quartz.Scheduler;
074 import org.quartz.SimpleScheduleBuilder;
075 import org.quartz.SimpleTrigger;
076 import org.quartz.Trigger;
077 import org.quartz.TriggerBuilder;
078 import org.quartz.TriggerKey;
079 import org.quartz.impl.StdSchedulerFactory;
080 import org.quartz.impl.jdbcjobstore.UpdateLockRowSemaphore;
081 import org.quartz.impl.matchers.GroupMatcher;
082
083
090 public class QuartzSchedulerEngine implements SchedulerEngine {
091
092 public void afterPropertiesSet() {
093 if (!PropsValues.SCHEDULER_ENABLED) {
094 return;
095 }
096
097 try {
098 quartzLocalService.checkQuartzTables();
099
100 _persistedScheduler = initializeScheduler(
101 "persisted.scheduler.", true);
102
103 _memoryScheduler = initializeScheduler("memory.scheduler.", false);
104 }
105 catch (Exception e) {
106 _log.error("Unable to initialize engine", e);
107 }
108 }
109
110 public void delete(String groupName) throws SchedulerException {
111 if (!PropsValues.SCHEDULER_ENABLED) {
112 return;
113 }
114
115 try {
116 Scheduler scheduler = getScheduler(groupName);
117
118 groupName = fixMaxLength(
119 getOriginalGroupName(groupName), GROUP_NAME_MAX_LENGTH);
120
121 Set<JobKey> jobKeys = scheduler.getJobKeys(
122 GroupMatcher.jobGroupEquals(groupName));
123
124 for (JobKey jobKey : jobKeys) {
125 unregisterMessageListener(scheduler, jobKey);
126
127 scheduler.deleteJob(jobKey);
128 }
129 }
130 catch (Exception e) {
131 throw new SchedulerException(
132 "Unable to delete jobs in group " + groupName, e);
133 }
134 }
135
136 public void delete(String jobName, String groupName)
137 throws SchedulerException {
138
139 if (!PropsValues.SCHEDULER_ENABLED) {
140 return;
141 }
142
143 try {
144 Scheduler scheduler = getScheduler(groupName);
145
146 jobName = fixMaxLength(jobName, JOB_NAME_MAX_LENGTH);
147 groupName = fixMaxLength(
148 getOriginalGroupName(groupName), GROUP_NAME_MAX_LENGTH);
149
150 JobKey jobKey = new JobKey(jobName, groupName);
151
152 unregisterMessageListener(scheduler, jobKey);
153
154 scheduler.deleteJob(jobKey);
155 }
156 catch (Exception e) {
157 throw new SchedulerException(
158 "Unable to delete job {jobName=" + jobName + ", groupName=" +
159 groupName + "}",
160 e);
161 }
162 }
163
164 public void destroy() {
165 try {
166 shutdown();
167 }
168 catch (SchedulerException se) {
169 if (_log.isWarnEnabled()) {
170 _log.warn("Unable to shutdown", se);
171 }
172 }
173 }
174
175 public SchedulerResponse getScheduledJob(String jobName, String groupName)
176 throws SchedulerException {
177
178 if (!PropsValues.SCHEDULER_ENABLED) {
179 return null;
180 }
181
182 try {
183 Scheduler scheduler = getScheduler(groupName);
184
185 jobName = fixMaxLength(jobName, JOB_NAME_MAX_LENGTH);
186 groupName = fixMaxLength(
187 getOriginalGroupName(groupName), GROUP_NAME_MAX_LENGTH);
188
189 JobKey jobKey = new JobKey(jobName, groupName);
190
191 return getScheduledJob(scheduler, jobKey);
192 }
193 catch (Exception e) {
194 throw new SchedulerException(
195 "Unable to get job {jobName=" + jobName + ", groupName=" +
196 groupName + "}",
197 e);
198 }
199 }
200
201 public List<SchedulerResponse> getScheduledJobs()
202 throws SchedulerException {
203
204 if (!PropsValues.SCHEDULER_ENABLED) {
205 return Collections.emptyList();
206 }
207
208 try {
209 List<String> groupNames = _persistedScheduler.getJobGroupNames();
210
211 List<SchedulerResponse> schedulerResponses =
212 new ArrayList<SchedulerResponse>();
213
214 for (String groupName : groupNames) {
215 schedulerResponses.addAll(
216 getScheduledJobs(_persistedScheduler, groupName));
217 }
218
219 groupNames = _memoryScheduler.getJobGroupNames();
220
221 for (String groupName : groupNames) {
222 schedulerResponses.addAll(
223 getScheduledJobs(_memoryScheduler, groupName));
224 }
225
226 return schedulerResponses;
227 }
228 catch (Exception e) {
229 throw new SchedulerException("Unable to get jobs", e);
230 }
231 }
232
233 public List<SchedulerResponse> getScheduledJobs(String groupName)
234 throws SchedulerException {
235
236 if (!PropsValues.SCHEDULER_ENABLED) {
237 return Collections.emptyList();
238 }
239
240 try {
241 Scheduler scheduler = getScheduler(groupName);
242
243 return getScheduledJobs(scheduler, groupName);
244 }
245 catch (Exception e) {
246 throw new SchedulerException(
247 "Unable to get jobs in group " + groupName, e);
248 }
249 }
250
251 public void pause(String groupName) throws SchedulerException {
252 if (!PropsValues.SCHEDULER_ENABLED) {
253 return;
254 }
255
256 try {
257 Scheduler scheduler = getScheduler(groupName);
258
259 groupName = fixMaxLength(
260 getOriginalGroupName(groupName), GROUP_NAME_MAX_LENGTH);
261
262 Set<JobKey> jobKeys = scheduler.getJobKeys(
263 GroupMatcher.jobGroupEquals(groupName));
264
265 for (JobKey jobKey : jobKeys) {
266 updateJobState(scheduler, jobKey, TriggerState.PAUSED, false);
267 }
268
269 scheduler.pauseJobs(GroupMatcher.jobGroupEquals(groupName));
270 }
271 catch (Exception e) {
272 throw new SchedulerException(
273 "Unable to pause jobs in group " + groupName, e);
274 }
275 }
276
277 public void pause(String jobName, String groupName)
278 throws SchedulerException {
279
280 if (!PropsValues.SCHEDULER_ENABLED) {
281 return;
282 }
283
284 try {
285 Scheduler scheduler = getScheduler(groupName);
286
287 jobName = fixMaxLength(jobName, JOB_NAME_MAX_LENGTH);
288 groupName = fixMaxLength(
289 getOriginalGroupName(groupName), GROUP_NAME_MAX_LENGTH);
290
291 JobKey jobKey = new JobKey(jobName, groupName);
292
293 updateJobState(scheduler, jobKey, TriggerState.PAUSED, false);
294
295 scheduler.pauseJob(jobKey);
296 }
297 catch (Exception e) {
298 throw new SchedulerException(
299 "Unable to pause job {jobName=" + jobName + ", groupName=" +
300 groupName + "}",
301 e);
302 }
303 }
304
305 public void resume(String groupName) throws SchedulerException {
306 if (!PropsValues.SCHEDULER_ENABLED) {
307 return;
308 }
309
310 try {
311 Scheduler scheduler = getScheduler(groupName);
312
313 groupName = fixMaxLength(
314 getOriginalGroupName(groupName), GROUP_NAME_MAX_LENGTH);
315
316 Set<JobKey> jobKeys = scheduler.getJobKeys(
317 GroupMatcher.jobGroupEquals(groupName));
318
319 for (JobKey jobKey : jobKeys) {
320 updateJobState(scheduler, jobKey, TriggerState.NORMAL, false);
321 }
322
323 scheduler.resumeJobs(GroupMatcher.jobGroupEquals(groupName));
324 }
325 catch (Exception e) {
326 throw new SchedulerException(
327 "Unable to resume jobs in group " + groupName, e);
328 }
329 }
330
331 public void resume(String jobName, String groupName)
332 throws SchedulerException {
333
334 if (!PropsValues.SCHEDULER_ENABLED) {
335 return;
336 }
337
338 try {
339 Scheduler scheduler = getScheduler(groupName);
340
341 jobName = fixMaxLength(jobName, JOB_NAME_MAX_LENGTH);
342 groupName = fixMaxLength(
343 getOriginalGroupName(groupName), GROUP_NAME_MAX_LENGTH);
344
345 JobKey jobKey = new JobKey(jobName, groupName);
346
347 updateJobState(scheduler, jobKey, TriggerState.NORMAL, false);
348
349 scheduler.resumeJob(jobKey);
350 }
351 catch (Exception e) {
352 throw new SchedulerException(
353 "Unable to resume job {jobName=" + jobName + ", groupName=" +
354 groupName + "}",
355 e);
356 }
357 }
358
359 public void schedule(
360 com.liferay.portal.kernel.scheduler.Trigger trigger,
361 String description, String destination, Message message)
362 throws SchedulerException {
363
364 if (!PropsValues.SCHEDULER_ENABLED) {
365 return;
366 }
367
368 try {
369 Scheduler scheduler = getScheduler(trigger.getGroupName());
370
371 StorageType storageType = getStorageType(trigger.getGroupName());
372
373 trigger = TriggerFactoryUtil.buildTrigger(
374 trigger.getTriggerType(), trigger.getJobName(),
375 getOriginalGroupName(trigger.getGroupName()),
376 trigger.getStartDate(), trigger.getEndDate(),
377 trigger.getTriggerContent());
378
379 Trigger quartzTrigger = getQuartzTrigger(trigger);
380
381 if (quartzTrigger == null) {
382 return;
383 }
384
385 description = fixMaxLength(description, DESCRIPTION_MAX_LENGTH);
386
387 if (message == null) {
388 message = new Message();
389 }
390 else {
391 message = message.clone();
392 }
393
394 registerMessageListeners(
395 trigger.getJobName(), trigger.getGroupName(), destination,
396 message);
397
398 schedule(
399 scheduler, storageType, quartzTrigger, description, destination,
400 message);
401 }
402 catch (RuntimeException re) {
403
404
405
406
407 }
408 catch (Exception e) {
409 throw new SchedulerException("Unable to schedule job", e);
410 }
411 }
412
413 public void shutdown() throws SchedulerException {
414 if (!PropsValues.SCHEDULER_ENABLED) {
415 return;
416 }
417
418 try {
419 if (!_persistedScheduler.isShutdown()) {
420 _persistedScheduler.shutdown(false);
421 }
422
423 if (!_memoryScheduler.isShutdown()) {
424 _memoryScheduler.shutdown(false);
425 }
426 }
427 catch (Exception e) {
428 throw new SchedulerException("Unable to shutdown scheduler", e);
429 }
430 }
431
432 public void start() throws SchedulerException {
433 if (!PropsValues.SCHEDULER_ENABLED) {
434 return;
435 }
436
437 try {
438 _persistedScheduler.start();
439
440 initJobState();
441
442 _memoryScheduler.start();
443 }
444 catch (Exception e) {
445 throw new SchedulerException("Unable to start scheduler", e);
446 }
447 }
448
449 public void suppressError(String jobName, String groupName)
450 throws SchedulerException {
451
452 if (!PropsValues.SCHEDULER_ENABLED) {
453 return;
454 }
455
456 try {
457 Scheduler scheduler = getScheduler(groupName);
458
459 jobName = fixMaxLength(jobName, JOB_NAME_MAX_LENGTH);
460 groupName = fixMaxLength(
461 getOriginalGroupName(groupName), GROUP_NAME_MAX_LENGTH);
462
463 JobKey jobKey = new JobKey(jobName, groupName);
464
465 updateJobState(scheduler, jobKey, null, true);
466 }
467 catch (Exception e) {
468 throw new SchedulerException(
469 "Unable to suppress error for job {jobName=" + jobName +
470 ", groupName=" + groupName + "}",
471 e);
472 }
473 }
474
475 public void unschedule(String groupName) throws SchedulerException {
476 if (!PropsValues.SCHEDULER_ENABLED) {
477 return;
478 }
479
480 try {
481 Scheduler scheduler = getScheduler(groupName);
482
483 groupName = fixMaxLength(
484 getOriginalGroupName(groupName), GROUP_NAME_MAX_LENGTH);
485
486 Set<JobKey> jobKeys = scheduler.getJobKeys(
487 GroupMatcher.jobGroupEquals(groupName));
488
489 for (JobKey jobKey : jobKeys) {
490 unschedule(scheduler, jobKey);
491 }
492 }
493 catch (Exception e) {
494 throw new SchedulerException(
495 "Unable to unschedule jobs in group " + groupName, e);
496 }
497 }
498
499 public void unschedule(String jobName, String groupName)
500 throws SchedulerException {
501
502 if (!PropsValues.SCHEDULER_ENABLED) {
503 return;
504 }
505
506 try {
507 Scheduler scheduler = getScheduler(groupName);
508
509 jobName = fixMaxLength(jobName, JOB_NAME_MAX_LENGTH);
510 groupName = fixMaxLength(
511 getOriginalGroupName(groupName), GROUP_NAME_MAX_LENGTH);
512
513 JobKey jobKey = new JobKey(jobName, groupName);
514
515 unschedule(scheduler, jobKey);
516 }
517 catch (Exception e) {
518 throw new SchedulerException(
519 "Unable to unschedule job {jobName=" + jobName +
520 ", groupName=" + groupName + "}",
521 e);
522 }
523 }
524
525 public void update(com.liferay.portal.kernel.scheduler.Trigger trigger)
526 throws SchedulerException {
527
528 if (!PropsValues.SCHEDULER_ENABLED) {
529 return;
530 }
531
532 try {
533 Scheduler scheduler = getScheduler(trigger.getGroupName());
534
535 trigger = TriggerFactoryUtil.buildTrigger(
536 trigger.getTriggerType(), trigger.getJobName(),
537 getOriginalGroupName(trigger.getGroupName()),
538 trigger.getStartDate(), trigger.getEndDate(),
539 trigger.getTriggerContent());
540
541 update(scheduler, trigger);
542 }
543 catch (Exception e) {
544 throw new SchedulerException("Unable to update trigger", e);
545 }
546 }
547
548 protected String fixMaxLength(String argument, int maxLength) {
549 if (argument == null) {
550 return null;
551 }
552
553 if (argument.length() > maxLength) {
554 argument = argument.substring(0, maxLength);
555 }
556
557 return argument;
558 }
559
560 protected String getFullName(String jobName, String groupName) {
561 return groupName.concat(StringPool.PERIOD).concat(jobName);
562 }
563
564 protected JobState getJobState(JobDataMap jobDataMap) {
565 Map<String, Object> jobStateMap = (Map<String, Object>)jobDataMap.get(
566 JOB_STATE);
567
568 return JobStateSerializeUtil.deserialize(jobStateMap);
569 }
570
571 protected Message getMessage(JobDataMap jobDataMap) {
572 String messageJSON = (String)jobDataMap.get(MESSAGE);
573
574 return (Message)JSONFactoryUtil.deserialize(messageJSON);
575 }
576
577 protected MessageListener getMessageListener(
578 String messageListenerClassName, ClassLoader classLoader)
579 throws SchedulerException {
580
581 MessageListener schedulerEventListener = null;
582
583 try {
584 Class<? extends MessageListener> clazz =
585 (Class<? extends MessageListener>)classLoader.loadClass(
586 messageListenerClassName);
587
588 schedulerEventListener = clazz.newInstance();
589
590 schedulerEventListener =
591 (MessageListener)ProxyUtil.newProxyInstance(
592 classLoader, new Class<?>[] {MessageListener.class},
593 new ClassLoaderBeanHandler(
594 schedulerEventListener, classLoader));
595 }
596 catch (Exception e) {
597 throw new SchedulerException(
598 "Unable to register message listener with name " +
599 messageListenerClassName,
600 e);
601 }
602
603 return schedulerEventListener;
604 }
605
606 protected String getOriginalGroupName(String groupName) {
607 int pos = groupName.indexOf(CharPool.POUND);
608
609 return groupName.substring(pos + 1);
610 }
611
612 protected Trigger getQuartzTrigger(
613 com.liferay.portal.kernel.scheduler.Trigger trigger)
614 throws SchedulerException {
615
616 if (trigger == null) {
617 return null;
618 }
619
620 Date endDate = trigger.getEndDate();
621 String jobName = fixMaxLength(
622 trigger.getJobName(), JOB_NAME_MAX_LENGTH);
623 String groupName = fixMaxLength(
624 trigger.getGroupName(), GROUP_NAME_MAX_LENGTH);
625
626 Date startDate = trigger.getStartDate();
627
628 if (startDate == null) {
629 if (ServerDetector.isTomcat()) {
630 startDate = new Date(System.currentTimeMillis() + Time.MINUTE);
631 }
632 else {
633 startDate = new Date(
634 System.currentTimeMillis() + Time.MINUTE * 3);
635 }
636 }
637
638 Trigger quartzTrigger = null;
639
640 TriggerType triggerType = trigger.getTriggerType();
641
642 if (triggerType.equals(TriggerType.CRON)) {
643 try {
644 TriggerBuilder<Trigger>triggerBuilder =
645 TriggerBuilder.newTrigger();
646
647 triggerBuilder.endAt(endDate);
648 triggerBuilder.forJob(jobName, groupName);
649 triggerBuilder.startAt(startDate);
650 triggerBuilder.withIdentity(jobName, groupName);
651
652 CronScheduleBuilder cronScheduleBuilder =
653 CronScheduleBuilder.cronSchedule(
654 (String)trigger.getTriggerContent());
655
656 triggerBuilder.withSchedule(cronScheduleBuilder);
657
658 quartzTrigger = triggerBuilder.build();
659 }
660 catch (ParseException pe) {
661 throw new SchedulerException(
662 "Unable to parse cron text " + trigger.getTriggerContent());
663 }
664 }
665 else if (triggerType.equals(TriggerType.SIMPLE)) {
666 long interval = (Long)trigger.getTriggerContent();
667
668 if (interval <= 0) {
669 if (_log.isDebugEnabled()) {
670 _log.debug(
671 "Not scheduling " + trigger.getJobName() +
672 " because interval is less than or equal to 0");
673 }
674
675 return null;
676 }
677
678 TriggerBuilder<Trigger>triggerBuilder = TriggerBuilder.newTrigger();
679
680 triggerBuilder.endAt(endDate);
681 triggerBuilder.forJob(jobName, groupName);
682 triggerBuilder.startAt(startDate);
683 triggerBuilder.withIdentity(jobName, groupName);
684
685 SimpleScheduleBuilder simpleScheduleBuilder =
686 SimpleScheduleBuilder.simpleSchedule();
687
688 simpleScheduleBuilder.withIntervalInMilliseconds(interval);
689 simpleScheduleBuilder.withRepeatCount(
690 SimpleTrigger.REPEAT_INDEFINITELY);
691
692 triggerBuilder.withSchedule(simpleScheduleBuilder);
693
694 quartzTrigger = triggerBuilder.build();
695 }
696 else {
697 throw new SchedulerException(
698 "Unknown trigger type " + trigger.getTriggerType());
699 }
700
701 return quartzTrigger;
702 }
703
704 protected SchedulerResponse getScheduledJob(
705 Scheduler scheduler, JobKey jobKey)
706 throws Exception {
707
708 JobDetail jobDetail = scheduler.getJobDetail(jobKey);
709
710 if (jobDetail == null) {
711 return null;
712 }
713
714 JobDataMap jobDataMap = jobDetail.getJobDataMap();
715
716 String description = jobDataMap.getString(DESCRIPTION);
717 String destinationName = jobDataMap.getString(DESTINATION_NAME);
718 Message message = getMessage(jobDataMap);
719 StorageType storageType = StorageType.valueOf(
720 jobDataMap.getString(STORAGE_TYPE));
721
722 SchedulerResponse schedulerResponse = null;
723
724 String jobName = jobKey.getName();
725 String groupName = jobKey.getGroup();
726
727 TriggerKey triggerKey = new TriggerKey(jobName, groupName);
728
729 Trigger trigger = scheduler.getTrigger(triggerKey);
730
731 JobState jobState = getJobState(jobDataMap);
732
733 message.put(JOB_STATE, jobState);
734
735 if (trigger == null) {
736 schedulerResponse = new SchedulerResponse();
737
738 schedulerResponse.setDescription(description);
739 schedulerResponse.setDestinationName(destinationName);
740 schedulerResponse.setGroupName(groupName);
741 schedulerResponse.setJobName(jobName);
742 schedulerResponse.setMessage(message);
743 schedulerResponse.setStorageType(storageType);
744 }
745 else {
746 message.put(END_TIME, trigger.getEndTime());
747 message.put(FINAL_FIRE_TIME, trigger.getFinalFireTime());
748 message.put(NEXT_FIRE_TIME, trigger.getNextFireTime());
749 message.put(PREVIOUS_FIRE_TIME, trigger.getPreviousFireTime());
750 message.put(START_TIME, trigger.getStartTime());
751
752 if (CronTrigger.class.isAssignableFrom(trigger.getClass())) {
753
754 CronTrigger cronTrigger = CronTrigger.class.cast(trigger);
755
756 schedulerResponse = new SchedulerResponse();
757
758 schedulerResponse.setDescription(description);
759 schedulerResponse.setDestinationName(destinationName);
760 schedulerResponse.setMessage(message);
761 schedulerResponse.setStorageType(storageType);
762 schedulerResponse.setTrigger(
763 new com.liferay.portal.kernel.scheduler.CronTrigger(
764 jobName, groupName, cronTrigger.getStartTime(),
765 cronTrigger.getEndTime(),
766 cronTrigger.getCronExpression()));
767 }
768 else if (SimpleTrigger.class.isAssignableFrom(trigger.getClass())) {
769 SimpleTrigger simpleTrigger = SimpleTrigger.class.cast(trigger);
770
771 schedulerResponse = new SchedulerResponse();
772
773 schedulerResponse.setDescription(description);
774 schedulerResponse.setDestinationName(destinationName);
775 schedulerResponse.setMessage(message);
776 schedulerResponse.setStorageType(storageType);
777 schedulerResponse.setTrigger(
778 new IntervalTrigger(
779 jobName, groupName, simpleTrigger.getStartTime(),
780 simpleTrigger.getEndTime(),
781 simpleTrigger.getRepeatInterval()));
782 }
783 }
784
785 return schedulerResponse;
786 }
787
788 protected List<SchedulerResponse> getScheduledJobs(
789 Scheduler scheduler, String groupName)
790 throws Exception {
791
792 groupName = fixMaxLength(
793 getOriginalGroupName(groupName), GROUP_NAME_MAX_LENGTH);
794
795 List<SchedulerResponse> schedulerResponses =
796 new ArrayList<SchedulerResponse>();
797
798 Set<JobKey> jobKeys = scheduler.getJobKeys(
799 GroupMatcher.jobGroupEquals(groupName));
800
801 for (JobKey jobKey : jobKeys) {
802 SchedulerResponse schedulerResponse = getScheduledJob(
803 scheduler, jobKey);
804
805 if (schedulerResponse != null) {
806 schedulerResponses.add(schedulerResponse);
807 }
808 }
809
810 return schedulerResponses;
811 }
812
813 protected Scheduler getScheduler(String groupName) throws Exception {
814 if (groupName.startsWith(StorageType.PERSISTED.toString())) {
815 return _persistedScheduler;
816 }
817 else {
818 return _memoryScheduler;
819 }
820 }
821
822 protected StorageType getStorageType(String groupName) {
823 int pos = groupName.indexOf(CharPool.POUND);
824
825 String storageTypeString = groupName.substring(0, pos);
826
827 return StorageType.valueOf(storageTypeString);
828 }
829
830 protected Scheduler initializeScheduler(
831 String propertiesPrefix, boolean useQuartzCluster)
832 throws Exception {
833
834 StdSchedulerFactory schedulerFactory = new StdSchedulerFactory();
835
836 Properties properties = PropsUtil.getProperties(propertiesPrefix, true);
837
838 if (useQuartzCluster) {
839 DB db = DBFactoryUtil.getDB();
840
841 String dbType = db.getType();
842
843 if (dbType.equals(DB.TYPE_SQLSERVER)) {
844 properties.setProperty(
845 "org.quartz.jobStore.lockHandler.class",
846 UpdateLockRowSemaphore.class.getName());
847 }
848
849 if (PropsValues.CLUSTER_LINK_ENABLED) {
850 if (dbType.equals(DB.TYPE_HYPERSONIC)) {
851 _log.error("Unable to cluster scheduler on Hypersonic");
852 }
853 else {
854 properties.put(
855 "org.quartz.jobStore.isClustered",
856 Boolean.TRUE.toString());
857 }
858 }
859 }
860
861 schedulerFactory.initialize(properties);
862
863 return schedulerFactory.getScheduler();
864 }
865
866 protected void initJobState() throws Exception {
867 List<String> groupNames = _persistedScheduler.getJobGroupNames();
868
869 for (String groupName : groupNames) {
870 Set<JobKey> jobkeys = _persistedScheduler.getJobKeys(
871 GroupMatcher.jobGroupEquals(groupName));
872
873 for (JobKey jobKey : jobkeys) {
874 Trigger trigger = _persistedScheduler.getTrigger(
875 new TriggerKey(jobKey.getName(), jobKey.getGroup()));
876
877 if (trigger != null) {
878 continue;
879 }
880
881 JobDetail jobDetail = _persistedScheduler.getJobDetail(jobKey);
882
883 JobDataMap jobDataMap = jobDetail.getJobDataMap();
884
885 Message message = getMessage(jobDataMap);
886
887 message.put(JOB_NAME, jobKey.getName());
888 message.put(GROUP_NAME, jobKey.getGroup());
889
890 SchedulerEngineHelperUtil.auditSchedulerJobs(
891 message, TriggerState.EXPIRED);
892
893 _persistedScheduler.deleteJob(jobKey);
894 }
895 }
896 }
897
898 protected void registerMessageListeners(
899 String jobName, String groupName, String destinationName,
900 Message message)
901 throws SchedulerException {
902
903 String messageListenerClassName = message.getString(
904 MESSAGE_LISTENER_CLASS_NAME);
905
906 if (Validator.isNull(messageListenerClassName)) {
907 return;
908 }
909
910 String portletId = message.getString(PORTLET_ID);
911
912 ClassLoader classLoader = null;
913
914 if (Validator.isNull(portletId)) {
915 classLoader = PACLClassLoaderUtil.getPortalClassLoader();
916 }
917 else {
918 classLoader = PortletClassLoaderUtil.getClassLoader(portletId);
919
920 if (classLoader == null) {
921
922
923
924
925
926 classLoader = ClassLoaderPool.getClassLoader(portletId);
927 }
928 }
929
930 if (classLoader == null) {
931 throw new SchedulerException(
932 "Unable to find class loader for portlet " + portletId);
933 }
934
935 MessageListener schedulerEventListener = getMessageListener(
936 messageListenerClassName, classLoader);
937
938 SchedulerEventMessageListenerWrapper schedulerEventListenerWrapper =
939 new SchedulerEventMessageListenerWrapper();
940
941 schedulerEventListenerWrapper.setGroupName(groupName);
942 schedulerEventListenerWrapper.setJobName(jobName);
943 schedulerEventListenerWrapper.setMessageListener(
944 schedulerEventListener);
945
946 schedulerEventListenerWrapper.afterPropertiesSet();
947
948 MessageBusUtil.registerMessageListener(
949 destinationName, schedulerEventListenerWrapper);
950
951 message.put(
952 MESSAGE_LISTENER_UUID,
953 schedulerEventListenerWrapper.getMessageListenerUUID());
954
955 message.put(RECEIVER_KEY, getFullName(jobName, groupName));
956 }
957
958 protected void schedule(
959 Scheduler scheduler, StorageType storageType, Trigger trigger,
960 String description, String destinationName, Message message)
961 throws Exception {
962
963 try {
964 JobBuilder jobBuilder = JobBuilder.newJob(MessageSenderJob.class);
965
966 jobBuilder.withIdentity(trigger.getJobKey());
967
968 JobDetail jobDetail = jobBuilder.build();
969
970 JobDataMap jobDataMap = jobDetail.getJobDataMap();
971
972 jobDataMap.put(DESCRIPTION, description);
973 jobDataMap.put(DESTINATION_NAME, destinationName);
974 jobDataMap.put(MESSAGE, JSONFactoryUtil.serialize(message));
975 jobDataMap.put(STORAGE_TYPE, storageType.toString());
976
977 JobState jobState = new JobState(
978 TriggerState.NORMAL, message.getInteger(EXCEPTIONS_MAX_SIZE));
979
980 jobDataMap.put(
981 JOB_STATE, JobStateSerializeUtil.serialize(jobState));
982
983 unregisterMessageListener(scheduler, trigger.getJobKey());
984
985 synchronized (this) {
986 scheduler.deleteJob(trigger.getJobKey());
987 scheduler.scheduleJob(jobDetail, trigger);
988 }
989 }
990 catch (ObjectAlreadyExistsException oaee) {
991 if (_log.isInfoEnabled()) {
992 _log.info("Message is already scheduled");
993 }
994 }
995 }
996
997 protected void unregisterMessageListener(Scheduler scheduler, JobKey jobKey)
998 throws Exception {
999
1000 JobDetail jobDetail = scheduler.getJobDetail(jobKey);
1001
1002 if (jobDetail == null) {
1003 return;
1004 }
1005
1006 JobDataMap jobDataMap = jobDetail.getJobDataMap();
1007
1008 if (jobDataMap == null) {
1009 return;
1010 }
1011
1012 Message message = getMessage(jobDataMap);
1013
1014 String messageListenerUUID = message.getString(MESSAGE_LISTENER_UUID);
1015
1016 if (messageListenerUUID == null) {
1017 return;
1018 }
1019
1020 String destinationName = jobDataMap.getString(DESTINATION_NAME);
1021
1022 MessageBus messageBus = MessageBusUtil.getMessageBus();
1023
1024 Destination destination = messageBus.getDestination(destinationName);
1025
1026 Set<MessageListener> messageListeners =
1027 destination.getMessageListeners();
1028
1029 for (MessageListener messageListener : messageListeners) {
1030 if (!(messageListener instanceof InvokerMessageListener)) {
1031 continue;
1032 }
1033
1034 InvokerMessageListener invokerMessageListener =
1035 (InvokerMessageListener)messageListener;
1036
1037 messageListener = invokerMessageListener.getMessageListener();
1038
1039 if (!(messageListener instanceof
1040 SchedulerEventMessageListenerWrapper)) {
1041
1042 continue;
1043 }
1044
1045 SchedulerEventMessageListenerWrapper schedulerMessageListener =
1046 (SchedulerEventMessageListenerWrapper)messageListener;
1047
1048 if (messageListenerUUID.equals(
1049 schedulerMessageListener.getMessageListenerUUID())) {
1050
1051 messageBus.unregisterMessageListener(
1052 destinationName, schedulerMessageListener);
1053
1054 return;
1055 }
1056 }
1057 }
1058
1059 protected void unschedule(Scheduler scheduler, JobKey jobKey)
1060 throws Exception {
1061
1062 JobDetail jobDetail = scheduler.getJobDetail(jobKey);
1063
1064 TriggerKey triggerKey = new TriggerKey(
1065 jobKey.getName(), jobKey.getGroup());
1066
1067 if (jobDetail == null) {
1068 return;
1069 }
1070
1071 unregisterMessageListener(scheduler, jobKey);
1072
1073 if (scheduler == _memoryScheduler) {
1074 scheduler.unscheduleJob(triggerKey);
1075
1076 return;
1077 }
1078
1079 JobDataMap jobDataMap = jobDetail.getJobDataMap();
1080
1081 JobState jobState = getJobState(jobDataMap);
1082
1083 Trigger trigger = scheduler.getTrigger(triggerKey);
1084
1085 jobState.setTriggerDate(END_TIME, new Date());
1086 jobState.setTriggerDate(FINAL_FIRE_TIME, trigger.getPreviousFireTime());
1087 jobState.setTriggerDate(NEXT_FIRE_TIME, null);
1088 jobState.setTriggerDate(
1089 PREVIOUS_FIRE_TIME, trigger.getPreviousFireTime());
1090 jobState.setTriggerDate(START_TIME, trigger.getStartTime());
1091
1092 jobState.setTriggerState(TriggerState.UNSCHEDULED);
1093
1094 jobState.clearExceptions();
1095
1096 jobDataMap.put(JOB_STATE, JobStateSerializeUtil.serialize(jobState));
1097
1098 scheduler.unscheduleJob(triggerKey);
1099
1100 scheduler.addJob(jobDetail, true);
1101 }
1102
1103 protected void update(
1104 Scheduler scheduler,
1105 com.liferay.portal.kernel.scheduler.Trigger trigger)
1106 throws Exception {
1107
1108 Trigger quartzTrigger = getQuartzTrigger(trigger);
1109
1110 if (quartzTrigger == null) {
1111 return;
1112 }
1113
1114 TriggerKey triggerKey = quartzTrigger.getKey();
1115
1116 if (scheduler.getTrigger(triggerKey) != null) {
1117 scheduler.rescheduleJob(triggerKey, quartzTrigger);
1118 }
1119 else {
1120 JobKey jobKey = quartzTrigger.getJobKey();
1121
1122 JobDetail jobDetail = scheduler.getJobDetail(jobKey);
1123
1124 if (jobDetail == null) {
1125 return;
1126 }
1127
1128 updateJobState(scheduler, jobKey, TriggerState.NORMAL, true);
1129
1130 synchronized (this) {
1131 scheduler.deleteJob(jobKey);
1132 scheduler.scheduleJob(jobDetail, quartzTrigger);
1133 }
1134 }
1135 }
1136
1137 protected void updateJobState(
1138 Scheduler scheduler, JobKey jobKey, TriggerState triggerState,
1139 boolean suppressError)
1140 throws Exception {
1141
1142 JobDetail jobDetail = scheduler.getJobDetail(jobKey);
1143
1144 JobDataMap jobDataMap = jobDetail.getJobDataMap();
1145
1146 JobState jobState = getJobState(jobDataMap);
1147
1148 if (triggerState != null) {
1149 jobState.setTriggerState(triggerState);
1150 }
1151
1152 if (suppressError) {
1153 jobState.clearExceptions();
1154 }
1155
1156 jobDataMap.put(JOB_STATE, JobStateSerializeUtil.serialize(jobState));
1157
1158 scheduler.addJob(jobDetail, true);
1159 }
1160
1161 @BeanReference(name = "com.liferay.portal.service.QuartzLocalService")
1162 protected QuartzLocalService quartzLocalService;
1163
1164 private static Log _log = LogFactoryUtil.getLog(
1165 QuartzSchedulerEngine.class);
1166
1167 private Scheduler _memoryScheduler;
1168 private Scheduler _persistedScheduler;
1169
1170 }