#585 - Multi-threaded tests for SyncIterator class

src/test/java/org/cactoos/iterator/SyncIteratorTest.java

@@ -23,16 +23,22 @@
  */
 package org.cactoos.iterator;
 
+import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.ReentrantReadWriteLock;
+import junit.framework.TestCase;
 import org.cactoos.list.ListOf;
+import org.cactoos.list.StickyList;
 import org.hamcrest.MatcherAssert;
 import org.hamcrest.Matchers;
 import org.junit.Test;
 
-// @todo #482:30min add multi-threaded tests which test that the lock syncs the
-//  access to the next method against next and hasNext calls and calls to the
-//  hasNext method against next calls.
 /**
  * Test for {@link SyncIterator}.
  *
@@ -73,4 +79,201 @@ public void syncIteratorReturnsCorrectValuesWithInternalLock() {
         );
     }
 
+    @Test
+    @SuppressWarnings("PMD.AvoidInstantiatingObjectsInLoops")
+    public void correctValuesForConcurrentNextNext()
+        throws InterruptedException {
+        for (int iter = 0; iter < 5000; iter += 1) {
+            final List<String> list = Arrays.asList("a", "b");
+            final SyncIterator<String> iterator = new SyncIterator<>(
+                list.iterator()
+            );
+            final List<Object> sync =
+                Collections.synchronizedList(
+                    new ArrayList<>(list.size())
+                );
+            final Runnable first = () -> {
+                sync.add(iterator.next());
+            };
+            final Runnable second = () -> {
+                sync.add(iterator.next());
+            };
+            new Concurrent(first, second).launch();
+            MatcherAssert.assertThat(
+                "Missing the list items(s) (next()).",
+                sync,
+                Matchers.containsInAnyOrder("a", "b")
+            );
+        }
+    }
+
+    @Test
+    @SuppressWarnings("PMD.AvoidInstantiatingObjectsInLoops")
+    public void correctValuesForConcurrentNextHasNext()
+        throws InterruptedException {
+        for (int iter = 0; iter < 5000; iter += 1) {
+            final List<String> list = Arrays.asList("a", "b");
+            final SyncIterator<String> iterator = new SyncIterator<>(
+                list.iterator()
+            );
+            final List<Object> sync =
+                Collections.synchronizedList(
+                    new ArrayList<>(list.size())
+                );
+            final Runnable first = () -> {
+                sync.add(iterator.next());
+            };
+            final Runnable second = () -> {
+                sync.add(iterator.next());
+            };
+            final Runnable third = () -> {
+                sync.add(iterator.hasNext());
+            };
+            new Concurrent(first, second, third).launch();
+            MatcherAssert.assertThat(
+                "Missing the list items(s) (next()).",
+                sync,
+                Matchers.allOf(
+                    Matchers.hasItem("a"),
+                    Matchers.hasItem("b")
+                )
+            );
+            MatcherAssert.assertThat(
+                "Missing hasNext() value.",
+                sync,
+                Matchers.anyOf(
+                    Matchers.hasItem(true),
+                    Matchers.hasItem(false)
+                )
+            );
+        }
+    }
+
+    @Test
+    @SuppressWarnings("PMD.AvoidInstantiatingObjectsInLoops")
+    public void correctValuesForConcurrentHasNextHasNext()
+        throws InterruptedException {
+        for (int iter = 0; iter < 5000; iter += 1) {
+            final List<String> list = Arrays.asList("a", "b");
+            final SyncIterator<String> iterator = new SyncIterator<>(
+                list.iterator()
+            );
+            final List<Object> sync =
+                Collections.synchronizedList(
+                    new ArrayList<>(list.size())
+                );
+            final Runnable first = () -> {
+                sync.add(iterator.hasNext());
+            };
+            final Runnable second = () -> {
+                sync.add(iterator.hasNext());
+            };
+            new Concurrent(first, second).launch();
+            MatcherAssert.assertThat(
+                "Missing hasNext() value(s).",
+                sync,
+                Matchers.contains(true, true)
+            );
+        }
+    }
+
+    /**
+     * Tests runnables for concurrency issues.
+     */
+    private final class Concurrent {
+
+        /**
+         * Runnables to run in different threads.
+         */
+        private final List<Runnable> runnables;
+
+        /**
+         * Collected exceptions.
+         */
+        private final List<Throwable> exceptions;
+
+        /**
+         * Thread pool.
+         */
+        private final ExecutorService pool;
+
+        /**
+         * All executor threads are ready.
+         */
+        private final CountDownLatch ready;
+
+        /**
+         * Start countdown with first thread.
+         */
+        private final CountDownLatch init;
+
+        /**
+         * All threads ready.
+         */
+        private final CountDownLatch done;
+
+        Concurrent(final Runnable... runnables) {
+            this.runnables = new StickyList<Runnable>(
+                new ListOf<Runnable>(runnables)
+            );
+            this.exceptions =  Collections.synchronizedList(
+                new ArrayList<Throwable>(
+                    runnables.length
+                )
+            );
+            this.pool = Executors.newFixedThreadPool(runnables.length);
+            this.ready = new CountDownLatch(runnables.length);
+            this.init = new CountDownLatch(1);
+            this.done = new CountDownLatch(runnables.length);
+        }
+
+        //@checkstyle IllegalCatchCheck (100 lines)
+        @SuppressWarnings({"PMD.ProhibitPlainJunitAssertionsRule",
+            "PMD.AvoidCatchingThrowable"})
+        public void launch() throws InterruptedException {
+            try {
+                for (final Runnable runnable : this.runnables) {
+                    this.pool.submit(
+                        () -> {
+                            this.ready.countDown();
+                            try {
+                                this.init.await();
+                                runnable.run();
+                            } catch (final Throwable ex) {
+                                this.exceptions.add(ex);
+                            } finally {
+                                this.done.countDown();
+                            }
+                        });
+                }
+                TestCase.assertTrue(
+                    "Timeout initializing threads! Perform longer thread init.",
+                    this.ready.await(
+                        this.runnables.size() * 50,
+                        TimeUnit.MILLISECONDS
+                    )
+                );
+                this.init.countDown();
+                TestCase.assertTrue(
+                    String.format(
+                        "Timeout! More than %d seconds",
+                        10
+                    ),
+                    this.done.await(100, TimeUnit.SECONDS)
+                );
+            } finally {
+                this.pool.shutdownNow();
+            }
+            TestCase.assertTrue(
+                String.format(
+                    "%s failed with exception(s) %s",
+                    "Error",
+                    this.exceptions.toString()
+                ),
+                this.exceptions.isEmpty()
+            );
+        }
+
+    }
+
 }