AbstractBlockingQueueTest.java

package by.andd3dfx.multithreading.queue;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.SneakyThrows;
import org.junit.Before;
import org.junit.Test;

import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.CompletableFuture;

import static org.assertj.core.api.Assertions.assertThat;
import static org.awaitility.Awaitility.await;
import static org.awaitility.Durations.ONE_HUNDRED_MILLISECONDS;
import static org.awaitility.Durations.ONE_SECOND;
import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertTrue;

public abstract class AbstractBlockingQueueTest {

    private final int ITEMS_AMOUNT = 20;
    private CustomBlockingQueue<QueueItem> queue;
    private Random random = new Random();

    @Before
    public void setUp() {
        queue = buildQueue(ITEMS_AMOUNT);
    }

    @Data
    @AllArgsConstructor
    public class QueueItem {
        String name;
        int number;
    }

    protected abstract CustomBlockingQueue<QueueItem> buildQueue(int size);

    @Test
    public void testEnqueueWaitingOfDequeue() {
        for (int i = 0; i < ITEMS_AMOUNT; i++) {
            enqueue("EnqueueDequeue", i);
        }
        var enqueueFuture = CompletableFuture.runAsync(() -> {
            enqueue("EnqueueDequeue", ITEMS_AMOUNT + 1);
        });
        sleep(500);
        assertFalse(enqueueFuture.isDone());

        dequeue();
        await()
                .atMost(ONE_SECOND)
                .pollInterval(ONE_HUNDRED_MILLISECONDS)
                .until(() -> enqueueFuture.isDone());
        assertTrue(enqueueFuture.isDone());
    }

    @Test
    public void testDequeueWaitingOfEnqueue() {
        var dequeueFuture = CompletableFuture.runAsync(() -> {
            dequeue();
        });
        sleep(500);
        assertFalse(dequeueFuture.isDone());

        enqueue("DequeueEnqueue", 1);
        await()
                .atMost(ONE_SECOND)
                .pollInterval(ONE_HUNDRED_MILLISECONDS)
                .until(() -> dequeueFuture.isDone());
        assertTrue(dequeueFuture.isDone());
    }

    @SneakyThrows
    @Test
    public void testParallelism() {
        Runnable enqueue1 = () -> {
            int counter = 0;
            while (counter < ITEMS_AMOUNT) {
                enqueue("A", counter);
                sleep();
                counter++;
            }
        };
        Runnable enqueue2 = () -> {
            int counter = 0;
            while (counter < ITEMS_AMOUNT) {
                enqueue("B", counter);
                sleep();
                counter++;
            }
        };

        Callable<List<QueueItem>> dequeue = () -> {
            List<QueueItem> result = new ArrayList<>(2 * ITEMS_AMOUNT);
            int counter = 0;
            while (counter < 2 * ITEMS_AMOUNT) {
                var item = dequeue();
                result.add(item);
                sleep();
                counter++;
            }
            return result;
        };

        CompletableFuture.runAsync(enqueue1);
        CompletableFuture.runAsync(enqueue2);
        CompletableFuture<Callable<List<QueueItem>>> dequeueFuture = CompletableFuture.supplyAsync(() -> dequeue);
        await()
                .atMost(ONE_SECOND)
                .pollInterval(ONE_HUNDRED_MILLISECONDS)
                .until(() -> dequeueFuture.isDone());

        // Test items distribution
        // Если значения распределены равномерно, средние значения для А и B будут примерно одинаковы,
        // если (худший случай) вначале идут значения одного потока, а потом другого -
        // средние значения будут N^2/2 и 3N^2/2 соответственно.
        // Поэтому проверяем, что разница между ними меньше чем N^2/2 (50% от макс. разницы)
        var itemsList = dequeueFuture.get().call();
        var valueA = 0;
        var valueB = 0;
        for (int i = 0; i < 2 * ITEMS_AMOUNT; i++) {
            var item = itemsList.get(i);
            switch (item.name) {
                case "A":
                    valueA += i;
                    break;
                case "B":
                    valueB += i;
                    break;
            }
        }
        System.out.println(MessageFormat.format("valueA={0}, valueB={1}", valueA, valueB));
        assertThat(Math.abs(valueA - valueB)).isLessThan(ITEMS_AMOUNT * ITEMS_AMOUNT / 2);
    }

    @SneakyThrows
    private void enqueue(String name, int counter) {
        queue.enqueue(new QueueItem(name, counter));
    }

    @SneakyThrows
    private QueueItem dequeue() {
        return queue.dequeue();
    }

    @SneakyThrows
    private void sleep() {
        Thread.sleep(random.nextInt(ITEMS_AMOUNT));
    }

    @SneakyThrows
    private void sleep(long ms) {
        Thread.sleep(ms);
    }
}