added export

Author: jruaux

pom.xml

@@ -103,6 +103,10 @@
 			<artifactId>javafaker</artifactId>
 			<version>0.18</version>
 		</dependency>
+		<dependency>
+			<groupId>mysql</groupId>
+			<artifactId>mysql-connector-java</artifactId>
+		</dependency>
 	</dependencies>
 
 	<build>

src/main/java/com/redislabs/riot/BaseCommand.java

@@ -0,0 +1,13 @@
+package com.redislabs.riot;
+
+import picocli.CommandLine;
+import picocli.CommandLine.Command;
+
+@Command(mixinStandardHelpOptions = true)
+public class BaseCommand implements Runnable {
+
+	@Override
+	public void run() { 
+		new CommandLine(this).usage(System.out);
+	}
+}

src/main/java/com/redislabs/riot/RiotApplication.java

@@ -1,48 +1,34 @@
 package com.redislabs.riot;
 
 import java.net.InetAddress;
-import java.text.NumberFormat;
 import java.util.List;
 import java.util.Locale;
 
-import org.springframework.batch.core.ExitStatus;
-import org.springframework.batch.core.Job;
-import org.springframework.batch.core.JobExecution;
-import org.springframework.batch.core.JobParameters;
-import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
-import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
-import org.springframework.batch.core.launch.support.SimpleJobLauncher;
-import org.springframework.batch.core.repository.JobRepository;
-import org.springframework.batch.core.repository.support.MapJobRepositoryFactoryBean;
-import org.springframework.batch.item.ItemProcessor;
-import org.springframework.batch.item.ItemStreamWriter;
-import org.springframework.batch.item.support.AbstractItemCountingItemStreamItemReader;
-import org.springframework.batch.support.transaction.ResourcelessTransactionManager;
 import org.springframework.boot.CommandLineRunner;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.core.task.SyncTaskExecutor;
-import org.springframework.transaction.PlatformTransactionManager;
 
-import com.redislabs.riot.batch.JobBuilder;
-import com.redislabs.riot.cli.BaseCommand;
-import com.redislabs.riot.cli.ImportCommand;
+import com.redislabs.riot.cli.Export;
+import com.redislabs.riot.cli.Import;
 import com.redislabs.riot.redis.RedisConnectionBuilder;
 
 import io.lettuce.core.RedisURI;
 import lombok.Getter;
 import picocli.CommandLine;
 import picocli.CommandLine.Command;
 import picocli.CommandLine.DefaultExceptionHandler;
-import picocli.CommandLine.HelpCommand;
 import picocli.CommandLine.Option;
 import picocli.CommandLine.RunLast;
 import redis.clients.jedis.Protocol;
 
 @SpringBootApplication
-@Command(name = "riot", subcommands = { HelpCommand.class, ImportCommand.class })
+@Command(name = "riot", subcommands = { Export.class, Import.class })
 public class RiotApplication extends BaseCommand implements CommandLineRunner {
 
+	public static void main(String[] args) {
+		SpringApplication.run(RiotApplication.class, args);
+	}
+
 	public static final String DEFAULT_HOST = "localhost";
 
 	public enum RedisDriver {
@@ -54,34 +40,26 @@ public enum RedisDriver {
 	 */
 	@Option(names = "--spring.output.ansi.enabled", hidden = true)
 	private String ansiEnabled;
-	@Option(names = "--max", description = "Maximum number of items to read.", paramLabel = "<count>")
-	private Integer maxCount;
-	@Option(names = "--threads", description = "Number of partitions to use for processing. (default: ${DEFAULT-VALUE}).")
-	private int threads = 1;
-	@Option(names = "--chunk-size", description = "The chunk size commit interval. (default: ${DEFAULT-VALUE}).")
-	private int chunkSize = JobBuilder.DEFAULT_CHUNK_SIZE;
-	@Option(names = "--sleep", description = "Sleep duration in milliseconds between each read.")
-	private Long sleep;
+
+	@Option(names = "--driver", description = "Redis driver: ${COMPLETION-CANDIDATES}. (default: ${DEFAULT-VALUE})")
+	@Getter
+	private RedisDriver driver = RedisDriver.Jedis;
 	@Option(names = "--host", description = "Redis server host. (default: localhost).")
 	private InetAddress host;
-	@Getter
 	@Option(names = "--port", description = "Redis server port. (default: ${DEFAULT-VALUE}).")
 	private int port = RedisURI.DEFAULT_REDIS_PORT;
 	@Option(names = "--command-timeout", description = "Redis command timeout in seconds for synchronous command execution (default: ${DEFAULT-VALUE}).")
 	private long commandTimeout = RedisURI.DEFAULT_TIMEOUT;
-	@Getter
 	@Option(names = "--connection-timeout", description = "Redis connect timeout in milliseconds. (default: ${DEFAULT-VALUE}).")
 	private int connectionTimeout = Protocol.DEFAULT_TIMEOUT;
 	@Option(names = "--socket-timeout", description = "Redis socket timeout in milliseconds. (default: ${DEFAULT-VALUE}).")
 	private int socketTimeout = Protocol.DEFAULT_TIMEOUT;
-	@Getter
 	@Option(names = "--password", description = "Redis database password.", interactive = true)
-	protected String password;
+	private String password;
 	@Option(names = "--max-idle", description = "Maximum number of idle connections in the pool. Use a negative value to indicate an unlimited number of idle connections. (default: ${DEFAULT-VALUE}).")
 	private int maxIdle = 8;
 	@Option(names = "--min-idle", description = "Target for the minimum number of idle connections to maintain in the pool. This setting only has an effect if it is positive. (default: ${DEFAULT-VALUE}).")
 	private int minIdle = 0;
-	@Getter
 	@Option(names = "--max-total", description = "Maximum number of connections that can be allocated by the pool at a given time. Use a negative value for no limit. (default: ${DEFAULT-VALUE})")
 	private int maxTotal = 8;
 	@Option(names = "--max-wait", description = "Maximum amount of time in milliseconds a connection allocation should block before throwing an exception when the pool is exhausted. Use a negative value to block indefinitely (default).")
@@ -90,25 +68,6 @@ public enum RedisDriver {
 	private int database = 0;
 	@Option(names = "--client-name", description = "Redis client name.")
 	private String clientName;
-	@Getter
-	@Option(names = "--driver", description = "Redis driver: ${COMPLETION-CANDIDATES}. (default: ${DEFAULT-VALUE})")
-	private RedisDriver driver = RedisDriver.Jedis;
-
-	public static void main(String[] args) {
-		SpringApplication.run(RiotApplication.class, args);
-	}
-
-	@Override
-	public void run(String... args) {
-		CommandLine commandLine = new CommandLine(this);
-		commandLine.registerConverter(Locale.class, s -> new Locale.Builder().setLanguageTag(s).build());
-		commandLine.setCaseInsensitiveEnumValuesAllowed(true);
-		RunLast handler = new RunLast();
-		handler.useOut(System.out);
-		DefaultExceptionHandler<List<Object>> exceptionHandler = CommandLine.defaultExceptionHandler();
-		exceptionHandler.useErr(System.err);
-		commandLine.parseWithHandlers(handler, exceptionHandler, args);
-	}
 
 	public RedisConnectionBuilder redisConnectionBuilder() {
 		RedisConnectionBuilder builder = new RedisConnectionBuilder();
@@ -127,52 +86,23 @@ public RedisConnectionBuilder redisConnectionBuilder() {
 		return builder;
 	}
 
-	public String getHostname() {
+	private String getHostname() {
 		if (host != null) {
 			return host.getHostName();
 		}
 		return DEFAULT_HOST;
 	}
 
-	public <I, O> ExitStatus call(AbstractItemCountingItemStreamItemReader<I> reader, ItemProcessor<I, O> processor,
-			ItemStreamWriter<O> writer) throws Exception {
-		PlatformTransactionManager transactionManager = new ResourcelessTransactionManager();
-		MapJobRepositoryFactoryBean jobRepositoryFactory = new MapJobRepositoryFactoryBean(transactionManager);
-		jobRepositoryFactory.afterPropertiesSet();
-		JobRepository jobRepository = jobRepositoryFactory.getObject();
-		JobBuilderFactory jobBuilderFactory = new JobBuilderFactory(jobRepository);
-		StepBuilderFactory stepBuilderFactory = new StepBuilderFactory(jobRepository, transactionManager);
-		JobBuilder<I, O> builder = new JobBuilder<>(jobBuilderFactory, stepBuilderFactory);
-		builder.setChunkSize(chunkSize);
-		if (maxCount != null) {
-			reader.setMaxItemCount(maxCount);
-		}
-		builder.setReader(reader);
-		builder.setProcessor(processor);
-		builder.setWriter(writer);
-		builder.setPartitions(threads);
-		builder.setSleep(sleep);
-		Job job = builder.build();
-		SimpleJobLauncher jobLauncher = new SimpleJobLauncher();
-		jobLauncher.setJobRepository(jobRepository);
-		jobLauncher.setTaskExecutor(new SyncTaskExecutor());
-		jobLauncher.afterPropertiesSet();
-		long startTime = System.currentTimeMillis();
-		System.out.println("Running job");
-		JobExecution execution = jobLauncher.run(job, new JobParameters());
-		if (execution.getExitStatus().equals(ExitStatus.FAILED)) {
-			execution.getAllFailureExceptions().forEach(e -> e.printStackTrace());
-		} else if (execution.getExitStatus().equals(ExitStatus.COMPLETED)) {
-			long endTime = System.currentTimeMillis();
-			long duration = endTime - startTime;
-			NumberFormat numberFormat = NumberFormat.getIntegerInstance();
-			double durationInSeconds = (double) duration / 1000;
-			int writeCount = execution.getStepExecutions().iterator().next().getWriteCount();
-			double throughput = writeCount / durationInSeconds;
-			System.out.println("Processed " + numberFormat.format(writeCount) + " items in " + durationInSeconds
-					+ " seconds (" + numberFormat.format(throughput) + " writes/sec)");
-		}
-		return execution.getExitStatus();
+	@Override
+	public void run(String... args) {
+		CommandLine commandLine = new CommandLine(this);
+		commandLine.registerConverter(Locale.class, s -> new Locale.Builder().setLanguageTag(s).build());
+		commandLine.setCaseInsensitiveEnumValuesAllowed(true);
+		RunLast handler = new RunLast();
+		handler.useOut(System.out);
+		DefaultExceptionHandler<List<Object>> exceptionHandler = CommandLine.defaultExceptionHandler();
+		exceptionHandler.useErr(System.err);
+		commandLine.parseWithHandlers(handler, exceptionHandler, args);
 	}
 
 }

src/main/java/com/redislabs/riot/batch/JobBuilder.java

@@ -9,7 +9,7 @@
 import org.springframework.batch.item.ItemProcessor;
 import org.springframework.batch.item.ItemReader;
 import org.springframework.batch.item.ItemStreamReader;
-import org.springframework.batch.item.ItemStreamWriter;
+import org.springframework.batch.item.ItemWriter;
 import org.springframework.core.task.SimpleAsyncTaskExecutor;
 
 import lombok.Setter;
@@ -28,7 +28,7 @@ public class JobBuilder<I, O> {
 	@Setter
 	private ItemStreamReader<I> reader;
 	@Setter
-	private ItemStreamWriter<O> writer;
+	private ItemWriter<O> writer;
 	@Setter
 	private ItemProcessor<I, O> processor;
 	private JobBuilderFactory jobBuilderFactory;

src/main/java/com/redislabs/riot/cli/BaseCommand.java

@@ -0,0 +1,33 @@
+
+package com.redislabs.riot.cli;
+
+import com.redislabs.riot.RiotApplication.RedisDriver;
+import com.redislabs.riot.redis.writer.JedisCommands;
+import com.redislabs.riot.redis.writer.LettuceCommands;
+import com.redislabs.riot.redis.writer.RedisCommands;
+
+import lombok.Setter;
+
+public class CommandsBuilder {
+
+	@Setter
+	private RedisDriver driver;
+
+	protected RedisCommands build() {
+		switch (driver) {
+		case Lettuce:
+			return lettuceCommands();
+		default:
+			return jedisCommands();
+		}
+	}
+
+	protected RedisCommands jedisCommands() {
+		return new JedisCommands();
+	}
+
+	protected RedisCommands lettuceCommands() {
+		return new LettuceCommands();
+	}
+
+}

src/main/java/com/redislabs/riot/cli/CommandsBuilder.java

@@ -0,0 +1,33 @@
+package com.redislabs.riot.cli;
+
+import javax.sql.DataSource;
+
+import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
+
+import com.zaxxer.hikari.HikariDataSource;
+
+import lombok.Data;
+import picocli.CommandLine.Option;
+
+@Data
+public class DatabaseConnectionOptions {
+
+	@Option(names = "--driver", description = "Fully qualified name of the JDBC driver. Auto-detected based on the URL by default.")
+	private String driverClassName;
+	@Option(names = "--url", description = "JDBC URL of the database.", required = true)
+	private String url;
+	@Option(names = "--username", description = "Login username of the database.")
+	private String username;
+	@Option(names = "--password", description = "Login password of the database.", interactive = true)
+	private String password;
+
+	public DataSource dataSource() {
+		DataSourceProperties properties = new DataSourceProperties();
+		properties.setUrl(url);
+		properties.setDriverClassName(driverClassName);
+		properties.setUsername(username);
+		properties.setPassword(password);
+		return properties.initializeDataSourceBuilder().type(HikariDataSource.class).build();
+	}
+
+}

src/main/java/com/redislabs/riot/cli/DatabaseConnectionOptions.java

@@ -0,0 +1,33 @@
+package com.redislabs.riot.cli;
+
+import java.util.Map;
+
+import org.springframework.batch.item.database.JdbcBatchItemWriter;
+import org.springframework.batch.item.database.builder.JdbcBatchItemWriterBuilder;
+import org.springframework.jdbc.core.namedparam.MapSqlParameterSource;
+
+import picocli.CommandLine.Command;
+import picocli.CommandLine.Mixin;
+import picocli.CommandLine.Option;
+
+@Command(name = "db", description = "Export to a database")
+public class DatabaseExport extends ExportSub {
+
+	@Mixin
+	private DatabaseConnectionOptions connection = new DatabaseConnectionOptions();
+
+	@Option(names = "--sql", description = "Insert query, e.g. \"INSERT INTO people (id, name) VALUES (:id, :name)\"", required = true)
+	private String sql;
+
+	@Override
+	protected JdbcBatchItemWriter<Map<String, Object>> writer() {
+		JdbcBatchItemWriterBuilder<Map<String, Object>> builder = new JdbcBatchItemWriterBuilder<Map<String, Object>>();
+		builder.itemSqlParameterSourceProvider(MapSqlParameterSource::new);
+		builder.dataSource(connection.dataSource());
+		builder.sql(sql);
+		JdbcBatchItemWriter<Map<String, Object>> writer = builder.build();
+		writer.afterPropertiesSet();
+		return writer;
+	}
+
+}