JAVA-8362: Remove batchscheduler code from spring-batch module
This commit is contained in:
Krzysiek
parent
4dc3a8b83a
commit
dcb9072f2a
|
|
@ -1,184 +0,0 @@
|
|||
package com.baeldung.batchscheduler;
|
||||
|
||||
import com.baeldung.batchscheduler.model.Book;
|
||||
import org.slf4j.Logger;
|
||||
import org.slf4j.LoggerFactory;
|
||||
import org.springframework.batch.core.Job;
|
||||
import org.springframework.batch.core.JobExecution;
|
||||
import org.springframework.batch.core.JobParametersBuilder;
|
||||
import org.springframework.batch.core.Step;
|
||||
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
|
||||
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
|
||||
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
|
||||
import org.springframework.batch.core.launch.JobLauncher;
|
||||
import org.springframework.batch.core.launch.support.SimpleJobLauncher;
|
||||
import org.springframework.batch.core.repository.JobRepository;
|
||||
import org.springframework.batch.core.repository.support.JobRepositoryFactoryBean;
|
||||
import org.springframework.batch.item.ItemWriter;
|
||||
import org.springframework.batch.item.file.FlatFileItemReader;
|
||||
import org.springframework.batch.item.file.builder.FlatFileItemReaderBuilder;
|
||||
import org.springframework.batch.item.file.mapping.BeanWrapperFieldSetMapper;
|
||||
import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.core.io.ClassPathResource;
|
||||
import org.springframework.jdbc.datasource.DriverManagerDataSource;
|
||||
import org.springframework.scheduling.TaskScheduler;
|
||||
import org.springframework.scheduling.annotation.EnableScheduling;
|
||||
import org.springframework.scheduling.annotation.Scheduled;
|
||||
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
|
||||
import org.springframework.scheduling.support.ScheduledMethodRunnable;
|
||||
|
||||
import javax.sql.DataSource;
|
||||
import java.util.Date;
|
||||
import java.util.IdentityHashMap;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.ScheduledFuture;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
@Configuration
|
||||
@EnableBatchProcessing
|
||||
@EnableScheduling
|
||||
public class SpringBatchScheduler {
|
||||
|
||||
private final Logger logger = LoggerFactory.getLogger(SpringBatchScheduler.class);
|
||||
|
||||
private AtomicBoolean enabled = new AtomicBoolean(true);
|
||||
|
||||
private AtomicInteger batchRunCounter = new AtomicInteger(0);
|
||||
|
||||
private final Map<Object, ScheduledFuture<?>> scheduledTasks = new IdentityHashMap<>();
|
||||
|
||||
@Autowired
|
||||
private JobBuilderFactory jobBuilderFactory;
|
||||
|
||||
@Autowired
|
||||
private StepBuilderFactory stepBuilderFactory;
|
||||
|
||||
@Scheduled(fixedRate = 2000)
|
||||
public void launchJob() throws Exception {
|
||||
Date date = new Date();
|
||||
logger.debug("scheduler starts at " + date);
|
||||
if (enabled.get()) {
|
||||
JobExecution jobExecution = jobLauncher().run(job(), new JobParametersBuilder().addDate("launchDate", date)
|
||||
.toJobParameters());
|
||||
batchRunCounter.incrementAndGet();
|
||||
logger.debug("Batch job ends with status as " + jobExecution.getStatus());
|
||||
}
|
||||
logger.debug("scheduler ends ");
|
||||
}
|
||||
|
||||
public void stop() {
|
||||
enabled.set(false);
|
||||
}
|
||||
|
||||
public void start() {
|
||||
enabled.set(true);
|
||||
}
|
||||
|
||||
@Bean
|
||||
public TaskScheduler poolScheduler() {
|
||||
return new CustomTaskScheduler();
|
||||
}
|
||||
|
||||
private class CustomTaskScheduler extends ThreadPoolTaskScheduler {
|
||||
|
||||
private static final long serialVersionUID = -7142624085505040603L;
|
||||
|
||||
@Override
|
||||
public ScheduledFuture<?> scheduleAtFixedRate(Runnable task, long period) {
|
||||
ScheduledFuture<?> future = super.scheduleAtFixedRate(task, period);
|
||||
|
||||
ScheduledMethodRunnable runnable = (ScheduledMethodRunnable) task;
|
||||
scheduledTasks.put(runnable.getTarget(), future);
|
||||
|
||||
return future;
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
public void cancelFutureSchedulerTasks() {
|
||||
scheduledTasks.forEach((k, v) -> {
|
||||
if (k instanceof SpringBatchScheduler) {
|
||||
v.cancel(false);
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
@Bean
|
||||
public Job job() {
|
||||
return jobBuilderFactory.get("job")
|
||||
.start(readBooks())
|
||||
.build();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public JobLauncher jobLauncher() throws Exception {
|
||||
SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
|
||||
jobLauncher.setJobRepository(jobRepository());
|
||||
jobLauncher.afterPropertiesSet();
|
||||
return jobLauncher;
|
||||
}
|
||||
|
||||
@Bean
|
||||
public JobRepository jobRepository() throws Exception {
|
||||
JobRepositoryFactoryBean factory = new JobRepositoryFactoryBean();
|
||||
factory.setDataSource(dataSource());
|
||||
factory.setTransactionManager(new ResourcelessTransactionManager());
|
||||
return factory.getObject();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public DataSource dataSource() {
|
||||
DriverManagerDataSource dataSource = new DriverManagerDataSource();
|
||||
dataSource.setDriverClassName("org.sqlite.JDBC");
|
||||
dataSource.setUrl("jdbc:sqlite:repository.sqlite");
|
||||
return dataSource;
|
||||
}
|
||||
|
||||
@Bean
|
||||
protected Step readBooks() {
|
||||
return stepBuilderFactory.get("readBooks")
|
||||
.<Book, Book> chunk(2)
|
||||
.reader(reader())
|
||||
.writer(writer())
|
||||
.build();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public FlatFileItemReader<Book> reader() {
|
||||
return new FlatFileItemReaderBuilder<Book>().name("bookItemReader")
|
||||
.resource(new ClassPathResource("books.csv"))
|
||||
.delimited()
|
||||
.names(new String[] { "id", "name" })
|
||||
.fieldSetMapper(new BeanWrapperFieldSetMapper<Book>() {
|
||||
{
|
||||
setTargetType(Book.class);
|
||||
}
|
||||
})
|
||||
.build();
|
||||
}
|
||||
|
||||
@Bean
|
||||
public ItemWriter<Book> writer() {
|
||||
return new ItemWriter<Book>() {
|
||||
|
||||
@Override
|
||||
public void write(List<? extends Book> items) throws Exception {
|
||||
logger.debug("writer..." + items.size());
|
||||
for (Book item : items) {
|
||||
logger.debug(item.toString());
|
||||
}
|
||||
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
public AtomicInteger getBatchRunCounter() {
|
||||
return batchRunCounter;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -1,35 +0,0 @@
|
|||
package com.baeldung.batchscheduler.model;
|
||||
|
||||
public class Book {
|
||||
private int id;
|
||||
private String name;
|
||||
|
||||
public Book() {}
|
||||
|
||||
public Book(int id, String name) {
|
||||
super();
|
||||
this.id = id;
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
public int getId() {
|
||||
return id;
|
||||
}
|
||||
|
||||
public void setId(int id) {
|
||||
this.id = id;
|
||||
}
|
||||
|
||||
public String getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
public void setName(String name) {
|
||||
this.name = name;
|
||||
}
|
||||
|
||||
public String toString() {
|
||||
return "Book [id=" + id + ", name=" + name + "]";
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -1,61 +0,0 @@
|
|||
package com.baeldung.batchscheduler;
|
||||
|
||||
import com.baeldung.batchscheduler.SpringBatchScheduler;
|
||||
import org.junit.Assert;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.context.ApplicationContext;
|
||||
import org.springframework.scheduling.annotation.ScheduledAnnotationBeanPostProcessor;
|
||||
import org.springframework.test.context.ContextConfiguration;
|
||||
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
|
||||
|
||||
import static org.awaitility.Awaitility.await;
|
||||
import static java.util.concurrent.TimeUnit.*;
|
||||
|
||||
@RunWith(SpringJUnit4ClassRunner.class)
|
||||
@ContextConfiguration(classes = SpringBatchScheduler.class)
|
||||
public class SpringBatchSchedulerIntegrationTest {
|
||||
|
||||
@Autowired
|
||||
private ApplicationContext context;
|
||||
|
||||
@Test
|
||||
public void stopJobsWhenSchedulerDisabled() throws Exception {
|
||||
SpringBatchScheduler schedulerBean = context.getBean(SpringBatchScheduler.class);
|
||||
await().untilAsserted(() -> Assert.assertEquals(2, schedulerBean.getBatchRunCounter()
|
||||
.get()));
|
||||
schedulerBean.stop();
|
||||
await().atLeast(3, SECONDS);
|
||||
|
||||
Assert.assertEquals(2, schedulerBean.getBatchRunCounter()
|
||||
.get());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void stopJobSchedulerWhenSchedulerDestroyed() throws Exception {
|
||||
ScheduledAnnotationBeanPostProcessor bean = context.getBean(ScheduledAnnotationBeanPostProcessor.class);
|
||||
SpringBatchScheduler schedulerBean = context.getBean(SpringBatchScheduler.class);
|
||||
await().untilAsserted(() -> Assert.assertEquals(2, schedulerBean.getBatchRunCounter()
|
||||
.get()));
|
||||
bean.postProcessBeforeDestruction(schedulerBean, "SpringBatchScheduler");
|
||||
await().atLeast(3, SECONDS);
|
||||
|
||||
Assert.assertEquals(2, schedulerBean.getBatchRunCounter()
|
||||
.get());
|
||||
}
|
||||
|
||||
@Test
|
||||
public void stopJobSchedulerWhenFutureTasksCancelled() throws Exception {
|
||||
SpringBatchScheduler schedulerBean = context.getBean(SpringBatchScheduler.class);
|
||||
await().untilAsserted(() -> Assert.assertEquals(2, schedulerBean.getBatchRunCounter()
|
||||
.get()));
|
||||
schedulerBean.cancelFutureSchedulerTasks();
|
||||
await().atLeast(3, SECONDS);
|
||||
|
||||
Assert.assertEquals(2, schedulerBean.getBatchRunCounter()
|
||||
.get());
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue