java-topology/defects/activemq/unit/ActiveMQTopicConsumerTest.java

package unit;

import java.util.*;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.ConcurrentHashMap;

/**
 * ActiveMQTopicConsumerTest — Java model of CWE-407 defect in ActiveMQ Topic.
 *
 * ACTIVEMQ-0001 (MEDIUM): Topic.addSubscription — CopyOnWriteArrayList.contains per subscribe.
 *
 *   consumers is declared as CopyOnWriteArrayList<Subscription>. On every subscribe:
 *
 *     synchronized (consumers) {
 *         if (!consumers.contains(sub)) {
 *             consumers.add(sub);
 *         }
 *     }
 *
 *   CopyOnWriteArrayList.contains() is O(N). With N subscriptions arriving sequentially,
 *   total work is O(1 + 2 + ... + N) = O(N²).
 *
 *   Fix: maintain a parallel Set<Subscription> backed by ConcurrentHashMap for O(1) membership.
 *   Use set.add() (which returns false on duplicate) instead of contains() + add().
 */
public class ActiveMQTopicConsumerTest {

    // -----------------------------------------------------------------------
    // Algorithm models — instrumented with operation counts
    // -----------------------------------------------------------------------

    /**
     * Defective: CopyOnWriteArrayList.contains() — O(n) scan per subscriber.
     * Models N sequential subscribe calls. Returns total comparison count.
     */
    static long slow(int totalSubscribers) {
        List<Integer> consumers = new ArrayList<>();  // models CopyOnWriteArrayList
        long comparisons = 0;
        for (int sub = 0; sub < totalSubscribers; sub++) {
            // CopyOnWriteArrayList.contains: linear scan of current list
            boolean found = false;
            for (int existing : consumers) {
                comparisons++;
                if (existing == sub) { found = true; break; }
            }
            if (!found) {
                consumers.add(sub);
            }
        }
        return comparisons;
    }

    /**
     * Fixed: Set.add() returns false on duplicate — O(1) membership per subscribe.
     * Models N sequential subscribe calls. Returns total operation count.
     */
    static long fast(int totalSubscribers) {
        List<Integer> consumers = new ArrayList<>();  // models CopyOnWriteArrayList (for dispatch)
        Set<Integer> consumerSet = new HashSet<>();   // parallel O(1) membership guard
        long ops = 0;
        for (int sub = 0; sub < totalSubscribers; sub++) {
            ops++;  // O(1) set.add()
            if (consumerSet.add(sub)) {
                consumers.add(sub);
            }
        }
        return ops;
    }

    // -----------------------------------------------------------------------
    // Tests
    // -----------------------------------------------------------------------

    /**
     * Test 1: N=200 distinct subscribers.
     * slow(): 0 + 1 + 2 + ... + 199 = N*(N-1)/2 = 19900 comparisons.
     * fast(): N = 200 operations.
     * Assert slow > fast * 10x.
     */
    static void test_activemq0001_comparison_ratio() {
        final int N = 200;
        long sOps = slow(N);
        long fOps = fast(N);

        double ratio = (double) sOps / fOps;
        System.out.printf("  ACTIVEMQ-0001 ratio  slow=%d  fast=%d  ratio=%.1fx%n",
                          sOps, fOps, ratio);
        assert sOps > fOps * 10 :
            "ACTIVEMQ-0001: expected slow > fast*10, got slow=" + sOps + " fast=" + fOps;
    }

    /**
     * Test 2: scaling — doubling N quadruples slow() cost, doubles fast().
     * slow(2N) / slow(N) ≈ 4x.  fast(2N) / fast(N) = 2x.
     */
    static void test_activemq0001_quadratic_scaling() {
        final int N_LO = 100, N_HI = 200;
        long sLo = slow(N_LO);
        long sHi = slow(N_HI);
        long fLo = fast(N_LO);
        long fHi = fast(N_HI);

        double sScale = (double) sHi / sLo;
        double fScale = (double) fHi / fLo;
        System.out.printf("  ACTIVEMQ-0001 N scaling  slowScale=%.1fx  fastScale=%.1fx%n",
                          sScale, fScale);
        // slow() is O(N²): doubling N should ~4x the cost
        assert sScale > 3.0 : "ACTIVEMQ-0001: slow should scale quadratically, got " + sScale;
        // fast() is O(N): doubling N should ~2x the cost
        assert fScale < 3.0 : "ACTIVEMQ-0001: fast should scale linearly, got " + fScale;
        // fast is always cheaper
        assert fHi < sHi : "ACTIVEMQ-0001: fast should be cheaper than slow at N=" + N_HI;
    }

    /**
     * Test 3: large scale — N=1000 subscribers.
     * slow(): ~500k comparisons. fast(): 1000 ops. Ratio > 100x.
     */
    static void test_activemq0001_large_scale() {
        final int N = 1000;
        long sOps = slow(N);
        long fOps = fast(N);

        double ratio = (double) sOps / fOps;
        System.out.printf("  ACTIVEMQ-0001 N=1000  slow=%d  fast=%d  ratio=%.1fx%n",
                          sOps, fOps, ratio);
        assert sOps > fOps * 100 :
            "ACTIVEMQ-0001: expected slow > fast*100 at N=1000, got slow=" + sOps + " fast=" + fOps;
    }

    /**
     * Test 4: correctness — both paths produce the same consumer list.
     * Use a mixed scenario: 50 unique subs + 20 duplicate re-subscribes.
     */
    static void test_activemq0001_correctness() {
        final int UNIQUE = 50;
        List<Integer> subscribeOrder = new ArrayList<>();
        for (int i = 0; i < UNIQUE; i++) subscribeOrder.add(i);
        // add 20 duplicates
        for (int i = 0; i < 20; i++) subscribeOrder.add(i % UNIQUE);

        // Defective path
        List<Integer> slowConsumers = new ArrayList<>();
        for (int sub : subscribeOrder) {
            if (!slowConsumers.contains(sub)) slowConsumers.add(sub);
        }

        // Fixed path
        List<Integer> fastConsumers = new ArrayList<>();
        Set<Integer> fastSet = new HashSet<>();
        for (int sub : subscribeOrder) {
            if (fastSet.add(sub)) fastConsumers.add(sub);
        }

        List<Integer> slowSorted = new ArrayList<>(slowConsumers);
        List<Integer> fastSorted = new ArrayList<>(fastConsumers);
        Collections.sort(slowSorted);
        Collections.sort(fastSorted);

        System.out.printf("  ACTIVEMQ-0001 correctness: consumers=%d (slow=%d fast=%d)%n",
                          UNIQUE, slowConsumers.size(), fastConsumers.size());
        assert slowSorted.equals(fastSorted) :
            "ACTIVEMQ-0001 correctness: consumer lists differ: " + slowSorted + " vs " + fastSorted;
        assert fastConsumers.size() == UNIQUE :
            "ACTIVEMQ-0001: expected " + UNIQUE + " unique consumers, got " + fastConsumers.size();
    }

    // -----------------------------------------------------------------------
    // Main
    // -----------------------------------------------------------------------

    public static void main(String[] args) {
        System.out.println("ActiveMQTopicConsumerTest — CWE-407 model tests (ACTIVEMQ-0001)");
        System.out.println("================================================================");

        run("test_activemq0001_comparison_ratio",  ActiveMQTopicConsumerTest::test_activemq0001_comparison_ratio);
        run("test_activemq0001_quadratic_scaling", ActiveMQTopicConsumerTest::test_activemq0001_quadratic_scaling);
        run("test_activemq0001_large_scale",       ActiveMQTopicConsumerTest::test_activemq0001_large_scale);
        run("test_activemq0001_correctness",       ActiveMQTopicConsumerTest::test_activemq0001_correctness);

        System.out.println("================================================================");
        System.out.println("4/4 PASS");
    }

    @FunctionalInterface interface TestFn { void run() throws Exception; }

    static void run(String name, TestFn fn) {
        System.out.print("  [RUN]  " + name + " ... ");
        try {
            fn.run();
            System.out.println("PASS");
        } catch (AssertionError e) {
            System.out.println("FAIL — " + e.getMessage());
            System.exit(1);
        } catch (Exception e) {
            System.out.println("ERR — " + e);
            System.exit(1);
        }
    }
}