/*
 * Copyright 2006-2013 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      https://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.springframework.batch.core.step.factory;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.batch.core.ChunkListener;
import org.springframework.batch.core.ItemProcessListener;
import org.springframework.batch.core.ItemReadListener;
import org.springframework.batch.core.ItemWriteListener;
import org.springframework.batch.core.Step;
import org.springframework.batch.core.StepExecutionListener;
import org.springframework.batch.core.StepListener;
import org.springframework.batch.core.repository.JobRepository;
import org.springframework.batch.core.step.builder.SimpleStepBuilder;
import org.springframework.batch.core.step.builder.StepBuilder;
import org.springframework.batch.core.step.tasklet.TaskletStep;
import org.springframework.batch.item.ItemProcessor;
import org.springframework.batch.item.ItemReader;
import org.springframework.batch.item.ItemStream;
import org.springframework.batch.item.ItemWriter;
import org.springframework.batch.repeat.CompletionPolicy;
import org.springframework.batch.repeat.RepeatOperations;
import org.springframework.batch.repeat.exception.DefaultExceptionHandler;
import org.springframework.batch.repeat.exception.ExceptionHandler;
import org.springframework.batch.repeat.policy.SimpleCompletionPolicy;
import org.springframework.batch.repeat.support.TaskExecutorRepeatTemplate;
import org.springframework.beans.factory.BeanNameAware;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.core.task.TaskExecutor;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.annotation.Isolation;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.interceptor.DefaultTransactionAttribute;
import org.springframework.transaction.interceptor.TransactionAttribute;

/**
 * Most common configuration options for simple steps should be found here. Use this factory bean instead of creating a
 * {@link Step} implementation manually.
 *
 * This factory does not support configuration of fault-tolerant behavior, use appropriate subclass of this factory bean
 * to configure skip or retry.
 *
 * @see FaultTolerantStepFactoryBean
 *
 * @author Dave Syer
 * @author Robert Kasanicky
 *
 */
public class SimpleStepFactoryBean<T, S> implements FactoryBean<Step>, BeanNameAware {

        private String name;

        private int startLimit = Integer.MAX_VALUE;

        private boolean allowStartIfComplete;

        private ItemReader<? extends T> itemReader;

        private ItemProcessor<? super T, ? extends S> itemProcessor;

        private ItemWriter<? super S> itemWriter;

        private PlatformTransactionManager transactionManager;

        private Propagation propagation = Propagation.REQUIRED;

        private Isolation isolation = Isolation.DEFAULT;

        private int transactionTimeout = DefaultTransactionAttribute.TIMEOUT_DEFAULT;

        private JobRepository jobRepository;

        private boolean singleton = true;

        private ItemStream[] streams = new ItemStream[0];

        private StepListener[] listeners = new StepListener[0];

        protected final Log logger = LogFactory.getLog(getClass());

        private int commitInterval = 0;

        private TaskExecutor taskExecutor;

        private RepeatOperations stepOperations;

        private RepeatOperations chunkOperations;

        private ExceptionHandler exceptionHandler = new DefaultExceptionHandler();

        private CompletionPolicy chunkCompletionPolicy;

        private int throttleLimit = TaskExecutorRepeatTemplate.DEFAULT_THROTTLE_LIMIT;

        private boolean isReaderTransactionalQueue = false;

        /**
         * Default constructor for {@link SimpleStepFactoryBean}.
         */
        public SimpleStepFactoryBean() {
                super();
        }

        /**
         * Flag to signal that the reader is transactional (usually a JMS consumer) so that items are re-presented after a
         * rollback. The default is false and readers are assumed to be forward-only.
         *
         * @param isReaderTransactionalQueue the value of the flag
         */
        public void setIsReaderTransactionalQueue(boolean isReaderTransactionalQueue) {
                this.isReaderTransactionalQueue = isReaderTransactionalQueue;
        }

        /**
         * Convenience method for subclasses.
         * @return true if the flag is set (default false)
         */
        protected boolean isReaderTransactionalQueue() {
                return isReaderTransactionalQueue;
        }

        /**
         * Set the bean name property, which will become the name of the {@link Step} when it is created.
         *
         * @see org.springframework.beans.factory.BeanNameAware#setBeanName(java.lang.String)
         */
        @Override
        public void setBeanName(String name) {
                this.name = name;
        }

        /**
         * Public getter for the name of the step.
         * @return the name
         */
        public String getName() {
                return name;
        }

        /**
         * The timeout for an individual transaction in the step.
         *
         * @param transactionTimeout the transaction timeout to set, defaults to infinite
         */
        public void setTransactionTimeout(int transactionTimeout) {
                this.transactionTimeout = transactionTimeout;
        }

        /**
         * @param propagation the propagation to set for business transactions
         */
        public void setPropagation(Propagation propagation) {
                this.propagation = propagation;
        }

        /**
         * @param isolation the isolation to set for business transactions
         */
        public void setIsolation(Isolation isolation) {
                this.isolation = isolation;
        }

        /**
         * Public setter for the start limit for the step.
         *
         * @param startLimit the startLimit to set
         */
        public void setStartLimit(int startLimit) {
                this.startLimit = startLimit;
        }

        /**
         * Public setter for the flag to indicate that the step should be replayed on a restart, even if successful the
         * first time.
         *
         * @param allowStartIfComplete the shouldAllowStartIfComplete to set
         */
        public void setAllowStartIfComplete(boolean allowStartIfComplete) {
                this.allowStartIfComplete = allowStartIfComplete;
        }

        /**
         * @param itemReader the {@link ItemReader} to set
         */
        public void setItemReader(ItemReader<? extends T> itemReader) {
                this.itemReader = itemReader;
        }

        /**
         * @param itemWriter the {@link ItemWriter} to set
         */
        public void setItemWriter(ItemWriter<? super S> itemWriter) {
                this.itemWriter = itemWriter;
        }

        /**
         * @param itemProcessor the {@link ItemProcessor} to set
         */
        public void setItemProcessor(ItemProcessor<? super T, ? extends S> itemProcessor) {
                this.itemProcessor = itemProcessor;
        }

        /**
         * The streams to inject into the {@link Step}. Any instance of {@link ItemStream} can be used, and will then
         * receive callbacks at the appropriate stage in the step.
         *
         * @param streams an array of listeners
         */
        public void setStreams(ItemStream[] streams) {
                this.streams = streams;
        }

        /**
         * The listeners to inject into the {@link Step}. Any instance of {@link StepListener} can be used, and will then
         * receive callbacks at the appropriate stage in the step.
         *
         * @param listeners an array of listeners
         */
        public void setListeners(StepListener[] listeners) {
                this.listeners = listeners;
        }

        /**
         * Protected getter for the {@link StepListener}s.
         * @return the listeners
         */
        protected StepListener[] getListeners() {
                return listeners;
        }

        /**
         * Protected getter for the {@link ItemReader} for subclasses to use.
         * @return the itemReader
         */
        protected ItemReader<? extends T> getItemReader() {
                return itemReader;
        }

        /**
         * Protected getter for the {@link ItemWriter} for subclasses to use
         * @return the itemWriter
         */
        protected ItemWriter<? super S> getItemWriter() {
                return itemWriter;
        }

        /**
         * Protected getter for the {@link ItemProcessor} for subclasses to use
         * @return the itemProcessor
         */
        protected ItemProcessor<? super T, ? extends S> getItemProcessor() {
                return itemProcessor;
        }

        /**
         * Public setter for {@link JobRepository}.
         *
         * @param jobRepository is a mandatory dependence (no default).
         */
        public void setJobRepository(JobRepository jobRepository) {
                this.jobRepository = jobRepository;
        }

        /**
         * Public setter for the {@link PlatformTransactionManager}.
         *
         * @param transactionManager the transaction manager to set
         */
        public void setTransactionManager(PlatformTransactionManager transactionManager) {
                this.transactionManager = transactionManager;
        }

        /**
         * Getter for the {@link TransactionAttribute} for subclasses only.
         * @return the transactionAttribute
         */
        @SuppressWarnings("serial")
        protected TransactionAttribute getTransactionAttribute() {

                DefaultTransactionAttribute attribute = new DefaultTransactionAttribute();
                attribute.setPropagationBehavior(propagation.value());
                attribute.setIsolationLevel(isolation.value());
                attribute.setTimeout(transactionTimeout);
                return new DefaultTransactionAttribute(attribute) {

                        /**
                         * Ignore the default behaviour and rollback on all exceptions that bubble up to the tasklet level. The
                         * tasklet has to deal with the rollback rules internally.
                         */
                        @Override
                        public boolean rollbackOn(Throwable ex) {
                                return true;
                        }

                };

        }

        /**
         * Create a {@link Step} from the configuration provided.
         *
         * @see FactoryBean#getObject()
         */
        @Override
        public final Step getObject() throws Exception {
                SimpleStepBuilder<T, S> builder = createBuilder(getName());
                applyConfiguration(builder);
                TaskletStep step = builder.build();
                return step;
        }

        protected SimpleStepBuilder<T, S> createBuilder(String name) {
                return new SimpleStepBuilder<>(new StepBuilder(name));
        }

        @Override
        public Class<TaskletStep> getObjectType() {
                return TaskletStep.class;
        }

        /**
         * Returns true by default, but in most cases a {@link Step} should <b>not</b> be treated as thread-safe. Clients are
         * recommended to create a new step for each job execution.
         *
         * @see org.springframework.beans.factory.FactoryBean#isSingleton()
         */
        @Override
        public boolean isSingleton() {
                return this.singleton;
        }

        /**
         * Public setter for the singleton flag.
         * @param singleton the value to set. Defaults to true.
         */
        public void setSingleton(boolean singleton) {
                this.singleton = singleton;
        }

        /**
         * Set the commit interval. Either set this or the chunkCompletionPolicy but not both.
         *
         * @param commitInterval 1 by default
         */
        public void setCommitInterval(int commitInterval) {
                this.commitInterval = commitInterval;
        }

        /**
         * Public setter for the {@link CompletionPolicy} applying to the chunk level. A transaction will be committed when
         * this policy decides to complete. Defaults to a {@link SimpleCompletionPolicy} with chunk size equal to the
         * commitInterval property.
         *
         * @param chunkCompletionPolicy the chunkCompletionPolicy to set
         */
        public void setChunkCompletionPolicy(CompletionPolicy chunkCompletionPolicy) {
                this.chunkCompletionPolicy = chunkCompletionPolicy;
        }

        /**
         * Protected getter for the step operations to make them available in subclasses.
         * @return the step operations
         */
        protected RepeatOperations getStepOperations() {
                return stepOperations;
        }

        /**
         * Public setter for the stepOperations.
         * @param stepOperations the stepOperations to set
         */
        public void setStepOperations(RepeatOperations stepOperations) {
                this.stepOperations = stepOperations;
        }

        /**
         * Public setter for the chunkOperations.
         * @param chunkOperations the chunkOperations to set
         */
        public void setChunkOperations(RepeatOperations chunkOperations) {
                this.chunkOperations = chunkOperations;
        }

        /**
         * Protected getter for the chunk operations to make them available in subclasses.
         * @return the step operations
         */
        protected RepeatOperations getChunkOperations() {
                return chunkOperations;
        }

        /**
         * Public setter for the {@link ExceptionHandler}.
         * @param exceptionHandler the exceptionHandler to set
         */
        public void setExceptionHandler(ExceptionHandler exceptionHandler) {
                this.exceptionHandler = exceptionHandler;
        }

        /**
         * Protected getter for the {@link ExceptionHandler}.
         * @return the {@link ExceptionHandler}
         */
        protected ExceptionHandler getExceptionHandler() {
                return exceptionHandler;
        }

        /**
         * Public setter for the {@link TaskExecutor}. If this is set, then it will be used to execute the chunk processing
         * inside the {@link Step}.
         *
         * @param taskExecutor the taskExecutor to set
         */
        public void setTaskExecutor(TaskExecutor taskExecutor) {
                this.taskExecutor = taskExecutor;
        }

        /**
         * Make the {@link TaskExecutor} available to subclasses
         * @return the taskExecutor to be used to execute chunks
         */
        protected TaskExecutor getTaskExecutor() {
                return taskExecutor;
        }

        /**
         * Public setter for the throttle limit. This limits the number of tasks queued for concurrent processing to prevent
         * thread pools from being overwhelmed. Defaults to {@link TaskExecutorRepeatTemplate#DEFAULT_THROTTLE_LIMIT}.
         * @param throttleLimit the throttle limit to set.
         */
        public void setThrottleLimit(int throttleLimit) {
                this.throttleLimit = throttleLimit;
        }

        protected void applyConfiguration(SimpleStepBuilder<T, S> builder) {

                builder.reader(itemReader);
                builder.processor(itemProcessor);
                builder.writer(itemWriter);
                for (StepExecutionListener listener : BatchListenerFactoryHelper.<StepExecutionListener> getListeners(
                                listeners, StepExecutionListener.class)) {
                        builder.listener(listener);
                }
                for (ChunkListener listener : BatchListenerFactoryHelper.<ChunkListener> getListeners(listeners,
                                ChunkListener.class)) {
                        builder.listener(listener);
                }
                for (ItemReadListener<T> listener : BatchListenerFactoryHelper.<ItemReadListener<T>> getListeners(listeners,
                                ItemReadListener.class)) {
                        builder.listener(listener);
                }
                for (ItemWriteListener<S> listener : BatchListenerFactoryHelper.<ItemWriteListener<S>> getListeners(listeners,
                                ItemWriteListener.class)) {
                        builder.listener(listener);
                }
                for (ItemProcessListener<T, S> listener : BatchListenerFactoryHelper.<ItemProcessListener<T, S>> getListeners(
                                listeners, ItemProcessListener.class)) {
                        builder.listener(listener);
                }
                builder.transactionManager(transactionManager);
                builder.transactionAttribute(getTransactionAttribute());
                builder.repository(jobRepository);
                builder.startLimit(startLimit);
                builder.allowStartIfComplete(allowStartIfComplete);
                builder.chunk(commitInterval);
                builder.chunk(chunkCompletionPolicy);
                builder.chunkOperations(chunkOperations);
                builder.stepOperations(stepOperations);
                builder.taskExecutor(taskExecutor);
                builder.throttleLimit(throttleLimit);
                builder.exceptionHandler(exceptionHandler);
                if (isReaderTransactionalQueue) {
                        builder.readerIsTransactionalQueue();
                }
                for (ItemStream stream : streams) {
                        builder.stream(stream);
                }

        }
}