refactor: Use JDBC reader factory in db import command

Author: julien

plugins/riot/src/main/java/com/redis/riot/DatabaseImport.java

@@ -2,15 +2,8 @@
 
 import java.util.Map;
 
-import javax.sql.DataSource;
-
 import org.springframework.batch.core.Job;
-import org.springframework.batch.item.database.JdbcCursorItemReader;
 import org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder;
-import org.springframework.jdbc.core.ColumnMapRowMapper;
-import org.springframework.util.Assert;
-
-import com.redis.riot.core.RiotException;
 
 import picocli.CommandLine.ArgGroup;
 import picocli.CommandLine.Command;
@@ -30,34 +23,10 @@ public class DatabaseImport extends AbstractRedisImportCommand {
 
 	@Override
 	protected Job job() {
-		return job(step(reader()));
-	}
-
-	protected JdbcCursorItemReader<Map<String, Object>> reader() {
-		Assert.hasLength(sql, "No SQL statement specified");
-		log.info("Creating data source with {}", dataSourceArgs);
-		DataSource dataSource;
-		try {
-			dataSource = dataSourceArgs.dataSource();
-		} catch (Exception e) {
-			throw new RiotException(e);
-		}
-		log.info("Creating JDBC reader with sql=\"{}\" {}", sql, readerArgs);
-		JdbcCursorItemReaderBuilder<Map<String, Object>> reader = new JdbcCursorItemReaderBuilder<>();
-		reader.dataSource(dataSource);
-		reader.sql(sql);
-		reader.saveState(false);
-		reader.rowMapper(new ColumnMapRowMapper());
-		reader.fetchSize(readerArgs.getFetchSize());
-		reader.maxRows(readerArgs.getMaxRows());
-		reader.queryTimeout(Math.toIntExact(readerArgs.getQueryTimeout().getValue().toMillis()));
-		reader.useSharedExtendedConnection(readerArgs.isUseSharedExtendedConnection());
-		reader.verifyCursorPosition(readerArgs.isVerifyCursorPosition());
-		if (readerArgs.getMaxItemCount() > 0) {
-			reader.maxItemCount(readerArgs.getMaxItemCount());
-		}
-		reader.name(sql);
-		return reader.build();
+		log.info("Creating JDBC reader with sql=\"{}\" {} {}", sql, dataSourceArgs, readerArgs);
+		JdbcCursorItemReaderBuilder<Map<String, Object>> reader = JdbcCursorItemReaderFactory.create(sql,
+				dataSourceArgs, readerArgs);
+		return job(step(reader.build()));
 	}
 
 	public String getSql() {

plugins/riot/src/main/java/com/redis/riot/JdbcCursorItemReaderFactory.java

@@ -1,38 +1,42 @@
 package com.redis.riot;
 
-import com.redis.riot.core.RiotException;
-import org.springframework.batch.item.database.JdbcCursorItemReader;
+import java.util.Map;
+
+import javax.sql.DataSource;
+
 import org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder;
 import org.springframework.jdbc.core.ColumnMapRowMapper;
+import org.springframework.util.Assert;
 
-import javax.sql.DataSource;
-import java.util.Map;
+import com.redis.riot.core.RiotException;
 
 public class JdbcCursorItemReaderFactory {
-    public static JdbcCursorItemReader<Map<String, Object>> createReader(String sql,
-                                                                         DataSourceArgs dataSourceArgs,
-                                                                         DatabaseReaderArgs readerArgs) {
-        DataSource dataSource;
-        try {
-            dataSource = dataSourceArgs.dataSource();
-        } catch (Exception e) {
-            throw new RiotException(e);
-        }
-
-        JdbcCursorItemReaderBuilder<Map<String, Object>> reader = new JdbcCursorItemReaderBuilder<>();
-        reader.dataSource(dataSource);
-        reader.sql(sql);
-        reader.saveState(false);
-        reader.rowMapper(new ColumnMapRowMapper());
-        reader.fetchSize(readerArgs.getFetchSize());
-        reader.maxRows(readerArgs.getMaxRows());
-        reader.queryTimeout(Math.toIntExact(readerArgs.getQueryTimeout().getValue().toMillis()));
-        reader.useSharedExtendedConnection(readerArgs.isUseSharedExtendedConnection());
-        reader.verifyCursorPosition(readerArgs.isVerifyCursorPosition());
-        if (readerArgs.getMaxItemCount() > 0) {
-            reader.maxItemCount(readerArgs.getMaxItemCount());
-        }
-        reader.name(sql);
-        return reader.build();
-    }
+
+	public static JdbcCursorItemReaderBuilder<Map<String, Object>> create(String sql, DataSourceArgs dataSourceArgs,
+			DatabaseReaderArgs readerArgs) {
+		Assert.hasLength(sql, "No SQL statement specified");
+		DataSource dataSource;
+		try {
+			dataSource = dataSourceArgs.dataSource();
+		} catch (Exception e) {
+			throw new RiotException(e);
+		}
+
+		JdbcCursorItemReaderBuilder<Map<String, Object>> reader = new JdbcCursorItemReaderBuilder<>();
+		reader.dataSource(dataSource);
+		reader.sql(sql);
+		reader.saveState(false);
+		reader.rowMapper(new ColumnMapRowMapper());
+		reader.fetchSize(readerArgs.getFetchSize());
+		reader.maxRows(readerArgs.getMaxRows());
+		reader.queryTimeout(Math.toIntExact(readerArgs.getQueryTimeout().getValue().toMillis()));
+		reader.useSharedExtendedConnection(readerArgs.isUseSharedExtendedConnection());
+		reader.verifyCursorPosition(readerArgs.isVerifyCursorPosition());
+		if (readerArgs.getMaxItemCount() > 0) {
+			reader.maxItemCount(readerArgs.getMaxItemCount());
+		}
+		reader.name(sql);
+		return reader;
+	}
+
 }