/*
 * Copyright © 2012-2025 by Data Virtuality GmbH
 */
package com.datavirtuality.dv.core.interfaces.concurrent;

import org.junit.jupiter.api.Test;

import java.util.concurrent.Semaphore;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.function.Consumer;
import java.util.function.Supplier;

import static org.assertj.core.api.Assertions.*;

/**
 * @author Roman Pedchenko
 * @date 2024.10.02
 */
class TestRefreshSemaphore {
    @Test
    void nullSemaphore() {
        final RefreshSemaphore.WriteLock writeLock = RefreshSemaphore.writeLock(null, true);
        assertThat(writeLock).isNotNull();
    }

    @Test
    void writeLock() throws Exception {
        AtomicBoolean full1 = new AtomicBoolean(true);
        AtomicBoolean full2 = new AtomicBoolean(true);
        RefreshSemaphore sem = new RefreshSemaphore("schema1");

        runTwoThreads(
                () -> RefreshSemaphore.writeLock(sem, false),
                writeLock -> full1.set(writeLock.canSkip()),
                () -> RefreshSemaphore.writeLock(sem, true),
                writeLock -> full2.set(writeLock.canSkip()));

        assertThat(full1.get()).isFalse();
        assertThat(full2.get()).isFalse();
    }

    @Test
    void readWriteLock() throws Exception {
        AtomicInteger i1 = new AtomicInteger(0);
        RefreshSemaphore sem = new RefreshSemaphore("schema1");

        for (int i = 0; i < 10; i++) {
            runTwoThreads(
                    () -> RefreshSemaphore.readLock(sem),
                    l -> i1.set(1),
                    () -> RefreshSemaphore.writeLock(sem, false),
                    l -> i1.set(2));
            assertThat(i1.get()).isEqualTo(2);
        }
    }

    @Test
    void writeReadLock() throws Exception {
        AtomicInteger i1 = new AtomicInteger(0);
        RefreshSemaphore sem = new RefreshSemaphore("schema1");

        for (int i = 0; i < 10; i++) {
            runTwoThreads(
                    () -> RefreshSemaphore.writeLock(sem, false),
                    l -> i1.set(1),
                    () -> RefreshSemaphore.readLock(sem),
                    l -> i1.set(2));

            assertThat(i1.get()).isEqualTo(2);
        }
    }

    @Test
    void writeLock2() throws Exception {
        AtomicBoolean full1 = new AtomicBoolean(true);
        AtomicBoolean full2 = new AtomicBoolean(false);
        RefreshSemaphore sem = new RefreshSemaphore("schema1");

        runTwoThreads(
                () -> RefreshSemaphore.writeLock(sem, true),
                writeLock -> full1.set(writeLock.canSkip()),
                () -> RefreshSemaphore.writeLock(sem, true),
                writeLock -> full2.set(writeLock.canSkip()));

        assertThat(full1.get()).isFalse();
        assertThat(full2.get()).isTrue();
    }

    private static <T extends RefreshSemaphore.Lock> void runTwoThreads(
            Supplier<T> tr1, Consumer<T> run1,
            Supplier<T> tr2, Consumer<T> run2) throws Exception {
        Semaphore helperLock = new Semaphore(1);
        Semaphore helperLock2 = new Semaphore(1);
        helperLock.acquireUninterruptibly();
        helperLock2.acquireUninterruptibly();
        final Thread t1 = new Thread(() -> {
            try (final T lock = tr1.get()) {
                helperLock2.release();
                run1.accept(lock);
                helperLock.acquireUninterruptibly();
                Thread.sleep(200);
            } catch (InterruptedException ignored) {
            }
        });
        t1.start();

        final Thread t2 = new Thread(() -> {
            helperLock2.acquireUninterruptibly();
            helperLock.release();
            try (final T lock = tr2.get()) {
                run2.accept(lock);
            }
        });
        t2.start();
        t1.join();
        t2.join();
    }

    @Test
    void writeLock3() throws Exception {
        AtomicBoolean full1 = new AtomicBoolean(true);
        AtomicBoolean full2 = new AtomicBoolean(false);
        AtomicBoolean full3 = new AtomicBoolean(false);
        RefreshSemaphore sem = new RefreshSemaphore("schema1");

        runThreeThreads(
                () -> RefreshSemaphore.writeLock(sem, true),
                writeLock -> full1.set(writeLock.canSkip()),
                () -> RefreshSemaphore.writeLock(sem, false),
                writeLock -> full2.set(writeLock.canSkip()),
                () -> RefreshSemaphore.writeLock(sem, true),
                writeLock -> full3.set(writeLock.canSkip())
        );

        assertThat(full1.get()).isFalse();
        assertThat(full2.get()).isTrue();
        assertThat(full3.get()).isFalse();
    }

    private static <T extends RefreshSemaphore.Lock> void runThreeThreads(
            Supplier<T> tr1, Consumer<T> run1,
            Supplier<T> tr2, Consumer<T> run2,
            Supplier<T> tr3, Consumer<T> run3) throws Exception {
        Semaphore helperLock = new Semaphore(1);
        Semaphore helperLock2 = new Semaphore(1);
        Semaphore helperLock3 = new Semaphore(1);
        helperLock.acquireUninterruptibly();
        helperLock2.acquireUninterruptibly();
        helperLock3.acquireUninterruptibly();
        final Thread t1 = new Thread(() -> {
            try (final T lock = tr1.get()) {
                helperLock2.release();
                run1.accept(lock);
                helperLock.acquireUninterruptibly();
                Thread.sleep(200);
            } catch (InterruptedException ignored) {
            }
        });
        t1.start();

        final Thread t2 = new Thread(() -> {
            helperLock2.acquireUninterruptibly();
            helperLock3.release();
            helperLock.release();
            try (final T lock = tr2.get()) {
                run2.accept(lock);
            }
        });
        t2.start();

        final Thread t3 = new Thread(() -> {
            helperLock3.acquireUninterruptibly();
            helperLock2.release();
            try (final T lock = tr3.get()) {
                run3.accept(lock);
            }
        });
        t3.start();
        t1.join();
        t2.join();
        t3.join();
    }
}
