Fix mistake in unit test for ClientProcessor, add concurrent unit tes…

…t for ClientProcessor
aplotnikov · Oct 28, 2018 · 52d9cb8 · 52d9cb8
1 parent 2c31b35
commit 52d9cb8
Show file tree

Hide file tree

Showing 2 changed files with 46 additions and 2 deletions.
diff --git a/src/main/java/io/github/aplotnikov/batch/processing/reactor/ClientProcessor.java b/src/main/java/io/github/aplotnikov/batch/processing/reactor/ClientProcessor.java
@@ -32,7 +32,7 @@ private boolean isCurrentResponseSuccessful() {
 
         do {
             isSuccessful = isLastResponseSuccessful.get();
-        } while (isLastResponseSuccessful.compareAndExchange(isSuccessful, !isSuccessful));
+        } while (!isLastResponseSuccessful.compareAndSet(isSuccessful, !isSuccessful));
 
         return isSuccessful;
     }

diff --git a/src/test/groovy/io/github/aplotnikov/batch/processing/reactor/ClientProcessorSpec.groovy b/src/test/groovy/io/github/aplotnikov/batch/processing/reactor/ClientProcessorSpec.groovy
@@ -1,10 +1,17 @@
 package io.github.aplotnikov.batch.processing.reactor
 
 import static io.github.aplotnikov.batch.processing.reactor.Response.Status.FAILED
+import static io.github.aplotnikov.batch.processing.reactor.Response.Status.SUCCESS
+import static java.util.concurrent.TimeUnit.SECONDS
 
 import spock.lang.Specification
 import spock.lang.Subject
 
+import java.util.concurrent.CountDownLatch
+import java.util.concurrent.ExecutorService
+import java.util.concurrent.Executors
+import java.util.concurrent.atomic.AtomicInteger
+
 class ClientProcessorSpec extends Specification {
 
     @Subject
@@ -29,7 +36,7 @@ class ClientProcessorSpec extends Specification {
         then:
             with(response) {
                 clientId == secondClient.id
-                status == FAILED
+                status == SUCCESS
             }
         when:
             response = processor.process(thirdClient)
@@ -39,4 +46,41 @@ class ClientProcessorSpec extends Specification {
                 status == FAILED
             }
     }
+
+    void 'should work correctly under concurrent usage'() {
+        given:
+            int threadNumber = 10
+        and:
+            ExecutorService threadPool = Executors.newFixedThreadPool(threadNumber)
+        and:
+            CountDownLatch afterInitBlocker = new CountDownLatch(threadNumber)
+            CountDownLatch allDone = new CountDownLatch(threadNumber)
+        and:
+            AtomicInteger clientNumber = new AtomicInteger(1)
+        and:
+            List<Response> actualResponses = [].asSynchronized()
+        and:
+            Runnable action = {
+                int id = clientNumber.getAndIncrement()
+
+                afterInitBlocker.countDown()
+                afterInitBlocker.await()
+
+                actualResponses << processor.process(new Client(id))
+
+                allDone.countDown()
+            }
+        when:
+            try {
+                (1..threadNumber).each {
+                    threadPool.submit(action)
+                }
+                allDone.await(10, SECONDS)
+            } finally {
+                threadPool.shutdown()
+            }
+        then:
+            actualResponses.size() == threadNumber
+            actualResponses.findAll { it.status == SUCCESS }.size() == threadNumber / 2
+    }
 }