Merge pull request #172 from redis/FIELDENG-558-cleanup-and-refactor-snowflake

refactor to use common reader

Author: jruaux

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;
@@ -34,30 +27,8 @@ protected Job job() {
 	}
 
 	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);
+		return JdbcCursorItemReaderFactory.create(sql, dataSourceArgs, readerArgs).build();
 	}
 
 	public String getSql() {

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

@@ -0,0 +1,42 @@
+package com.redis.riot;
+
+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 com.redis.riot.core.RiotException;
+
+public class JdbcCursorItemReaderFactory {
+
+	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;
+	}
+
+}