1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
|
package org.onap.vid.job.impl;
import org.apache.commons.lang3.StringUtils;
import org.onap.vid.job.Job;
import org.onap.vid.job.JobCommand;
import org.onap.vid.job.JobsBrokerService;
import org.onap.vid.job.NextCommand;
import org.onap.vid.job.command.JobCommandFactory;
import org.onap.vid.properties.Features;
import org.onap.portalsdk.core.logging.logic.EELFLoggerDelegate;
import org.quartz.JobExecutionContext;
import org.springframework.scheduling.quartz.QuartzJobBean;
import org.springframework.stereotype.Component;
import org.togglz.core.manager.FeatureManager;
import java.util.Optional;
import java.util.UUID;
import static org.onap.vid.job.Job.JobStatus.FAILED;
import static org.onap.vid.job.Job.JobStatus.STOPPED;
@Component
public class JobWorker extends QuartzJobBean {
private static final EELFLoggerDelegate LOGGER = EELFLoggerDelegate.getLogger(JobWorker.class);
private JobsBrokerService jobsBrokerService;
private FeatureManager featureManager;
private JobCommandFactory jobCommandFactory;
private Job.JobStatus topic;
@Override
protected void executeInternal(JobExecutionContext context) {
Optional<Job> job;
if (!isMsoNewApiActive()) {
return;
}
job = pullJob();
while (job.isPresent()) {
Job nextJob = executeJobAndGetNext(job.get());
pushBack(nextJob);
job = pullJob();
}
}
private Optional<Job> pullJob() {
try {
return jobsBrokerService.pull(topic, UUID.randomUUID().toString());
} catch (Exception e) {
LOGGER.error(EELFLoggerDelegate.errorLogger, "failed to pull job from queue, breaking: {}", e, e);
return Optional.empty();
}
}
private void pushBack(Job nextJob) {
try {
jobsBrokerService.pushBack(nextJob);
} catch (Exception e) {
LOGGER.error(EELFLoggerDelegate.errorLogger, "failed pushing back job to queue: {}", e, e);
}
}
protected Job executeJobAndGetNext(Job job) {
LOGGER.debug(EELFLoggerDelegate.debugLogger, "going to execute job {} of {}: {}/{}",
StringUtils.substring(String.valueOf(job.getUuid()), 0, 8),
StringUtils.substring(String.valueOf(job.getTemplateId()), 0, 8),
job.getStatus(), job.getType());
NextCommand nextCommand = executeCommandAndGetNext(job);
Job nextJob = setNextCommandInJob(nextCommand, job);
return nextJob;
}
private NextCommand executeCommandAndGetNext(Job job) {
NextCommand nextCommand;
try {
final JobCommand jobCommand = jobCommandFactory.toCommand(job);
nextCommand = jobCommand.call();
} catch (Exception e) {
LOGGER.error(EELFLoggerDelegate.errorLogger, "error while executing job from queue: {}", e, e);
nextCommand = new NextCommand(FAILED);
}
if (nextCommand == null) {
nextCommand = new NextCommand(STOPPED);
}
return nextCommand;
}
private Job setNextCommandInJob(NextCommand nextCommand, Job job) {
LOGGER.debug(EELFLoggerDelegate.debugLogger, "transforming job {} of {}: {}/{} -> {}{}",
StringUtils.substring(String.valueOf(job.getUuid()), 0, 8),
StringUtils.substring(String.valueOf(job.getTemplateId()), 0, 8),
job.getStatus(), job.getType(),
nextCommand.getStatus(),
nextCommand.getCommand() != null ? ("/" + nextCommand.getCommand().getType()) : "");
job.setStatus(nextCommand.getStatus());
if (nextCommand.getCommand() != null) {
job.setTypeAndData(nextCommand.getCommand().getType(), nextCommand.getCommand().getData());
}
return job;
}
private boolean isMsoNewApiActive() {
return featureManager.isActive(Features.FLAG_ASYNC_INSTANTIATION);
}
//used by quartz to inject JobsBrokerService into the job
//see JobSchedulerInitializer
public void setJobsBrokerService(JobsBrokerService jobsBrokerService) {
this.jobsBrokerService = jobsBrokerService;
}
public void setFeatureManager(FeatureManager featureManager) {
this.featureManager = featureManager;
}
public void setJobCommandFactory(JobCommandFactory jobCommandFactory) {
this.jobCommandFactory = jobCommandFactory;
}
public void setTopic(Job.JobStatus topic) {
this.topic = topic;
}
}
|
