fixed logging of error messages

build.gradle

@@ -18,7 +18,7 @@ repositories {
 }
 
 dependencies {
-    api 'com.redislabs:picocli-redis:1.14.1'
+    api 'com.redislabs:picocli-redis:1.14.2'
 	implementation 'org.slf4j:slf4j-jdk14:1.7.28'
     implementation 'org.springframework.batch:spring-batch-core:4.2.1.RELEASE'
     implementation 'org.springframework.boot:spring-boot-autoconfigure:2.2.5.RELEASE'
@@ -148,8 +148,7 @@ asciidoctor {
   outputOptions {
   	separateOutputDirs = false
   }
-  attributes \
-    'commandsdir': 'src/test/resources/commands'  
+  attributes 'commandsdir': 'src/test/resources/commands'
 }
 
 task printVersion {

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-6.2.2-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-6.3-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists

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

@@ -81,6 +81,11 @@ private Map<String, String> fakerFields() {
 		return fields;
 	}
 
+	@Override
+	protected String getMainFlowName() {
+		return "generator";
+	}
+
 	@Override
 	protected String taskName() {
 		return "Generating";

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

@@ -55,6 +55,11 @@ protected ItemWriter<Map<String, Object>> writer() throws Exception {
 		return file.writer();
 	}
 
+	@Override
+	protected String getMainFlowName() {
+		return "export";
+	}
+
 	@Override
 	protected String taskName() {
 		return "Exporting";

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

@@ -45,7 +45,11 @@ protected ItemProcessor<Map<String, Object>, Map<String, Object>> processor() th
 			return composite;
 		}
 		return processor;
+	}
 
+	@Override
+	protected String getMainFlowName() {
+		return "import";
 	}
 
 	@Override

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

@@ -1,46 +1,74 @@
 package com.redislabs.riot.cli;
 
 import com.redislabs.riot.transfer.Metrics;
-
+import com.redislabs.riot.transfer.MetricsProvider;
+import com.redislabs.riot.transfer.Transfer;
+import lombok.Builder;
+import lombok.extern.slf4j.Slf4j;
 import me.tongfei.progressbar.ProgressBar;
 import me.tongfei.progressbar.ProgressBarBuilder;
 
-public class ProgressBarReporter implements ProgressReporter {
-
-	private ProgressBarBuilder builder = new ProgressBarBuilder();
-	private ProgressBar bar;
-
-	public ProgressBarReporter initialMax(long initialMax) {
-		builder.setInitialMax(initialMax);
-		return this;
-	}
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
 
-	public ProgressBarReporter taskName(String taskName) {
-		builder.setTaskName(taskName);
-		return this;
-	}
+@Slf4j
+public class ProgressBarReporter implements Runnable, Transfer.Listener {
 
-	public ProgressBarReporter unitName(String unitName) {
-		builder.setUnit(" " + unitName + "s", 1);
-		return this;
-	}
+    private ScheduledExecutorService executor;
+    private MetricsProvider metricsProvider;
+    private Long period = null;
+    private TimeUnit timeUnit;
+    private ScheduledFuture<?> scheduledFuture;
+    private ProgressBarBuilder builder = new ProgressBarBuilder();
+    private ProgressBar bar;
 
-	@Override
-	public void start() {
-		this.bar = builder.build();
-	}
+    @Builder
+    private ProgressBarReporter(Long initialMax, String taskName, String unitName, MetricsProvider metricsProvider, Long period, TimeUnit timeUnit) {
+        if (initialMax != null) {
+            builder.setInitialMax(initialMax);
+        }
+        builder.setTaskName(taskName);
+        if (unitName != null) {
+            builder.setUnit(" " + unitName + "s", 1);
+        }
+        if (metricsProvider != null && period != null && timeUnit != null) {
+            this.metricsProvider = metricsProvider;
+            this.period = period;
+            this.timeUnit = timeUnit;
+            this.executor = Executors.newSingleThreadScheduledExecutor();
+            this.scheduledFuture = executor.scheduleAtFixedRate(this, 0, period, timeUnit);
+        }
+    }
 
-	@Override
-	public void onUpdate(Metrics update) {
-		bar.stepTo(update.getWrites());
-		if (update.getRunningThreads() > 1) {
-			bar.setExtraMessage(" (" + update.getRunningThreads() + " threads)");
-		}
-	}
+    @Override
+    public void onOpen() {
+        this.bar = builder.build();
+    }
 
-	@Override
-	public void stop() {
-		this.bar.close();
-	}
+    @Override
+    public void onClose() {
+        if (scheduledFuture != null) {
+            scheduledFuture.cancel(true);
+        }
+        if (executor != null) {
+            executor.shutdown();
+        }
+        run();
+        this.bar.close();
+    }
 
+    @Override
+    public void run() {
+        if (bar == null) {
+            return;
+        }
+        Metrics metrics = metricsProvider.getMetrics();
+        bar.stepTo(metrics.getWrites());
+        int runningThreads = metrics.getRunningThreads();
+        if (runningThreads > 1) {
+            bar.setExtraMessage(" (" + runningThreads + " threads)");
+        }
+    }
 }

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

@@ -47,16 +47,21 @@ public String getChannel(RedisOptions redisOptions) {
 		return channel.replace(DATABASE_TOKEN, String.valueOf(redisOptions.getDatabase()));
 	}
 
+	@Override
+	protected String getMainFlowName() {
+		return "replication";
+	}
+
 	@Override
 	protected RedisItemReader<KeyDump> reader() throws Exception {
 		return reader(KeyDumpReader.builder().timeout(getTimeout()).build());
 	}
 
 	@SuppressWarnings("unchecked")
 	@Override
-	protected Transfer<KeyDump, KeyDump> transfer(ItemReader<KeyDump> reader, ItemProcessor<KeyDump, KeyDump> processor,
+	protected Transfer<KeyDump, KeyDump> transfer(String mainFlowName, ItemReader<KeyDump> reader, ItemProcessor<KeyDump, KeyDump> processor,
 			ItemWriter<KeyDump> writer) {
-		Transfer<KeyDump, KeyDump> transfer = super.transfer(reader, processor, writer);
+		Transfer<KeyDump, KeyDump> transfer = super.transfer(mainFlowName, reader, processor, writer);
 		if (listen) {
 			RedisOptions source = redisOptions();
 			KeyspaceNotificationsIterator iterator = KeyspaceNotificationsIterator.builder()
@@ -75,7 +80,7 @@ protected Transfer<KeyDump, KeyDump> transfer(ItemReader<KeyDump> reader, ItemPr
 
 	@Override
 	protected String taskName() {
-		return "Replicating " + redisOptions().getServers().get(0) + " to " + target.getServers().get(0);
+		return "Replicating";
 	}
 
 	@Override