refactor to use common reader

Author: jeremyplichta

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

@@ -0,0 +1,38 @@
+package com.redis.riot;
+
+import com.redis.riot.core.RiotException;
+import org.springframework.batch.item.database.JdbcCursorItemReader;
+import org.springframework.batch.item.database.builder.JdbcCursorItemReaderBuilder;
+import org.springframework.jdbc.core.ColumnMapRowMapper;
+
+import javax.sql.DataSource;
+import java.util.Map;
+
+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();
+    }
+}