001/*
002 * Copyright 2006-2018 the original author or authors.
003 *
004 * Licensed under the Apache License, Version 2.0 (the "License");
005 * you may not use this file except in compliance with the License.
006 * You may obtain a copy of the License at
007 *
008 *      https://www.apache.org/licenses/LICENSE-2.0
009 *
010 * Unless required by applicable law or agreed to in writing, software
011 * distributed under the License is distributed on an "AS IS" BASIS,
012 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
013 * See the License for the specific language governing permissions and
014 * limitations under the License.
015 */
016package org.springframework.batch.core.configuration.support;
017
018import java.util.ArrayList;
019import java.util.Collection;
020import java.util.HashSet;
021import java.util.Map;
022import java.util.concurrent.ConcurrentHashMap;
023
024import org.apache.commons.logging.Log;
025import org.apache.commons.logging.LogFactory;
026
027import org.springframework.batch.core.Job;
028import org.springframework.batch.core.Step;
029import org.springframework.batch.core.configuration.DuplicateJobException;
030import org.springframework.batch.core.configuration.JobFactory;
031import org.springframework.batch.core.configuration.JobRegistry;
032import org.springframework.batch.core.configuration.StepRegistry;
033import org.springframework.batch.core.launch.NoSuchJobException;
034import org.springframework.batch.core.step.StepLocator;
035import org.springframework.beans.factory.InitializingBean;
036import org.springframework.context.ApplicationContext;
037import org.springframework.context.ConfigurableApplicationContext;
038import org.springframework.lang.Nullable;
039import org.springframework.util.Assert;
040
041/**
042 * Default implementation of {@link JobLoader}. Uses a {@link JobRegistry} to
043 * manage a population of loaded jobs and clears them up when asked. An optional
044 * {@link StepRegistry} might also be set to register the step(s) available for
045 * each registered job.
046 *
047 * @author Dave Syer
048 * @author Stephane Nicoll
049 * @author Mahmoud Ben Hassine
050 */
051public class DefaultJobLoader implements JobLoader, InitializingBean {
052
053        private static Log logger = LogFactory.getLog(DefaultJobLoader.class);
054
055        private JobRegistry jobRegistry;
056        private StepRegistry stepRegistry;
057
058        private Map<ApplicationContextFactory, ConfigurableApplicationContext> contexts = new ConcurrentHashMap<ApplicationContextFactory, ConfigurableApplicationContext>();
059
060        private Map<ConfigurableApplicationContext, Collection<String>> contextToJobNames = new ConcurrentHashMap<ConfigurableApplicationContext, Collection<String>>();
061
062        /**
063         * Default constructor useful for declarative configuration.
064         */
065        public DefaultJobLoader() {
066                this(null, null);
067        }
068
069        /**
070         * Creates a job loader with the job registry provided.
071         *
072         * @param jobRegistry a {@link JobRegistry}
073         */
074        public DefaultJobLoader(JobRegistry jobRegistry) {
075                this(jobRegistry, null);
076        }
077
078        /**
079         * Creates a job loader with the job and step registries provided.
080         *
081         * @param jobRegistry a {@link JobRegistry}
082         * @param stepRegistry a {@link StepRegistry} (can be {@code null})
083         */
084        public DefaultJobLoader(JobRegistry jobRegistry, @Nullable StepRegistry stepRegistry) {
085                this.jobRegistry = jobRegistry;
086                this.stepRegistry = stepRegistry;
087        }
088
089        /**
090         * The {@link JobRegistry} to use for jobs created.
091         *
092         * @param jobRegistry the job registry
093         */
094        public void setJobRegistry(JobRegistry jobRegistry) {
095                this.jobRegistry = jobRegistry;
096        }
097
098        /**
099         * The {@link StepRegistry} to use for the steps of created jobs.
100         *
101         * @param stepRegistry the step registry
102         */
103        public void setStepRegistry(StepRegistry stepRegistry) {
104                this.stepRegistry = stepRegistry;
105        }
106
107        /**
108         * Unregister all the jobs and close all the contexts created by this
109         * loader.
110         *
111         * @see JobLoader#clear()
112         */
113        @Override
114        public void clear() {
115                for (ConfigurableApplicationContext context : contexts.values()) {
116                        if (context.isActive()) {
117                                context.close();
118                        }
119                }
120                for (String jobName : jobRegistry.getJobNames()) {
121                        doUnregister(jobName);
122                }
123                contexts.clear();
124                contextToJobNames.clear();
125        }
126
127        @Override
128        public Collection<Job> reload(ApplicationContextFactory factory) {
129
130                // If the same factory is loaded twice the context can be closed
131                if (contexts.containsKey(factory)) {
132                        ConfigurableApplicationContext context = contexts.get(factory);
133                        for (String name : contextToJobNames.get(context)) {
134                                if (logger.isDebugEnabled()) {
135                                        logger.debug("Unregistering job: " + name + " from context: " + context.getDisplayName());
136                                }
137                                doUnregister(name);
138                        }
139                        context.close();
140                }
141
142                try {
143                        return doLoad(factory, true);
144                }
145                catch (DuplicateJobException e) {
146                        throw new IllegalStateException("Found duplicate job in reload (it should have been unregistered "
147                                        + "if it was previously registered in this loader)", e);
148                }
149        }
150
151        @Override
152        public Collection<Job> load(ApplicationContextFactory factory) throws DuplicateJobException {
153                return doLoad(factory, false);
154        }
155
156        @SuppressWarnings("resource")
157        private Collection<Job> doLoad(ApplicationContextFactory factory, boolean unregister) throws DuplicateJobException {
158
159                Collection<String> jobNamesBefore = jobRegistry.getJobNames();
160                ConfigurableApplicationContext context = factory.createApplicationContext();
161                Collection<String> jobNamesAfter = jobRegistry.getJobNames();
162                // Try to detect auto-registration (e.g. through a bean post processor)
163                boolean autoRegistrationDetected = jobNamesAfter.size() > jobNamesBefore.size();
164
165                Collection<String> jobsRegistered = new HashSet<String>();
166                if (autoRegistrationDetected) {
167                        for (String name : jobNamesAfter) {
168                                if (!jobNamesBefore.contains(name)) {
169                                        jobsRegistered.add(name);
170                                }
171                        }
172                }
173
174                contexts.put(factory, context);
175                String[] names = context.getBeanNamesForType(Job.class);
176
177                for (String name : names) {
178
179                        if (!autoRegistrationDetected) {
180
181                                Job job = (Job) context.getBean(name);
182                                String jobName = job.getName();
183
184                                // On reload try to unregister first
185                                if (unregister) {
186                                        if (logger.isDebugEnabled()) {
187                                                logger.debug("Unregistering job: " + jobName + " from context: " + context.getDisplayName());
188                                        }
189                                        doUnregister(jobName);
190                                }
191
192                                if (logger.isDebugEnabled()) {
193                                        logger.debug("Registering job: " + jobName + " from context: " + context.getDisplayName());
194                                }
195                                doRegister(context, job);
196                                jobsRegistered.add(jobName);
197                        }
198
199                }
200
201                Collection<Job> result = new ArrayList<Job>();
202                for (String name : jobsRegistered) {
203                        try {
204                                result.add(jobRegistry.getJob(name));
205                        }
206                        catch (NoSuchJobException e) {
207                                // should not happen;
208                                throw new IllegalStateException("Could not retrieve job that was should have been registered", e);
209                        }
210
211                }
212
213                contextToJobNames.put(context, jobsRegistered);
214
215                return result;
216
217        }
218
219        /**
220         * Returns all the {@link Step} instances defined by the specified {@link StepLocator}.
221         * <br>
222         * The specified <tt>jobApplicationContext</tt> is used to collect additional steps that
223         * are not exposed by the step locator
224         *
225         * @param stepLocator the given step locator
226         * @param jobApplicationContext the application context of the job
227         * @return all the {@link Step} defined by the given step locator and context
228         * @see StepLocator
229         */
230        private Collection<Step> getSteps(final StepLocator stepLocator, final ApplicationContext jobApplicationContext) {
231                final Collection<String> stepNames = stepLocator.getStepNames();
232                final Collection<Step> result = new ArrayList<Step>();
233                for (String stepName : stepNames) {
234                        result.add(stepLocator.getStep(stepName));
235                }
236
237                // Because some steps are referenced by name, we need to look in the context to see if there
238                // are more Step instances defined. Right now they are registered as being available in the
239                // context of the job but we have no idea if they are linked to that Job or not.
240                final Map<String, Step> allSteps = jobApplicationContext.getBeansOfType(Step.class);
241                for (Map.Entry<String, Step> entry : allSteps.entrySet()) {
242                        if (!stepNames.contains(entry.getKey())) {
243                                result.add(entry.getValue());
244                        }
245                }
246                return result;
247        }
248
249        /**
250         * Registers the specified {@link Job} defined in the specified {@link ConfigurableApplicationContext}.
251         * <br>
252         * Makes sure to update the {@link StepRegistry} if it is available.
253         *
254         * @param context the context in which the job is defined
255         * @param job the job to register
256         * @throws DuplicateJobException if that job is already registered
257         */
258        private void doRegister(ConfigurableApplicationContext context, Job job) throws DuplicateJobException {
259                final JobFactory jobFactory = new ReferenceJobFactory(job);
260                jobRegistry.register(jobFactory);
261
262                if (stepRegistry != null) {
263                        if (!(job instanceof StepLocator)) {
264                                throw new UnsupportedOperationException("Cannot locate steps from a Job that is not a StepLocator: job="
265                                                + job.getName() + " does not implement StepLocator");
266                        }
267                        stepRegistry.register(job.getName(), getSteps((StepLocator) job, context));
268                }
269        }
270
271        /**
272         * Unregisters the job identified by the specified <tt>jobName</tt>.
273         *
274         * @param jobName the name of the job to unregister
275         */
276        private void doUnregister(String jobName)  {
277                jobRegistry.unregister(jobName);
278                if (stepRegistry != null) {
279                        stepRegistry.unregisterStepsFromJob(jobName);
280                }
281
282        }
283
284        @Override
285        public void afterPropertiesSet() {
286                Assert.notNull(jobRegistry, "Job registry could not be null.");
287        }
288}