spring batch

kanban-server/build.gradle

@@ -62,6 +62,8 @@ dependencies {
     compile 'org.springframework.security:spring-security-config:5.0.7.RELEASE'
     compile 'org.springframework:spring-context-support:5.0.7.RELEASE'
     compile 'net.sf.ehcache:ehcache-core:2.6.11'
+    compile("org.springframework.boot:spring-boot-starter-batch")
+    implementation('org.springframework.boot:spring-boot-starter-data-mongodb:2.0.3.RELEASE')
 
 //    compile 'org.liquibase:liquibase-core:3.6.2'
 

kanban-server/src/main/java/ru/otus/spring/hw/kanban/KanbanApplication.java

@@ -1,10 +1,21 @@
 package ru.otus.spring.hw.kanban;
 
+import org.springframework.batch.core.Job;
+import org.springframework.batch.core.JobParameters;
+import org.springframework.batch.core.JobParametersInvalidException;
+import org.springframework.batch.core.launch.JobLauncher;
+import org.springframework.batch.core.repository.JobExecutionAlreadyRunningException;
+import org.springframework.batch.core.repository.JobInstanceAlreadyCompleteException;
+import org.springframework.batch.core.repository.JobRestartException;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
 import org.springframework.data.jpa.repository.config.EnableJpaRepositories;
+import org.springframework.data.mongodb.repository.config.EnableMongoRepositories;
 import org.springframework.security.crypto.password.PasswordEncoder;
+import ru.otus.spring.hw.kanban.conversion.ToMongoBatchConfig;
 import ru.otus.spring.hw.kanban.security.UserAccount;
 import ru.otus.spring.hw.kanban.security.UserAccountRepository;
 
@@ -13,6 +24,10 @@
 
 @SpringBootApplication(scanBasePackages = "ru.otus.spring.hw.kanban")
 @EnableJpaRepositories(basePackages = {"ru.otus.spring.hw.kanban.repository", "ru.otus.spring.hw.kanban.security"})
+@EnableMongoRepositories(basePackages = {"ru.otus.spring.hw.kanban.repository.mongo"})
+
+@Configuration
+@Import(ToMongoBatchConfig.class)
 public class KanbanApplication {
 
   public static void main(String[] args) {
@@ -26,6 +41,17 @@ public static void main(String[] args) {
   @Autowired
   PasswordEncoder encoder;
 
+
+  @Autowired
+  private JobLauncher jobLauncher;
+
+  @Autowired
+  private Job convertBoardsJob;
+
+  public void start() throws JobParametersInvalidException, JobExecutionAlreadyRunningException, JobRestartException, JobInstanceAlreadyCompleteException {
+    jobLauncher.run(convertBoardsJob, new JobParameters());
+  }
+
   @PostConstruct
   public void initAdmin() {
     if (repository.findByUsername("admin") == null) {
@@ -35,5 +61,16 @@ public void initAdmin() {
                           .build();
       repository.save(acc);
     }
+    try {
+      start();
+    } catch (JobParametersInvalidException e) {
+      e.printStackTrace();
+    } catch (JobExecutionAlreadyRunningException e) {
+      e.printStackTrace();
+    } catch (JobRestartException e) {
+      e.printStackTrace();
+    } catch (JobInstanceAlreadyCompleteException e) {
+      e.printStackTrace();
+    }
   }
 }

kanban-server/src/main/java/ru/otus/spring/hw/kanban/conversion/BoardItemProcessor.java

@@ -0,0 +1,51 @@
+package ru.otus.spring.hw.kanban.conversion;
+
+import org.springframework.batch.item.ItemProcessor;
+import org.springframework.stereotype.Service;
+import ru.otus.spring.hw.kanban.domain.Board;
+import ru.otus.spring.hw.kanban.domain.Stage;
+import ru.otus.spring.hw.kanban.domain.Task;
+
+import java.util.HashSet;
+import java.util.Set;
+import java.util.function.Function;
+import java.util.stream.Collectors;
+
+@Service
+public class BoardItemProcessor implements ItemProcessor<Board, ru.otus.spring.hw.kanban.domain.mongo.Board> {
+
+  @Override
+  public ru.otus.spring.hw.kanban.domain.mongo.Board process(Board board) throws Exception {
+    ru.otus.spring.hw.kanban.domain.mongo.Board mongoBoard = new ru.otus.spring.hw.kanban.domain.mongo.Board();
+    mongoBoard.setName(board.getName());
+
+    Set<ru.otus.spring.hw.kanban.domain.mongo.Stage> mongoStages = new HashSet<>();
+
+    Function<Task, ru.otus.spring.hw.kanban.domain.mongo.Task> convert = (task) -> {
+      ru.otus.spring.hw.kanban.domain.mongo.Task mongoTask = new ru.otus.spring.hw.kanban.domain.mongo.Task();
+      mongoTask.setDescription(task.getDescription());
+      mongoTask.setExecutor(task.getExecutor());
+      mongoTask.setName(task.getName());
+      mongoTask.setPriority(task.getPriority());
+      return mongoTask;
+    };
+
+    for (Stage stage : board.getStages()) {
+      ru.otus.spring.hw.kanban.domain.mongo.Stage mongoStage = new ru.otus.spring.hw.kanban.domain.mongo.Stage();
+      mongoStage.setDescription(stage.getDescription());
+      mongoStage.setName(stage.getName());
+      mongoStage.setTasks(
+        stage.getTasks()
+          .stream()
+          .map(task -> convert.apply(task))
+          .collect(Collectors.toSet())
+      );
+      mongoStages.add(mongoStage);
+
+    }
+    mongoBoard.setStages(mongoStages);
+    return mongoBoard;
+
+
+  }
+}

kanban-server/src/main/java/ru/otus/spring/hw/kanban/conversion/NoPersistenceBatchConfigurer.java

@@ -0,0 +1,16 @@
+package ru.otus.spring.hw.kanban.conversion;
+
+//public class NoPersistenceBatchConfigurer {
+//}
+
+import org.springframework.batch.core.configuration.annotation.DefaultBatchConfigurer;
+import org.springframework.stereotype.Component;
+
+import javax.sql.DataSource;
+
+@Component
+public class NoPersistenceBatchConfigurer extends DefaultBatchConfigurer {
+  @Override
+  public void setDataSource(DataSource dataSource) {
+  }
+}

kanban-server/src/main/java/ru/otus/spring/hw/kanban/conversion/ToMongoBatchConfig.java

@@ -0,0 +1,178 @@
+package ru.otus.spring.hw.kanban.conversion;
+
+import com.mongodb.MongoClient;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.batch.core.*;
+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.scope.context.ChunkContext;
+import org.springframework.batch.item.*;
+import org.springframework.batch.item.data.MongoItemWriter;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.context.annotation.Bean;
+import org.springframework.data.mongodb.MongoDbFactory;
+import org.springframework.data.mongodb.core.MongoTemplate;
+import org.springframework.data.mongodb.core.SimpleMongoDbFactory;
+import ru.otus.spring.hw.kanban.domain.Board;
+import ru.otus.spring.hw.kanban.domain.Stage;
+import ru.otus.spring.hw.kanban.repository.BoardRepository;
+import ru.otus.spring.hw.kanban.repository.StageRepository;
+import ru.otus.spring.hw.kanban.repository.TaskRepository;
+import ru.otus.spring.hw.kanban.repository.mongo.BoardMongoRepository;
+
+import java.util.*;
+import java.util.stream.Collectors;
+
+@EnableBatchProcessing
+public class ToMongoBatchConfig {
+
+  private final Logger logger = LoggerFactory.getLogger("Batch");
+
+  @Autowired
+  public JobBuilderFactory jobBuilderFactory;
+
+  @Autowired
+  public StepBuilderFactory stepBuilderFactory;
+
+
+  @Autowired
+  BoardItemProcessor boardItemProcessor;
+
+  @Autowired
+  BoardRepository boardRepository;
+
+
+  @Autowired
+  BoardMongoRepository boardMongoRepository;
+
+  @Autowired
+  StageRepository stageRepository;
+
+  @Autowired
+  TaskRepository taskRepository;
+
+  @Bean
+  public ItemReader<Board> reader() {
+    return new ItemReader<Board>() {
+
+      private Integer currentId = 0;
+
+      @Override
+      public Board read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
+        Board board = boardRepository.findFirstByIdGreaterThan(currentId).orElse(null);
+        if (board != null) {
+          Set<Stage> stagesSet = new HashSet<>();
+          List<Stage> stages = stageRepository.findStagesByBoard(board);
+
+          for (Stage stage : stages) {
+            stage.setTasks(taskRepository.findAllByStage(stage)
+              .stream()
+              .collect(Collectors.toSet()));
+            stagesSet.add(stage);
+          }
+          board.setStages(stagesSet);
+          currentId = board.getId();
+        }
+
+        return board;
+      }
+    };
+  }
+
+
+  @Bean
+  public ItemWriter<ru.otus.spring.hw.kanban.domain.mongo.Board> writer() {
+    MongoItemWriter<ru.otus.spring.hw.kanban.domain.mongo.Board> writer = new MongoItemWriter<>();
+    try {
+      writer.setTemplate(mongoTemplate());
+    } catch (Exception e) {
+      logger.error(e.toString());
+    }
+    writer.setCollection("board");
+    return writer;
+  }
+
+
+  @Bean
+  public MongoDbFactory mongoDbFactory() throws Exception {
+    return new SimpleMongoDbFactory(new MongoClient(), "kanban");
+  }
+
+  @Bean
+  public MongoTemplate mongoTemplate() throws Exception {
+    MongoTemplate mongoTemplate = new MongoTemplate(mongoDbFactory());
+    return mongoTemplate;
+  }
+
+
+  @Bean
+  public Step step1() {
+    return stepBuilderFactory.get("step1")
+      .<Board, ru.otus.spring.hw.kanban.domain.mongo.Board>chunk(5)
+      .reader(reader())
+      .processor(boardItemProcessor)
+      .writer(writer())
+      .listener(new ItemReadListener<Board>() {
+        public void beforeRead() {
+          logger.info("Начало чтения");
+        }
+
+        public void afterRead(Board o) {
+          logger.info("Конец чтения " + o.toString());
+        }
+
+        public void onReadError(Exception e) {
+          logger.info("Ошибка чтения");
+        }
+      })
+      .listener(new ItemWriteListener() {
+        public void beforeWrite(List list) {
+          logger.info("Начало записи");
+        }
+
+        public void afterWrite(List list) {
+          logger.info("Конец записи");
+        }
+
+        public void onWriteError(Exception e, List list) {
+          logger.info("Ошибка записи");
+        }
+      })
+      .listener(new ItemProcessListener() {
+        public void beforeProcess(Object o) {
+          logger.info("Начало обработки");
+        }
+
+        public void afterProcess(Object o, Object o2) {
+          logger.info("Конец обработки");
+        }
+
+        public void onProcessError(Object o, Exception e) {
+          logger.info("Ошбка обработки");
+        }
+      })
+      .listener(new ChunkListener() {
+        public void beforeChunk(ChunkContext chunkContext) {
+          logger.info("Начало пачки");
+        }
+
+        public void afterChunk(ChunkContext chunkContext) {
+          logger.info("Конец пачки");
+        }
+
+        public void afterChunkError(ChunkContext chunkContext) {
+          logger.info("Ошибка пачки");
+        }
+      })
+      .build();
+  }
+
+  @Bean
+  public Job convertBoardsJob() {
+    return jobBuilderFactory.get("convertBoardsJob")
+      .start(step1())
+      .build();
+  }
+}

kanban-server/src/main/java/ru/otus/spring/hw/kanban/domain/Board.java

@@ -1,10 +1,22 @@
 package ru.otus.spring.hw.kanban.domain;
 
 import javax.persistence.*;
+import java.util.HashSet;
 import java.util.Set;
 
 @Entity
 public class Board {
+    public Set<Stage> getStages() {
+        if (stages == null) {
+            stages = new HashSet<>();
+        }
+        return stages;
+    }
+
+    public void setStages(Set<Stage> stages) {
+        this.stages = stages;
+    }
+
     @OneToMany(cascade = {CascadeType.PERSIST, CascadeType.MERGE}, fetch = FetchType.LAZY)
     private Set<Stage> stages;
     @Column(length = 500)

kanban-server/src/main/java/ru/otus/spring/hw/kanban/domain/Stage.java

@@ -1,6 +1,8 @@
 package ru.otus.spring.hw.kanban.domain;
 
 import javax.persistence.*;
+import java.util.HashSet;
+import java.util.Set;
 
 @Entity
 public class Stage {
@@ -16,6 +18,21 @@ public class Stage {
     @ManyToOne(fetch = FetchType.LAZY)
     private Board board;
 
+    public Set<Task> getTasks() {
+        if (tasks == null) {
+            tasks = new HashSet<>();
+        }
+        return tasks;
+    }
+
+    public void setTasks(Set<Task> tasks) {
+        this.tasks = tasks;
+    }
+
+    @OneToMany(fetch = FetchType.LAZY)
+    private Set<Task> tasks;
+
+
     public Stage(String name, String description, Board board) {
         this.name = name;
         this.description = description;