23298381073

Committed 19 Mar 2026 01:55PM UTC coverage: 63.092% (-0.008%) from 63.1%

Build # 23298381073

Build Type

push

github

Committed by

rkorytkowski

Commit Message

TRUNK-6558 Legacy task without creator fix

Coverage Stats

0 of 4 new or added lines in 2 files covered. (0.0%)

9 existing lines in 3 files now uncovered.

23132 of 36664 relevant lines covered (63.09%)

0.63 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

3.41

/api/src/main/java/org/openmrs/scheduler/jobrunr/JobRunrSchedulerService.java

/**
 * This Source Code Form is subject to the terms of the Mozilla Public License,
 * v. 2.0. If a copy of the MPL was not distributed with this file, You can
 * obtain one at http://mozilla.org/MPL/2.0/. OpenMRS is also distributed under
 * the terms of the Healthcare Disclaimer located at http://openmrs.org/license.
 *
 * Copyright (C) OpenMRS Inc. OpenMRS is a registered trademark and the OpenMRS
 * graphic logo is a trademark of OpenMRS Inc.
 */
package org.openmrs.scheduler.jobrunr;

import java.time.Duration;
import java.time.Instant;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.temporal.ChronoUnit;
import java.util.Collection;
import java.util.Date;
import java.util.Iterator;
import java.util.List;
import java.util.Optional;
import java.util.Spliterator;
import java.util.Spliterators;
import java.util.UUID;
import java.util.function.Consumer;
import java.util.stream.Collectors;
import java.util.stream.Stream;
import java.util.stream.StreamSupport;

import org.jobrunr.jobs.Job;
import org.jobrunr.jobs.JobDetails;
import org.jobrunr.jobs.JobId;
import org.jobrunr.jobs.RecurringJob;
import org.jobrunr.jobs.states.StateName;
import org.jobrunr.scheduling.JobRequestScheduler;
import org.jobrunr.scheduling.JobScheduler;
import org.jobrunr.storage.JobNotFoundException;
import org.jobrunr.storage.PageRequest;
import org.jobrunr.storage.StorageProvider;
import org.openmrs.User;
import org.openmrs.api.APIException;
import org.openmrs.api.context.Context;
import org.openmrs.api.impl.BaseOpenmrsService;
import org.openmrs.scheduler.RecurringTaskDetails;
import org.openmrs.scheduler.SchedulerException;
import org.openmrs.scheduler.SchedulerService;
import org.openmrs.scheduler.SchedulerUtil;
import org.openmrs.scheduler.Task;
import org.openmrs.scheduler.TaskData;
import org.openmrs.scheduler.TaskDefinition;
import org.openmrs.scheduler.TaskDetails;
import org.openmrs.scheduler.TaskState;
import org.openmrs.scheduler.db.SchedulerDAO;
import org.openmrs.util.PrivilegeConstants;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

/**
 * @since 2.9.x
 */
@Service("schedulerService")
@Transactional
public class JobRunrSchedulerService extends BaseOpenmrsService implements SchedulerService {
        
        private static final Logger log = LoggerFactory.getLogger(JobRunrSchedulerService.class);
        
        private JobRequestScheduler jobRequestScheduler;
        
        private JobScheduler jobScheduler;
        
        private StorageProvider storageProvider;
        
        private SchedulerDAO schedulerDAO;
        
        public JobRunrSchedulerService(StorageProvider storageProvider, JobRequestScheduler jobRequestScheduler, 
                                                                   JobScheduler jobScheduler, SchedulerDAO schedulerDAO) {
                this.jobRequestScheduler = jobRequestScheduler;
                this.jobScheduler = jobScheduler;
                this.storageProvider = storageProvider;
                this.schedulerDAO = schedulerDAO;
        }
        
        @Override
        public void onStartup() {
                for (TaskDefinition taskDefinition: schedulerDAO.getTasks()) {
                        if (Boolean.TRUE.equals(taskDefinition.getStartOnStartup())) {
                                String scheduledBy = taskDefinition.getCreator() != null ? taskDefinition.getCreator().getSystemId() : "daemon";
                                JobId jobId = jobRequestScheduler.enqueue(UUID.fromString(taskDefinition.getUuid()),
                                        new JobRequestAdapter(taskDefinition, scheduledBy));
                                String name = taskDefinition.getName();
                                if (name == null) {
                                        name = taskDefinition.getTaskClass();
                                }
                                updateJobWithName(jobId, name);
                                log.info("Scheduled legacy task {} [{}] to run on startup", name, taskDefinition.getUuid());
                        }
                        
                        Optional<RecurringTaskDetails> recurringTask = getRecurringTask(taskDefinition.getUuid());
                        if (!recurringTask.isPresent()) {
                                Optional<TaskDetails> task = getTask(taskDefinition.getUuid());
                                if (!task.isPresent()) {
                                        try {
                                                scheduleTask(taskDefinition);
                                                log.info("Scheduled legacy task {} [{}] to run on schedule", taskDefinition.getName(), taskDefinition.getUuid());
                                        } catch (SchedulerException e) {
                                                throw new APIException(e);
                                        }
                                }
                        }
                }
        }

        @Override
        public String getStatus(Integer id) {
                TaskDefinition task = getTask(id);
                if (task != null && Boolean.TRUE.equals(task.getStarted())) {
                        return "Scheduled";
                }
                return "Not Running";
        }

        @Override
        public void shutdownTask(TaskDefinition task) throws SchedulerException {
                if (task != null) {
                        getTask(task.getUuid()).ifPresent(t -> deleteTask(task.getUuid()));
                        getRecurringTask(task.getUuid()).ifPresent(t -> deleteRecurringTask(task.getUuid()));
                        task.setStarted(false);
                        saveTaskDefinition(task);
                }
        }

        @Override
        public Task scheduleTask(TaskDefinition task) throws SchedulerException {
                if (task != null) {
                        try {
                                String name = task.getName();
                                if (name == null) {
                                        name = task.getTaskClass();
                                }
                                String scheduledBy = task.getCreator() != null ? task.getCreator().getSystemId() : "daemon";
                                
                                if (task.getRepeatInterval() != null && task.getRepeatInterval() > 0) {
                                        if (task.getStartTime() == null) {
                                                String recurringJobId = jobRequestScheduler.scheduleRecurrently(task.getUuid(),
                                                        Duration.ofSeconds(task.getRepeatInterval()),
                                                        new JobRequestAdapter(task, scheduledBy));
                                                updateRecurringJobWithName(recurringJobId, name);
                                        } else {
                                                Date nextExecution = SchedulerUtil.getNextExecution(task);
                                                JobId jobId = jobScheduler.schedule(UUID.randomUUID(), nextExecution.toInstant(),
                                                        (JobRunrSchedulerService service) -> service.scheduleRecurrently(task.getUuid()));
                                                updateJobWithName(jobId, task.getName());
                                                // Create a placeholder recurring task that will be updated by the above task to the correct interval
                                                String recurringJobId = jobRequestScheduler.scheduleRecurrently(task.getUuid(),
                                                        Duration.between(Instant.now(), nextExecution.toInstant()).plus(1, ChronoUnit.DAYS),
                                                        new JobRequestAdapter(task, scheduledBy));
                                                updateRecurringJobWithName(recurringJobId, name);
                                        }
                                } else if (task.getStartTime() != null){
                                        Instant runAt = task.getStartTime().toInstant();
                                        JobId jobId = jobRequestScheduler.schedule(UUID.fromString(task.getUuid()), runAt, 
                                                new JobRequestAdapter(task, scheduledBy));
                                        updateJobWithName(jobId, name);
                                } else {
                                        JobId jobId = jobRequestScheduler.enqueue(UUID.fromString(task.getUuid()),
                                                new JobRequestAdapter(task, scheduledBy));
                                        updateJobWithName(jobId, name);
                                }
                                task.setStarted(true);
                                if (task.getId() != null) {
                                        schedulerDAO.updateTask(task);
                                } else {
                                        schedulerDAO.createTask(task);
                                }
                        }
                        catch (Exception e) {
                                throw new SchedulerException("Failed to schedule task", e);
                        }
                }
                return null;
        }

        /**
         * Helper task to schedule recurrently at a given time.
         * @param uuid
         */
        public void scheduleRecurrently(String uuid) {
                TaskDefinition task = getTaskByUuid(uuid);
                if (task != null) {
                        String scheduledBy = task.getCreator() != null ? task.getCreator().getSystemId() : "daemon";
                        String jobId = jobRequestScheduler.scheduleRecurrently(task.getUuid(), Duration.ofSeconds(task.getRepeatInterval()),
                                new JobRequestAdapter(task, scheduledBy));
                        updateRecurringJobWithName(jobId, task.getName());
                }
        }

        @Override
        public Task rescheduleTask(TaskDefinition task) throws SchedulerException {
                shutdownTask(task);
                return scheduleTask(task);
        }

        @Override
        public void rescheduleAllTasks() throws SchedulerException {
                for (TaskDefinition task : getScheduledTasks()) {
                        try {
                                rescheduleTask(task);
                        }
                        catch (Exception e) {
                                log.error("Failed to reschedule task " + task.getName(), e);
                        }
                }
        }

        @Override
        public Collection<TaskDefinition> getScheduledTasks() {
                return getRegisteredTasks().stream()
                                .filter(t -> Boolean.TRUE.equals(t.getStarted()))
                                .collect(Collectors.toList());
        }

        @Override
        public Collection<TaskDefinition> getRegisteredTasks() {
                return schedulerDAO.getTasks();
        }

        @Override
        public TaskDefinition getTask(Integer id) {
                return schedulerDAO.getTask(id);
        }

        @Override
        public TaskDefinition getTaskByUuid(String uuid) {
                return schedulerDAO.getTaskByUuid(uuid);
        }

        @Override
        public TaskDefinition getTaskByName(String name) {
                return schedulerDAO.getTaskByName(name);
        }

        @Override
        public void deleteTask(Integer id) {
                TaskDefinition task = getTask(id);
                if (task != null) {
                        if (Boolean.TRUE.equals(task.getStarted())) {
                                try {
                                        shutdownTask(task);
                                } catch (SchedulerException e) {
                                        throw new APIException(e);
                                }
                        }
                        schedulerDAO.deleteTask(id);
                }
        }

        @Override
        public void saveTaskDefinition(TaskDefinition task) {
                if (task.getId() != null) {
                        schedulerDAO.updateTask(task);
                } else {
                        schedulerDAO.createTask(task);
                }
        }

        @Override
        public void scheduleIfNotRunning(TaskDefinition taskDef) {
                if (taskDef != null && !Boolean.TRUE.equals(taskDef.getStarted())) {
                        try {
                                scheduleTask(taskDef);
                        } catch (SchedulerException e) {
                                throw new APIException(e);
                        }
                }
        }
        
        @Override
        public Optional<TaskDetails> getTask(String uuid) {
                try {
                        Job job = storageProvider.getJobById(JobId.parse(uuid));
                        if (!hasPrivileges(job.getJobDetails())) {
                                return Optional.empty();
                        } else if (StateName.DELETED.equals(job.getState())) {
                                return Optional.empty();
                        } else {
                                return Optional.of(new JobRunrTaskDetails(job));
                        }
                } catch (JobNotFoundException e) {
                        return Optional.empty();
                }
        }

        @Override
        public Optional<RecurringTaskDetails> getRecurringTask(String uuid) {
                return storageProvider.getRecurringJobs().stream()
                        .filter(r -> r.getId().equals(uuid) && hasPrivileges(r.getJobDetails()))
                        .findFirst().map(JobRunrRecurringTaskDetails::new);
        }
        
        @Override
        public void deleteTask(String uuid) {
                if (getTask(uuid).isPresent()) {
                        jobRequestScheduler.delete(JobId.parse(uuid));
                }
        }
        
        @Override
        public void deleteRecurringTask(String uuid) {
                if (getRecurringTask(uuid).isPresent()) {
                        jobRequestScheduler.delete(uuid);
                }
        }
        
        @Override
        public Stream<TaskDetails> getTasks(TaskState state, Instant before) {
                return StreamSupport.stream(new Spliterators.AbstractSpliterator<TaskDetails>(Long.MAX_VALUE, Spliterator.ORDERED | Spliterator.NONNULL) {
                        private int offset = 0;
                        private Iterator<Job> currentBatch;
                        private final boolean isSchedulerManager = isSchedulerManager(Context.getAuthenticatedUser());
                        private final String userSystemId = Context.getAuthenticatedUser().getSystemId();

                        @Override
                        public boolean tryAdvance(Consumer<? super TaskDetails> action) {
                                if (currentBatch == null || !currentBatch.hasNext()) {
                                        int limit = 100;
                                        List<Job> jobs = storageProvider.getJobs(StateName.valueOf(state.name()), before, PageRequest.ascOnUpdatedAt(offset, limit));
                                        if (jobs == null || jobs.isEmpty()) {
                                                return false;
                                        }
                                        if (!isSchedulerManager) {
                                                // List only tasks for the currently authenticated user
                                                currentBatch = jobs.stream().filter(job -> isScheduledBy(job.getJobDetails(), userSystemId)).iterator();
                                        } else {
                                                currentBatch = jobs.iterator();
                                        }
                                        offset += jobs.size();
                                }
                                
                                if (currentBatch.hasNext()) {
                                        action.accept(new JobRunrTaskDetails(currentBatch.next()));
                                        return true;
                                }
                                return false;
                        }
                }, false);
        }

        public boolean isSchedulerManager(User user) {
                return user.hasPrivilege(PrivilegeConstants.MANAGE_SCHEDULER);
        }
        
        public boolean hasPrivileges(JobDetails jobDetails) {
                User user = Context.getAuthenticatedUser();
                return isSchedulerManager(user) || isScheduledBy(jobDetails, user.getSystemId());
        }

        public boolean isScheduledBy(JobDetails jobDetails, String userSystemId) {
                return jobDetails.getJobParameters().stream()
                        .anyMatch(jp -> {
                                if (jp.getObject() instanceof JobRequestAdapter) {
                                        JobRequestAdapter jobRequestAdapter = (JobRequestAdapter) jp.getObject();
                                        return jobRequestAdapter.getUserSystemId().equals(userSystemId);
                                }
                                return false;
                        });
        }
        
        @Override
        public Stream<RecurringTaskDetails> getRecurringTasks() {
                User user = Context.getAuthenticatedUser();
                boolean isSchedulerManager = isSchedulerManager(user);
                return storageProvider.getRecurringJobs().stream().filter(
                        j -> isSchedulerManager || isScheduledBy(j.getJobDetails(), user.getSystemId()))
                        .map(JobRunrRecurringTaskDetails::new);
        }
        
        @Override
        public TaskDetails schedule(TaskData taskData) {
                return schedule(taskData.getClass().getSimpleName(), taskData);
        }

        @Override
        public TaskDetails schedule(String name, TaskData taskData) {
                String scheduledBy = getScheduledBySystemId();
                JobId jobId = jobRequestScheduler.enqueue(new JobRequestAdapter(taskData, scheduledBy));
                Job job = updateJobWithName(jobId, name);
                return new JobRunrTaskDetails(job);
        }
        
        private Job updateJobWithName(JobId jobId, String name) {
                try {
                        Job job = storageProvider.getJobById(jobId);
                        
                        if (name != null) {
                                job.setJobName(name);
                                job = storageProvider.save(job);
                        }
                        return job;
                } catch (JobNotFoundException e) {
                        throw new IllegalStateException(e);
                }
        }
        
        private RecurringJob updateRecurringJobWithName(String uuid, String name) {
                RecurringJob recurringJob = storageProvider.getRecurringJobs().stream()
                        .filter(rj -> rj.getId().equals(uuid)).findFirst()
                        .orElseThrow(() -> new APIException("Job " + uuid + " not found"));

                if (name != null) {
                        recurringJob.setJobName(name);
                        storageProvider.saveRecurringJob(recurringJob);
                }
                return recurringJob;
        }
        
        @Override
        public TaskDetails schedule(TaskData taskData, Instant runAt) {
                return schedule(taskData.getClass().getSimpleName(), taskData, runAt);
        }

        @Override
        public TaskDetails schedule(String name, TaskData taskData, Instant runAt) {
                String scheduledBy = getScheduledBySystemId();
                JobId jobId = jobRequestScheduler.schedule(runAt, new JobRequestAdapter(taskData, scheduledBy));
                Job job = updateJobWithName(jobId, name);
                return new JobRunrTaskDetails(job);
        }

        @Override
        public TaskDetails schedule(TaskData taskData, ZonedDateTime runAt) {
                return schedule(taskData.getClass().getSimpleName(), taskData, runAt);
        }

        @Override
        public TaskDetails schedule(String name, TaskData taskData, ZonedDateTime runAt) {
                String scheduledBy = getScheduledBySystemId();
                JobId jobId = jobRequestScheduler.schedule(runAt, new JobRequestAdapter(taskData, scheduledBy));
                Job job = updateJobWithName(jobId, name);
                return new JobRunrTaskDetails(job);
        }

        @Override
        public RecurringTaskDetails scheduleRecurrently(TaskData taskData, String cron) {
                return scheduleRecurrently(taskData.getClass().getSimpleName(), taskData, cron);
        }
        
        @Override
        public RecurringTaskDetails scheduleRecurrently(String name, TaskData taskData, String cron) {
                String scheduledBy = getScheduledBySystemId();
                String jobId = jobRequestScheduler.scheduleRecurrently(UUID.randomUUID().toString(), cron, 
                        new JobRequestAdapter(taskData, scheduledBy));
                RecurringJob job = updateRecurringJobWithName(jobId, name);
                return new JobRunrRecurringTaskDetails(job);
        }

        @Override
        public RecurringTaskDetails scheduleRecurrently(TaskData taskData, String cron, ZoneId zoneId) {
                return scheduleRecurrently(taskData.getClass().getSimpleName(), taskData, cron, zoneId);
        }

        @Override
        public RecurringTaskDetails scheduleRecurrently(String name, TaskData taskData, String cron, ZoneId zoneId) {
                String scheduledBy = getScheduledBySystemId();
                String jobId = jobRequestScheduler.scheduleRecurrently(UUID.randomUUID().toString(), cron, zoneId, 
                        new JobRequestAdapter(taskData, scheduledBy));
                RecurringJob job = updateRecurringJobWithName(jobId, name);
                return new JobRunrRecurringTaskDetails(job);
        }

        @Override
        public RecurringTaskDetails scheduleRecurrently(TaskData taskData, Duration interval) {
                return scheduleRecurrently(taskData.getClass().getSimpleName(), taskData, interval);
        }
        
        @Override
        public RecurringTaskDetails scheduleRecurrently(String name, TaskData taskData, Duration interval) {
                String scheduledBy = getScheduledBySystemId();
                String jobId = jobRequestScheduler.scheduleRecurrently(UUID.randomUUID().toString(), interval, 
                        new JobRequestAdapter(taskData, scheduledBy));
                RecurringJob job = updateRecurringJobWithName(jobId, name);
                return new JobRunrRecurringTaskDetails(job);
        }
        
        private String getScheduledBySystemId() {
                return Context.getAuthenticatedUser().getSystemId();
        }
}        

1	/**
2	* This Source Code Form is subject to the terms of the Mozilla Public License,
3	* v. 2.0. If a copy of the MPL was not distributed with this file, You can
4	* obtain one at http://mozilla.org/MPL/2.0/. OpenMRS is also distributed under
5	* the terms of the Healthcare Disclaimer located at http://openmrs.org/license.
6	*
7	* Copyright (C) OpenMRS Inc. OpenMRS is a registered trademark and the OpenMRS
8	* graphic logo is a trademark of OpenMRS Inc.
9	*/
10	package org.openmrs.scheduler.jobrunr;
11
12	import java.time.Duration;
13	import java.time.Instant;
14	import java.time.ZoneId;
15	import java.time.ZonedDateTime;
16	import java.time.temporal.ChronoUnit;
17	import java.util.Collection;
18	import java.util.Date;
19	import java.util.Iterator;
20	import java.util.List;
21	import java.util.Optional;
22	import java.util.Spliterator;
23	import java.util.Spliterators;
24	import java.util.UUID;
25	import java.util.function.Consumer;
26	import java.util.stream.Collectors;
27	import java.util.stream.Stream;
28	import java.util.stream.StreamSupport;
29
30	import org.jobrunr.jobs.Job;
31	import org.jobrunr.jobs.JobDetails;
32	import org.jobrunr.jobs.JobId;
33	import org.jobrunr.jobs.RecurringJob;
34	import org.jobrunr.jobs.states.StateName;
35	import org.jobrunr.scheduling.JobRequestScheduler;
36	import org.jobrunr.scheduling.JobScheduler;
37	import org.jobrunr.storage.JobNotFoundException;
38	import org.jobrunr.storage.PageRequest;
39	import org.jobrunr.storage.StorageProvider;
40	import org.openmrs.User;
41	import org.openmrs.api.APIException;
42	import org.openmrs.api.context.Context;
43	import org.openmrs.api.impl.BaseOpenmrsService;
44	import org.openmrs.scheduler.RecurringTaskDetails;
45	import org.openmrs.scheduler.SchedulerException;
46	import org.openmrs.scheduler.SchedulerService;
47	import org.openmrs.scheduler.SchedulerUtil;
48	import org.openmrs.scheduler.Task;
49	import org.openmrs.scheduler.TaskData;
50	import org.openmrs.scheduler.TaskDefinition;
51	import org.openmrs.scheduler.TaskDetails;
52	import org.openmrs.scheduler.TaskState;
53	import org.openmrs.scheduler.db.SchedulerDAO;
54	import org.openmrs.util.PrivilegeConstants;
55	import org.slf4j.Logger;
56	import org.slf4j.LoggerFactory;
57	import org.springframework.stereotype.Service;
58	import org.springframework.transaction.annotation.Transactional;
59
60	/**
61	* @since 2.9.x
62	*/
63	@Service("schedulerService")
64	@Transactional
65	public class JobRunrSchedulerService extends BaseOpenmrsService implements SchedulerService {
66
67	private static final Logger log = LoggerFactory.getLogger(JobRunrSchedulerService.class);	1✔
68
69	private JobRequestScheduler jobRequestScheduler;
70
71	private JobScheduler jobScheduler;
72
73	private StorageProvider storageProvider;
74
75	private SchedulerDAO schedulerDAO;
76
77	public JobRunrSchedulerService(StorageProvider storageProvider, JobRequestScheduler jobRequestScheduler,
78	JobScheduler jobScheduler, SchedulerDAO schedulerDAO) {	1✔
79	this.jobRequestScheduler = jobRequestScheduler;	1✔
80	this.jobScheduler = jobScheduler;	1✔
81	this.storageProvider = storageProvider;	1✔
82	this.schedulerDAO = schedulerDAO;	1✔
83	}	1✔
84
85	@Override
86	public void onStartup() {
87	for (TaskDefinition taskDefinition: schedulerDAO.getTasks()) {	×
88	if (Boolean.TRUE.equals(taskDefinition.getStartOnStartup())) {	×
NEW 89	String scheduledBy = taskDefinition.getCreator() != null ? taskDefinition.getCreator().getSystemId() : "daemon";	×
90	JobId jobId = jobRequestScheduler.enqueue(UUID.fromString(taskDefinition.getUuid()),	×
91	new JobRequestAdapter(taskDefinition, scheduledBy));
92	String name = taskDefinition.getName();	×
93	if (name == null) {	×
94	name = taskDefinition.getTaskClass();	×
95	}
96	updateJobWithName(jobId, name);	×
97	log.info("Scheduled legacy task {} [{}] to run on startup", name, taskDefinition.getUuid());	×
98	}
99
100	Optional<RecurringTaskDetails> recurringTask = getRecurringTask(taskDefinition.getUuid());	×
101	if (!recurringTask.isPresent()) {	×
102	Optional<TaskDetails> task = getTask(taskDefinition.getUuid());	×
103	if (!task.isPresent()) {	×
104	try {
105	scheduleTask(taskDefinition);	×
106	log.info("Scheduled legacy task {} [{}] to run on schedule", taskDefinition.getName(), taskDefinition.getUuid());	×
107	} catch (SchedulerException e) {	×
108	throw new APIException(e);	×
109	}	×
110	}
111	}
112	}	×
113	}	×
114
115	@Override
116	public String getStatus(Integer id) {
117	TaskDefinition task = getTask(id);	×
118	if (task != null && Boolean.TRUE.equals(task.getStarted())) {	×
119	return "Scheduled";	×
120	}
121	return "Not Running";	×
122	}
123
124	@Override
125	public void shutdownTask(TaskDefinition task) throws SchedulerException {
126	if (task != null) {	×
127	getTask(task.getUuid()).ifPresent(t -> deleteTask(task.getUuid()));	×
128	getRecurringTask(task.getUuid()).ifPresent(t -> deleteRecurringTask(task.getUuid()));	×
129	task.setStarted(false);	×
130	saveTaskDefinition(task);	×
131	}
132	}	×
133
134	@Override
135	public Task scheduleTask(TaskDefinition task) throws SchedulerException {
136	if (task != null) {	×
137	try {
138	String name = task.getName();	×
139	if (name == null) {	×
140	name = task.getTaskClass();	×
141	}
NEW 142	String scheduledBy = task.getCreator() != null ? task.getCreator().getSystemId() : "daemon";	×
143
144	if (task.getRepeatInterval() != null && task.getRepeatInterval() > 0) {	×
145	if (task.getStartTime() == null) {	×
146	String recurringJobId = jobRequestScheduler.scheduleRecurrently(task.getUuid(),	×
147	Duration.ofSeconds(task.getRepeatInterval()),	×
148	new JobRequestAdapter(task, scheduledBy));
149	updateRecurringJobWithName(recurringJobId, name);	×
150	} else {	×
151	Date nextExecution = SchedulerUtil.getNextExecution(task);	×
152	JobId jobId = jobScheduler.schedule(UUID.randomUUID(), nextExecution.toInstant(),	×
153	(JobRunrSchedulerService service) -> service.scheduleRecurrently(task.getUuid()));	×
154	updateJobWithName(jobId, task.getName());	×
155	// Create a placeholder recurring task that will be updated by the above task to the correct interval
156	String recurringJobId = jobRequestScheduler.scheduleRecurrently(task.getUuid(),	×
157	Duration.between(Instant.now(), nextExecution.toInstant()).plus(1, ChronoUnit.DAYS),	×
158	new JobRequestAdapter(task, scheduledBy));
159	updateRecurringJobWithName(recurringJobId, name);	×
160	}	×
161	} else if (task.getStartTime() != null){	×
162	Instant runAt = task.getStartTime().toInstant();	×
163	JobId jobId = jobRequestScheduler.schedule(UUID.fromString(task.getUuid()), runAt,	×
164	new JobRequestAdapter(task, scheduledBy));
165	updateJobWithName(jobId, name);	×
166	} else {	×
167	JobId jobId = jobRequestScheduler.enqueue(UUID.fromString(task.getUuid()),	×
168	new JobRequestAdapter(task, scheduledBy));
169	updateJobWithName(jobId, name);	×
170	}
171	task.setStarted(true);	×
172	if (task.getId() != null) {	×
173	schedulerDAO.updateTask(task);	×
174	} else {
175	schedulerDAO.createTask(task);	×
176	}
177	}
178	catch (Exception e) {	×
179	throw new SchedulerException("Failed to schedule task", e);	×
180	}	×
181	}
182	return null;	×
183	}
184
185	/**
186	* Helper task to schedule recurrently at a given time.
187	* @param uuid
188	*/
189	public void scheduleRecurrently(String uuid) {
190	TaskDefinition task = getTaskByUuid(uuid);	×
191	if (task != null) {	×
NEW 192	String scheduledBy = task.getCreator() != null ? task.getCreator().getSystemId() : "daemon";	×
193	String jobId = jobRequestScheduler.scheduleRecurrently(task.getUuid(), Duration.ofSeconds(task.getRepeatInterval()),	×
194	new JobRequestAdapter(task, scheduledBy));
195	updateRecurringJobWithName(jobId, task.getName());	×
196	}
197	}	×
198
199	@Override
200	public Task rescheduleTask(TaskDefinition task) throws SchedulerException {
201	shutdownTask(task);	×
202	return scheduleTask(task);	×
203	}
204
205	@Override
206	public void rescheduleAllTasks() throws SchedulerException {
207	for (TaskDefinition task : getScheduledTasks()) {	×
208	try {
209	rescheduleTask(task);	×
210	}
211	catch (Exception e) {	×
212	log.error("Failed to reschedule task " + task.getName(), e);	×
213	}	×
214	}	×
215	}	×
216
217	@Override
218	public Collection<TaskDefinition> getScheduledTasks() {
219	return getRegisteredTasks().stream()	×
220	.filter(t -> Boolean.TRUE.equals(t.getStarted()))	×
221	.collect(Collectors.toList());	×
222	}
223
224	@Override
225	public Collection<TaskDefinition> getRegisteredTasks() {
226	return schedulerDAO.getTasks();	×
227	}
228
229	@Override
230	public TaskDefinition getTask(Integer id) {
231	return schedulerDAO.getTask(id);	×
232	}
233
234	@Override
235	public TaskDefinition getTaskByUuid(String uuid) {
236	return schedulerDAO.getTaskByUuid(uuid);	×
237	}
238
239	@Override
240	public TaskDefinition getTaskByName(String name) {
241	return schedulerDAO.getTaskByName(name);	×
242	}
243
244	@Override
245	public void deleteTask(Integer id) {
246	TaskDefinition task = getTask(id);	×
247	if (task != null) {	×
248	if (Boolean.TRUE.equals(task.getStarted())) {	×
249	try {
250	shutdownTask(task);	×
251	} catch (SchedulerException e) {	×
252	throw new APIException(e);	×
253	}	×
254	}
255	schedulerDAO.deleteTask(id);	×
256	}
257	}	×
258
259	@Override
260	public void saveTaskDefinition(TaskDefinition task) {
261	if (task.getId() != null) {	×
262	schedulerDAO.updateTask(task);	×
263	} else {
264	schedulerDAO.createTask(task);	×
265	}
266	}	×
267
268	@Override
269	public void scheduleIfNotRunning(TaskDefinition taskDef) {
270	if (taskDef != null && !Boolean.TRUE.equals(taskDef.getStarted())) {	×
271	try {
272	scheduleTask(taskDef);	×
273	} catch (SchedulerException e) {	×
274	throw new APIException(e);	×
275	}	×
276	}
277	}	×
278
279	@Override
280	public Optional<TaskDetails> getTask(String uuid) {
281	try {
282	Job job = storageProvider.getJobById(JobId.parse(uuid));	×
283	if (!hasPrivileges(job.getJobDetails())) {	×
284	return Optional.empty();	×
285	} else if (StateName.DELETED.equals(job.getState())) {	×
286	return Optional.empty();	×
287	} else {
288	return Optional.of(new JobRunrTaskDetails(job));	×
289	}
290	} catch (JobNotFoundException e) {	×
291	return Optional.empty();	×
292	}
293	}
294
295	@Override
296	public Optional<RecurringTaskDetails> getRecurringTask(String uuid) {
297	return storageProvider.getRecurringJobs().stream()	×
298	.filter(r -> r.getId().equals(uuid) && hasPrivileges(r.getJobDetails()))	×
299	.findFirst().map(JobRunrRecurringTaskDetails::new);	×
300	}
301
302	@Override
303	public void deleteTask(String uuid) {
304	if (getTask(uuid).isPresent()) {	×
305	jobRequestScheduler.delete(JobId.parse(uuid));	×
306	}
307	}	×
308
309	@Override
310	public void deleteRecurringTask(String uuid) {
311	if (getRecurringTask(uuid).isPresent()) {	×
312	jobRequestScheduler.delete(uuid);	×
313	}
314	}	×
315
316	@Override
317	public Stream<TaskDetails> getTasks(TaskState state, Instant before) {
318	return StreamSupport.stream(new Spliterators.AbstractSpliterator<TaskDetails>(Long.MAX_VALUE, Spliterator.ORDERED \| Spliterator.NONNULL) {	×
319	private int offset = 0;	×
320	private Iterator<Job> currentBatch;
321	private final boolean isSchedulerManager = isSchedulerManager(Context.getAuthenticatedUser());	×
322	private final String userSystemId = Context.getAuthenticatedUser().getSystemId();	×
323
324	@Override
325	public boolean tryAdvance(Consumer<? super TaskDetails> action) {
326	if (currentBatch == null \|\| !currentBatch.hasNext()) {	×
327	int limit = 100;	×
328	List<Job> jobs = storageProvider.getJobs(StateName.valueOf(state.name()), before, PageRequest.ascOnUpdatedAt(offset, limit));	×
329	if (jobs == null \|\| jobs.isEmpty()) {	×
330	return false;	×
331	}
332	if (!isSchedulerManager) {	×
333	// List only tasks for the currently authenticated user
334	currentBatch = jobs.stream().filter(job -> isScheduledBy(job.getJobDetails(), userSystemId)).iterator();	×
335	} else {
336	currentBatch = jobs.iterator();	×
337	}
338	offset += jobs.size();	×
339	}
340
341	if (currentBatch.hasNext()) {	×
342	action.accept(new JobRunrTaskDetails(currentBatch.next()));	×
343	return true;	×
344	}
345	return false;	×
346	}
347	}, false);
348	}
349
350	public boolean isSchedulerManager(User user) {
351	return user.hasPrivilege(PrivilegeConstants.MANAGE_SCHEDULER);	×
352	}
353
354	public boolean hasPrivileges(JobDetails jobDetails) {
355	User user = Context.getAuthenticatedUser();	×
356	return isSchedulerManager(user) \|\| isScheduledBy(jobDetails, user.getSystemId());	×
357	}
358
359	public boolean isScheduledBy(JobDetails jobDetails, String userSystemId) {
360	return jobDetails.getJobParameters().stream()	×
361	.anyMatch(jp -> {	×
362	if (jp.getObject() instanceof JobRequestAdapter) {	×
363	JobRequestAdapter jobRequestAdapter = (JobRequestAdapter) jp.getObject();	×
364	return jobRequestAdapter.getUserSystemId().equals(userSystemId);	×
365	}
366	return false;	×
367	});
368	}
369
370	@Override
371	public Stream<RecurringTaskDetails> getRecurringTasks() {
372	User user = Context.getAuthenticatedUser();	×
373	boolean isSchedulerManager = isSchedulerManager(user);	×
374	return storageProvider.getRecurringJobs().stream().filter(	×
375	j -> isSchedulerManager \|\| isScheduledBy(j.getJobDetails(), user.getSystemId()))	×
376	.map(JobRunrRecurringTaskDetails::new);	×
377	}
378
379	@Override
380	public TaskDetails schedule(TaskData taskData) {
381	return schedule(taskData.getClass().getSimpleName(), taskData);	×
382	}
383
384	@Override
385	public TaskDetails schedule(String name, TaskData taskData) {
386	String scheduledBy = getScheduledBySystemId();	×
387	JobId jobId = jobRequestScheduler.enqueue(new JobRequestAdapter(taskData, scheduledBy));	×
388	Job job = updateJobWithName(jobId, name);	×
389	return new JobRunrTaskDetails(job);	×
390	}
391
392	private Job updateJobWithName(JobId jobId, String name) {
393	try {
394	Job job = storageProvider.getJobById(jobId);	×
395
396	if (name != null) {	×
397	job.setJobName(name);	×
398	job = storageProvider.save(job);	×
399	}
400	return job;	×
401	} catch (JobNotFoundException e) {	×
402	throw new IllegalStateException(e);	×
403	}
404	}
405
406	private RecurringJob updateRecurringJobWithName(String uuid, String name) {
407	RecurringJob recurringJob = storageProvider.getRecurringJobs().stream()	×
408	.filter(rj -> rj.getId().equals(uuid)).findFirst()	×
409	.orElseThrow(() -> new APIException("Job " + uuid + " not found"));	×
410
411	if (name != null) {	×
412	recurringJob.setJobName(name);	×
413	storageProvider.saveRecurringJob(recurringJob);	×
414	}
415	return recurringJob;	×
416	}
417
418	@Override
419	public TaskDetails schedule(TaskData taskData, Instant runAt) {
420	return schedule(taskData.getClass().getSimpleName(), taskData, runAt);	×
421	}
422
423	@Override
424	public TaskDetails schedule(String name, TaskData taskData, Instant runAt) {
425	String scheduledBy = getScheduledBySystemId();	×
426	JobId jobId = jobRequestScheduler.schedule(runAt, new JobRequestAdapter(taskData, scheduledBy));	×
427	Job job = updateJobWithName(jobId, name);	×
428	return new JobRunrTaskDetails(job);	×
429	}
430
431	@Override
432	public TaskDetails schedule(TaskData taskData, ZonedDateTime runAt) {
433	return schedule(taskData.getClass().getSimpleName(), taskData, runAt);	×
434	}
435
436	@Override
437	public TaskDetails schedule(String name, TaskData taskData, ZonedDateTime runAt) {
438	String scheduledBy = getScheduledBySystemId();	×
439	JobId jobId = jobRequestScheduler.schedule(runAt, new JobRequestAdapter(taskData, scheduledBy));	×
440	Job job = updateJobWithName(jobId, name);	×
441	return new JobRunrTaskDetails(job);	×
442	}
443
444	@Override
445	public RecurringTaskDetails scheduleRecurrently(TaskData taskData, String cron) {
446	return scheduleRecurrently(taskData.getClass().getSimpleName(), taskData, cron);	×
447	}
448
449	@Override
450	public RecurringTaskDetails scheduleRecurrently(String name, TaskData taskData, String cron) {
451	String scheduledBy = getScheduledBySystemId();	×
452	String jobId = jobRequestScheduler.scheduleRecurrently(UUID.randomUUID().toString(), cron,	×
453	new JobRequestAdapter(taskData, scheduledBy));
454	RecurringJob job = updateRecurringJobWithName(jobId, name);	×
455	return new JobRunrRecurringTaskDetails(job);	×
456	}
457
458	@Override
459	public RecurringTaskDetails scheduleRecurrently(TaskData taskData, String cron, ZoneId zoneId) {
460	return scheduleRecurrently(taskData.getClass().getSimpleName(), taskData, cron, zoneId);	×
461	}
462
463	@Override
464	public RecurringTaskDetails scheduleRecurrently(String name, TaskData taskData, String cron, ZoneId zoneId) {
465	String scheduledBy = getScheduledBySystemId();	×
466	String jobId = jobRequestScheduler.scheduleRecurrently(UUID.randomUUID().toString(), cron, zoneId,	×
467	new JobRequestAdapter(taskData, scheduledBy));
468	RecurringJob job = updateRecurringJobWithName(jobId, name);	×
469	return new JobRunrRecurringTaskDetails(job);	×
470	}
471
472	@Override
473	public RecurringTaskDetails scheduleRecurrently(TaskData taskData, Duration interval) {
474	return scheduleRecurrently(taskData.getClass().getSimpleName(), taskData, interval);	×
475	}
476
477	@Override
478	public RecurringTaskDetails scheduleRecurrently(String name, TaskData taskData, Duration interval) {
479	String scheduledBy = getScheduledBySystemId();	×
480	String jobId = jobRequestScheduler.scheduleRecurrently(UUID.randomUUID().toString(), interval,	×
481	new JobRequestAdapter(taskData, scheduledBy));
482	RecurringJob job = updateRecurringJobWithName(jobId, name);	×
483	return new JobRunrRecurringTaskDetails(job);	×
484	}
485
486	private String getScheduledBySystemId() {
487	return Context.getAuthenticatedUser().getSystemId();	×
488	}
489	}

openmrs / openmrs-core / 23298381073

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous