Added custom AblyTestSuite to run tests faster on local and CI

Author: sacOO7

lib/src/test/java/io/ably/lib/test/common/toolkit/AblyTestSuite.java

@@ -0,0 +1,12 @@
+package io.ably.lib.test.common.toolkit;
+
+import org.junit.runners.Suite;
+import org.junit.runners.model.InitializationError;
+import org.junit.runners.model.RunnerBuilder;
+
+public class AblyTestSuite extends Suite {
+    public AblyTestSuite(Class<?> klass, RunnerBuilder builder) throws InitializationError {
+        super(klass, builder);
+        this.setScheduler(new ParallelScheduler());
+    }
+}

lib/src/test/java/io/ably/lib/test/common/toolkit/CatThrower.java

@@ -0,0 +1,15 @@
+package io.ably.lib.test.common.toolkit;
+
+public class CatThrower<T extends Throwable> {
+
+    /**
+     * Will throw the given {@link Throwable}.
+     */
+    public static void sneakyThrow(Throwable t) {
+        new CatThrower<Error>().sneakyThrow2(t);
+    }
+
+    private void sneakyThrow2(Throwable t) throws T {
+        throw (T) t;
+    }
+}

lib/src/test/java/io/ably/lib/test/common/toolkit/MultiException.java

@@ -0,0 +1,71 @@
+package io.ably.lib.test.common.toolkit;
+
+
+import java.io.PrintWriter;
+import java.io.StringWriter;
+import java.util.ArrayList;
+import java.util.List;
+
+public class MultiException extends RuntimeException {
+    private static final long serialVersionUID = 1L;
+    private static final String EXCEPTION_SEPARATOR = "\n\t______________________________________________________________________\n";
+    private final List<Throwable> nested = new ArrayList<>();
+
+    public MultiException() {
+        super("Multiple exceptions");
+    }
+
+    public void add(Throwable throwable) {
+        if (throwable != null) {
+            synchronized (nested) {
+                if (throwable instanceof MultiException) {
+                    MultiException other = (MultiException) throwable;
+                    synchronized (other.nested) {
+                        nested.addAll(other.nested);
+                    }
+                } else {
+                    nested.add(throwable);
+                }
+            }
+        }
+    }
+
+    /**
+     * If this <code>MultiException</code> is empty then no action is taken,
+     * if it contains a single <code>Throwable</code> that is thrown,
+     * otherwise this <code>MultiException</code> is thrown.
+     */
+    public void throwIfNotEmpty() {
+        synchronized (nested) {
+            if (nested.isEmpty()) {
+                // Do nothing
+            } else if (nested.size() == 1) {
+                Throwable t = nested.get(0);
+                CatThrower.sneakyThrow(t);
+            } else {
+                throw this;
+            }
+        }
+    }
+
+    @Override
+    public String getMessage() {
+        synchronized (nested) {
+            if (nested.isEmpty()) {
+                return "<no nested exceptions>";
+            } else {
+                StringBuilder sb = new StringBuilder();
+                int n = nested.size();
+                sb.append(n).append(n == 1 ? " nested exception:" : " nested exceptions:");
+                for (Throwable t : nested) {
+                    sb.append(EXCEPTION_SEPARATOR).append("\n\t");
+                    StringWriter sw = new StringWriter();
+                    t.printStackTrace(new PrintWriter(sw));
+                    sb.append(sw.toString().replace("\n", "\n\t").trim());
+                }
+                sb.append(EXCEPTION_SEPARATOR);
+                return sb.toString();
+            }
+        }
+    }
+}

lib/src/test/java/io/ably/lib/test/common/toolkit/ParallelScheduler.java

@@ -0,0 +1,85 @@
+package io.ably.lib.test.common.toolkit;
+
+import org.junit.runners.model.RunnerScheduler;
+
+import java.util.Deque;
+import java.util.LinkedList;
+import java.util.concurrent.ForkJoinPool;
+import java.util.concurrent.ForkJoinTask;
+import java.util.concurrent.ForkJoinWorkerThread;
+
+import static java.util.concurrent.ForkJoinTask.inForkJoinPool;
+
+/**
+ * ParallelScheduler runs each test class with threadpool of size equal to CPU cores.
+ * Default value can be changed by setting System.setProperty("maxParallelTestThreads", "4")
+ */
+class ParallelScheduler implements RunnerScheduler {
+
+    static ForkJoinPool forkJoinPool = setUpForkJoinPool();
+
+    static ForkJoinPool setUpForkJoinPool() {
+        int numThreads;
+        try {
+            String configuredNumThreads = System.getProperty("maxParallelTestThreads");
+            numThreads = Math.max(2, Integer.parseInt(configuredNumThreads));
+        } catch (Exception ignored) {
+            Runtime runtime = Runtime.getRuntime();
+            numThreads = Math.max(2, runtime.availableProcessors());
+        }
+        ForkJoinPool.ForkJoinWorkerThreadFactory threadFactory = pool -> {
+            if (pool.getPoolSize() >= pool.getParallelism()) {
+                return null;
+            } else {
+                ForkJoinWorkerThread thread = ForkJoinPool.defaultForkJoinWorkerThreadFactory.newThread(pool);
+                thread.setName("JUnit-" + thread.getName());
+                return thread;
+            }
+        };
+        return new ForkJoinPool(numThreads, threadFactory, null, false);
+    }
+
+    private final Deque<ForkJoinTask<?>> _asyncTasks = new LinkedList<>();
+    private Runnable _lastScheduledChild;
+
+    @Override
+    public void schedule(Runnable childStatement) {
+        if (_lastScheduledChild != null) {
+            // Execute previously scheduled child asynchronously ...
+            if (inForkJoinPool()) {
+                _asyncTasks.addFirst(ForkJoinTask.adapt(_lastScheduledChild).fork());
+            } else {
+                _asyncTasks.addFirst(forkJoinPool.submit(_lastScheduledChild));
+            }
+        }
+        // Note: We don't schedule the childStatement immediately here,
+        // but remember it, so that we can synchronously execute the
+        // last scheduled child in the finished method() -- this way,
+        // the current thread does not immediately call join() in the
+        // finished() method, which might block it ...
+        _lastScheduledChild = childStatement;
+    }
+
+    @Override
+    public void finished() {
+        MultiException me = new MultiException();
+        if (_lastScheduledChild != null) {
+            if (inForkJoinPool()) {
+                // Execute the last scheduled child in the current thread ...
+                try { _lastScheduledChild.run(); } catch (Throwable t) { me.add(t); }
+            } else {
+                // Submit the last scheduled child to the ForkJoinPool too,
+                // because all tests should run in the worker threads ...
+                _asyncTasks.addFirst(forkJoinPool.submit(_lastScheduledChild));
+            }
+            // Make sure all asynchronously executed children are done, before we return ...
+            for (ForkJoinTask<?> task : _asyncTasks) {
+                // Note: Because we have added all tasks via addFirst into _asyncTasks,
+                // task.join() is able to steal tasks from other worker threads, if there
+                // are tasks, which have not been started yet.
+                try { task.join(); } catch (Throwable t) { me.add(t); }
+            }
+            me.throwIfNotEmpty();
+        }
+    }
+}

lib/src/test/java/io/ably/lib/test/realtime/RealtimeSuite.java

@@ -1,6 +1,6 @@
 package io.ably.lib.test.realtime;
 
-import com.googlecode.junittoolbox.ParallelSuite;
+import io.ably.lib.test.common.toolkit.AblyTestSuite;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.runner.JUnitCore;
@@ -11,7 +11,7 @@
 
 import io.ably.lib.test.common.Setup;
 
-@RunWith(ParallelSuite.class)
+@RunWith(AblyTestSuite.class)
 @SuiteClasses({
     ConnectionManagerTest.class,
     RealtimeHttpHeaderTest.class,