refactor: centralized resource methods in FileUtils

connectors/riot-file/riot-file.gradle

@@ -16,6 +16,7 @@
  * limitations under the License.
  */
 dependencies {
+    implementation project(':riot-core')
     implementation group: 'info.picocli', name: 'picocli', version: picocliVersion
     annotationProcessor group: 'info.picocli', name: 'picocli-codegen', version: picocliVersion
     implementation 'org.springframework.batch:spring-batch-infrastructure'

connectors/riot-file/src/main/java/com/redis/riot/file/AwsArgs.java

@@ -1,13 +1,13 @@
 package com.redis.riot.file;
 
+import java.net.MalformedURLException;
 import java.net.URI;
 
 import org.springframework.core.io.Resource;
-import org.springframework.util.Assert;
+import org.springframework.core.io.UrlResource;
 import org.springframework.util.StringUtils;
 
 import io.awspring.cloud.s3.InMemoryBufferingS3OutputStreamProvider;
-import io.awspring.cloud.s3.Location;
 import io.awspring.cloud.s3.PropertiesS3ObjectContentTypeResolver;
 import io.awspring.cloud.s3.S3Resource;
 import picocli.CommandLine.ArgGroup;
@@ -31,11 +31,6 @@ public class AwsArgs {
 	@Option(names = "--s3-endpoint", description = "Service endpoint with which the AWS client should communicate (e.g. https://sns.us-west-1.amazonaws.com).", paramLabel = "<url>")
 	private URI endpoint;
 
-	public static boolean isSimpleStorageResource(String location) {
-		Assert.notNull(location, "Location must not be null");
-		return location.toLowerCase().startsWith(Location.S3_PROTOCOL_PREFIX);
-	}
-
 	public Resource resource(String location) {
 		S3ClientBuilder clientBuilder = S3Client.builder();
 		if (region != null) {
@@ -84,4 +79,8 @@ public URI getEndpoint() {
 	public void setEndpoint(URI endpoint) {
 		this.endpoint = endpoint;
 	}
+
+	public static void main(String[] args) throws MalformedURLException {
+		new UrlResource("s3://riot-bucket-jrx/beers.json");
+	}
 }

connectors/riot-file/src/main/java/com/redis/riot/file/FileArgs.java

@@ -37,15 +37,18 @@ public class FileArgs {
 	private char quoteCharacter = DEFAULT_QUOTE_CHARACTER;
 
 	public Resource resource(String location) throws IOException {
-		if (AwsArgs.isSimpleStorageResource(location)) {
-			return amazonS3Args.resource(location);
-		}
-		if (GoogleStorageArgs.isGoogleStorageResource(location)) {
-			return googleStorageArgs.resource(location);
+		if (FileUtils.isStdin(location)) {
+			return new FilenameInputStreamResource(System.in, "stdin", "Standard Input");
 		}
 		if (ResourceUtils.isUrl(location)) {
 			return new UncustomizedUrlResource(location);
 		}
+		if (FileUtils.isAwsStorage(location)) {
+			return amazonS3Args.resource(location);
+		}
+		if (FileUtils.isGoogleStorage(location)) {
+			return googleStorageArgs.resource(location);
+		}
 		return new FileSystemResource(location);
 	}
 

connectors/riot-file/src/main/java/com/redis/riot/file/FileReaderArgs.java

@@ -37,30 +37,12 @@ public class FileReaderArgs extends FileArgs {
 	private int maxItemCount = DEFAULT_MAX_ITEM_COUNT;
 
 	@Override
-	public Resource resource(String location) {
-		if (FileUtils.isStdin(location)) {
-			return new FilenameInputStreamResource(System.in, "stdin", "Standard Input");
-		}
-		Resource resource;
-		try {
-			resource = super.resource(location);
-		} catch (IOException e) {
-			throw new RuntimeIOException("Could not create resource for file " + location, e);
-		}
-		InputStream inputStream;
-		try {
-			inputStream = resource.getInputStream();
-		} catch (IOException e) {
-			throw new RuntimeIOException("Could not open input stream for resource " + resource, e);
-		}
+	public Resource resource(String location) throws IOException {
+		Resource resource = super.resource(location);
+		InputStream inputStream = resource.getInputStream();
 		if (isGzipped() || FileUtils.isGzip(location)) {
-			GZIPInputStream gzipInputStream;
-			try {
-				gzipInputStream = new GZIPInputStream(inputStream);
-			} catch (IOException e) {
-				throw new RuntimeIOException("Could not create gzip input stream for resource " + resource, e);
-			}
-			return new FilenameInputStreamResource(gzipInputStream, resource.getFilename(), resource.getDescription());
+			return new FilenameInputStreamResource(new GZIPInputStream(inputStream), resource.getFilename(),
+					resource.getDescription());
 		}
 		return resource;
 	}

connectors/riot-file/src/main/java/com/redis/riot/file/FileUtils.java

@@ -5,7 +5,6 @@
 import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
-import java.text.MessageFormat;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
@@ -15,12 +14,17 @@
 
 import org.springframework.core.io.Resource;
 import org.springframework.util.ResourceUtils;
+import org.springframework.util.StringUtils;
 
+import com.redis.riot.core.RiotException;
+
+import io.awspring.cloud.s3.Location;
 import picocli.CommandLine;
 import software.amazon.awssdk.regions.Region;
 
 public abstract class FileUtils {
 
+	public static final String GOOGLE_STORAGE_PROTOCOL_PREFIX = "gs://";
 	public static final Pattern EXTENSION_PATTERN = Pattern.compile("(?i)\\.(?<extension>\\w+)(?:\\.(?<gz>gz))?$");
 
 	public static final String CSV = "csv";
@@ -34,10 +38,6 @@ public abstract class FileUtils {
 	private FileUtils() {
 	}
 
-	public static boolean isGzip(String file) {
-		return extensionGroup(file, "gz") != null;
-	}
-
 	private static String extensionGroup(String file, String group) {
 		Matcher matcher = EXTENSION_PATTERN.matcher(file);
 		if (matcher.find()) {
@@ -75,7 +75,7 @@ public static String fileExtension(Resource resource) {
 
 	/**
 	 * 
-	 * @param file Filename that might include a glob pattern
+	 * @param file File path that might include a glob pattern
 	 * @return List of file
 	 * @throws IOException
 	 */
@@ -86,13 +86,8 @@ public static Stream<String> expand(String file) {
 		return Stream.of(file);
 	}
 
-	public static boolean isStdin(String file) {
-		return "-".equals(file);
-	}
-
 	public static boolean isFile(String file) {
-		return !AwsArgs.isSimpleStorageResource(file) && !GoogleStorageArgs.isGoogleStorageResource(file)
-				&& !ResourceUtils.isUrl(file) && !isStdin(file);
+		return !isAwsStorage(file) && !isGoogleStorage(file) && !ResourceUtils.isUrl(file) && !isStdin(file);
 	}
 
 	public static List<Path> expand(Path path) {
@@ -107,13 +102,29 @@ public static List<Path> expand(Path path) {
 			stream.iterator().forEachRemaining(paths::add);
 			return paths;
 		} catch (IOException e) {
-			throw new RuntimeIOException(
-					MessageFormat.format("Could not list files in directory {0} with glob pattern {1}", dir, glob), e);
+			throw new RiotException(
+					String.format("Could not list files in directory %s with glob pattern %s", dir, glob), e);
 		}
 	}
 
 	public static void registerConverters(CommandLine commandLine) {
 		commandLine.registerConverter(Region.class, Region::of);
 	}
 
+	public static boolean isGzip(String file) {
+		return extensionGroup(file, "gz") != null;
+	}
+
+	public static boolean isStdin(String file) {
+		return "-".equals(file);
+	}
+
+	public static boolean isGoogleStorage(String location) {
+		return StringUtils.hasLength(location) && location.toLowerCase().startsWith(GOOGLE_STORAGE_PROTOCOL_PREFIX);
+	}
+
+	public static boolean isAwsStorage(String location) {
+		return StringUtils.hasLength(location) && location.toLowerCase().startsWith(Location.S3_PROTOCOL_PREFIX);
+	}
+
 }

connectors/riot-file/src/main/java/com/redis/riot/file/FileWriterArgs.java

@@ -52,33 +52,18 @@ public FileWriterArgs() {
 	}
 
 	@Override
-	public WritableResource resource(String location) {
+	public WritableResource resource(String location) throws IOException {
 		if (location == null) {
 			return new SystemOutResource();
 		}
-		Resource resource;
-		try {
-			resource = super.resource(location);
-		} catch (IOException e) {
-			throw new RuntimeIOException("Could not get resource " + location, e);
-		}
+		Resource resource = super.resource(location);
 		Assert.notNull(resource, "Could not resolve file " + location);
 		Assert.isInstanceOf(WritableResource.class, resource);
 		WritableResource writableResource = (WritableResource) resource;
 		if (isGzipped() || FileUtils.isGzip(location)) {
-			OutputStream outputStream;
-			try {
-				outputStream = writableResource.getOutputStream();
-			} catch (IOException e) {
-				throw new RuntimeIOException("Could not open output stream on resource " + writableResource, e);
-			}
-			GZIPOutputStream gzipOutputStream;
-			try {
-				gzipOutputStream = new GZIPOutputStream(outputStream);
-			} catch (IOException e) {
-				throw new RuntimeIOException("Could not open gzip output stream on resource " + writableResource, e);
-			}
-			return new OutputStreamResource(gzipOutputStream, resource.getFilename(), resource.getDescription());
+			OutputStream outputStream = writableResource.getOutputStream();
+			return new OutputStreamResource(new GZIPOutputStream(outputStream), resource.getFilename(),
+					resource.getDescription());
 		}
 		return writableResource;
 	}

connectors/riot-file/src/main/java/com/redis/riot/file/GoogleStorageArgs.java

@@ -8,7 +8,6 @@
 import java.util.Base64;
 
 import org.springframework.core.io.Resource;
-import org.springframework.util.Assert;
 
 import com.google.auth.oauth2.GoogleCredentials;
 import com.google.cloud.ServiceOptions;
@@ -22,20 +21,18 @@
 
 public class GoogleStorageArgs {
 
-	public static final String GOOGLE_STORAGE_PROTOCOL_PREFIX = "gs://";
-
 	public static final GcpScope DEFAULT_SCOPE = GcpScope.STORAGE_READ_ONLY;
 
-	@Option(names = "--gcs-key-file", description = "GCS private key (e.g. /usr/local/key.json).", paramLabel = "<file>")
+	@Option(names = "--gcs-key-file", description = "Google Cloud Storage private key (e.g. /usr/local/key.json).", paramLabel = "<file>")
 	private File keyFile;
 
-	@Option(names = "--gcs-project", description = "GCP project id.", paramLabel = "<id>")
+	@Option(names = "--gcs-project", description = "Google Cloud Storage project id.", paramLabel = "<id>")
 	private String projectId;
 
-	@Option(names = "--gcs-key", arity = "0..1", interactive = true, description = "GCS Base64 encoded key.", paramLabel = "<key>")
+	@Option(names = "--gcs-key", arity = "0..1", interactive = true, description = "Google Cloud Storage Base64 encoded key.", paramLabel = "<key>")
 	private String encodedKey;
 
-	@Option(names = "--gcs-scope", description = "GCP scope (default: ${DEFAULT-VALUE}).", paramLabel = "<scope>", hidden = true)
+	@Option(names = "--gcs-scope", description = "Google Cloud Storage scope (default: ${DEFAULT-VALUE}).", paramLabel = "<scope>", hidden = true)
 	private GcpScope scope = DEFAULT_SCOPE;
 
 	public GcpScope getScope() {
@@ -87,9 +84,4 @@ public Resource resource(String location) throws IOException {
 		return new GoogleStorageResource(builder.build().getService(), location);
 	}
 
-	public static boolean isGoogleStorageResource(String location) {
-		Assert.notNull(location, "Location must not be null");
-		return location.toLowerCase().startsWith(GOOGLE_STORAGE_PROTOCOL_PREFIX);
-	}
-
 }

core/riot-core/src/main/java/com/redis/riot/core/AbstractCommand.java

@@ -74,7 +74,7 @@ private static void setBoolean(String property, boolean value) {
 		System.setProperty(property, String.valueOf(value));
 	}
 
-	protected abstract void execute() throws RiotException;
+	protected abstract void execute() throws Exception;
 
 	public LoggingArgs getLoggingArgs() {
 		return loggingArgs;

core/riot-core/src/main/java/com/redis/riot/core/AbstractJobCommand.java

@@ -77,23 +77,21 @@ public void afterPropertiesSet() throws Exception {
 	}
 
 	@Override
-	protected void execute() throws RiotException {
+	protected void execute() throws Exception {
 		JobExecution jobExecution;
 		Job job = job();
 		try {
 			jobExecution = jobLauncher.run(job, new JobParameters());
-		} catch (JobExecutionException e) {
-			throw new RiotException("Could not run job " + job.getName(), e);
 		} finally {
 			shutdown();
 		}
 		if (JobUtils.isFailed(jobExecution.getExitStatus())) {
 			for (StepExecution stepExecution : jobExecution.getStepExecutions()) {
 				if (JobUtils.isFailed(stepExecution.getExitStatus())) {
-					throw new RiotException(stepExecution.getExitStatus().getExitDescription());
+					throw new JobExecutionException(stepExecution.getExitStatus().getExitDescription());
 				}
 			}
-			throw new RiotException(jobExecution.getExitStatus().getExitDescription());
+			throw new JobExecutionException(jobExecution.getExitStatus().getExitDescription());
 		}
 	}
 
@@ -133,7 +131,7 @@ protected boolean shouldShowProgress() {
 		return stepArgs.getProgressArgs().getStyle() != ProgressStyle.NONE;
 	}
 
-	protected abstract Job job() throws RiotException;
+	protected abstract Job job() throws Exception;
 
 	private <I, O> TaskletStep step(Step<I, O> step) {
 		SimpleStepBuilder<I, O> builder = simpleStepBuilder(step);

core/riot-core/src/main/java/com/redis/riot/core/RiotException.java

@@ -1,23 +1,18 @@
 package com.redis.riot.core;
 
 @SuppressWarnings("serial")
-public class RiotException extends Exception {
+public class RiotException extends RuntimeException {
 
-	public RiotException(String message, Exception cause) {
-		super(message, cause);
+	public RiotException(Throwable e) {
+		super(e);
 	}
 
 	public RiotException(String message) {
 		super(message);
 	}
 
-	public RiotException(Exception cause) {
-		super(cause);
-	}
-
-	@Override
-	public synchronized Exception getCause() {
-		return (Exception) super.getCause();
+	public RiotException(String message, Throwable e) {
+		super(message, e);
 	}
 
 }

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

@@ -22,7 +22,7 @@ public class EvaluationContextArgs {
 	public static final String NUMBER_VAR = "number";
 	public static final String REDIS_VAR = "redis";
 
-	@Option(arity = "1..*", names = "--var", description = "SpEL expressions for context variables, in the form var=\"exp\"", paramLabel = "<v=exp>")
+	@Option(arity = "1..*", names = "--var", description = "SpEL expressions for context variables, in the form var=\"exp\". For details see https://docs.spring.io/spring-framework/reference/core/expressions.html", paramLabel = "<v=exp>")
 	private Map<String, Expression> varExpressions = new LinkedHashMap<>();
 
 	@Option(names = "--date-format", description = "Date/time format (default: ${DEFAULT-VALUE}). For details see https://www.baeldung.com/java-simple-date-format#date_time_patterns", paramLabel = "<fmt>")

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

@@ -23,7 +23,7 @@ public class FakerImport extends AbstractImportCommand {
 	public static final int DEFAULT_COUNT = 1000;
 	public static final Locale DEFAULT_LOCALE = Locale.ENGLISH;
 
-	@Parameters(arity = "1..*", description = "SpEL expressions in the form field1=\"exp\" field2=\"exp\"...", paramLabel = "EXPRESSION")
+	@Parameters(arity = "1..*", description = "SpEL expressions in the form field1=\"exp\" field2=\"exp\" etc. For details see https://docs.spring.io/spring-framework/reference/core/expressions.html", paramLabel = "EXPRESSION")
 	private Map<String, Expression> fields = new LinkedHashMap<>();
 
 	@Option(names = "--count", description = "Number of items to generate (default: ${DEFAULT-VALUE}).", paramLabel = "<int>")