/*
 * Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
 * or more contributor license agreements. Licensed under the Elastic License;
 * you may not use this file except in compliance with the Elastic License.
 */
package org.elasticsearch.watcher.watch;

import org.elasticsearch.ElasticsearchIllegalStateException;
import org.elasticsearch.test.ElasticsearchTestCase;
import org.junit.Test;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;

import static org.hamcrest.Matchers.*;

/**
 */
public class WatchLockServiceTests extends ElasticsearchTestCase {

    @Test
    public void testLocking_notStarted() {
        WatchLockService lockService = new WatchLockService();
        try {
            lockService.acquire("_name");
            fail("exception expected");
        } catch (Exception e) {
            assertThat(e.getMessage(), equalTo("not started"));
        }
    }

    @Test
    public void testLocking() {
        WatchLockService lockService = new WatchLockService();
        lockService.start();
        WatchLockService.Lock lock = lockService.acquire("_name");
        assertThat(lockService.getWatchLocks().hasLockedKeys(), is(true));
        lock.release();
        assertThat(lockService.getWatchLocks().hasLockedKeys(), is(false));
        lockService.stop();
    }

    @Test
    public void testLocking_alreadyHeld() {
        WatchLockService lockService = new WatchLockService();
        lockService.start();
        WatchLockService.Lock lock1 = lockService.acquire("_name");
        try {
            lockService.acquire("_name");
            fail("exception expected");
        } catch (ElasticsearchIllegalStateException e) {
            assertThat(e.getMessage(), containsString("Lock already acquired"));
        }
        lock1.release();
        lockService.stop();
    }

    @Test
    public void testLocking_fair() throws Exception {
        final WatchLockService lockService = new WatchLockService();
        lockService.start();
        final AtomicInteger value = new AtomicInteger(0);
        List<Thread> threads = new ArrayList<>();

        class FairRunner implements Runnable {

            final int expectedValue;

            FairRunner(int expectedValue) {
                this.expectedValue = expectedValue;
            }

            @Override
            public void run() {
                WatchLockService.Lock lock = lockService.acquire("_name");
                try {
                    int actualValue = value.getAndIncrement();
                    assertThat(actualValue, equalTo(expectedValue));
                    Thread.sleep(50);
                } catch(InterruptedException ie) {
                } finally {
                    lock.release();
                }
            }
        }

        for(int i = 0; i < 100; ++i) {
            FairRunner f = new FairRunner(i);
            threads.add(new Thread(f));
        }

        for(Thread t : threads) {
            t.start();
            Thread.sleep(10);
        }

        for(Thread t : threads) {
            t.join();
        }
    }

}

@@ -462,24 +462,68 @@
 .util.List;%0A
+import java.util.concurrent.CountDownLatch;%0A
 import java.
@@ -2286,16 +2286,85 @@
 edValue;
+%0A            final CountDownLatch startLatch = new CountDownLatch(1);
 %0A%0A      
@@ -2518,24 +2518,64 @@
 oid run() %7B%0A
+                startLatch.countDown();%0A
             
@@ -2979,32 +2979,87 @@
    %7D%0A        %7D%0A%0A
+        List%3CFairRunner%3E runners = new ArrayList%3C%3E();%0A%0A
         for(int 
@@ -3127,16 +3127,44 @@
 ner(i);%0A
+            runners.add(f);%0A
         
@@ -3222,48 +3222,124 @@
 for(
-Thread t : threads) %7B%0A            t.star
+int i = 0; i %3C threads.size(); ++i) %7B%0A            threads.get(i).start();%0A            runners.get(i).startLatch.awai
 t();

Add CountDownLatches to ensure order of operation.