fixed issue with threads not waiting using CountDownLatch

README.md

@@ -104,7 +104,7 @@ Sample Maven based project are available at...
 * [Java](https://github.com/dataanon/dataanon-java-sample) 
 
 ----------------------
-## Tips
+## Notes
 
 1. In Whitelist approach provide source database connection user with READONLY access.
 2. Use `where` and `limit` to limit the number of rows during anonymization. Very useful for testing purpose.

pom.xml

@@ -5,7 +5,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>com.github.dataanon</groupId>
     <artifactId>data-anon</artifactId>
-    <version>0.9.0</version>
+    <version>0.9.1</version>
     <packaging>jar</packaging>
 
     <name>data-anon</name>

src/main/kotlin/com/github/dataanon/dsl/Strategy.kt

@@ -7,24 +7,32 @@ import com.github.dataanon.model.Table
 import com.github.dataanon.utils.ProgressBarGenerator
 import reactor.core.publisher.Flux
 import reactor.core.scheduler.Schedulers
+import java.util.concurrent.CountDownLatch
+
+
 
 abstract class Strategy {
     protected val tables = mutableListOf<Table>()
 
     fun execute(progressBarEnabled: Boolean = true) {
+        val latch = CountDownLatch(tables.size)
         Flux.fromIterable(tables)
-                .parallel()
+                .parallel(tables.size)
                 .runOn(Schedulers.parallel())
-                .log()
-                .subscribe { table -> executeOnTable(table, progressBarEnabled) }
+                .subscribe { table -> executeOnTable(table, progressBarEnabled, latch) }
+        latch.await()
     }
 
-    private fun executeOnTable(table: Table, progressBarEnabled: Boolean) {
-        val reader      = TableReader(sourceDbConfig(), table)
-        val progressBar = ProgressBarGenerator(progressBarEnabled, table.name, { reader.totalNoOfRecords() })
-        val writer      = TableWriter(destDbConfig(), table, progressBar)
+    private fun executeOnTable(table: Table, progressBarEnabled: Boolean, latch: CountDownLatch) {
+        try {
+            val reader      = TableReader(sourceDbConfig(), table)
+            val progressBar = ProgressBarGenerator(progressBarEnabled, table.name, { reader.totalNoOfRecords() })
+            val writer      = TableWriter(destDbConfig(), table, progressBar)
 
-        Flux.fromIterable(Iterable { reader }).map { table.execute(it) }.subscribe(writer)
+            Flux.fromIterable(Iterable { reader }).map { table.execute(it) }.subscribe(writer)
+        } finally {
+            latch.countDown()
+        }
     }
 
     abstract protected fun sourceDbConfig(): DbConfig