package com.hubspot.blazar.queue;

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

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;

import org.junit.After;
import org.junit.Before;
import org.junit.Test;

import com.google.common.eventbus.EventBus;
import com.google.common.eventbus.Subscribe;
import com.google.common.util.concurrent.Uninterruptibles;
import com.google.inject.Guice;
import com.hubspot.blazar.base.BranchSetting;
import com.hubspot.blazar.base.BuildTrigger;
import com.hubspot.blazar.base.BuildTrigger.Type;
import com.hubspot.blazar.data.BlazarDataModule;
import com.hubspot.blazar.guice.BlazarEventBusModule;
import com.hubspot.blazar.guice.BlazarQueueProcessorModule;
import com.hubspot.blazar.test.base.service.BlazarTestBase;
import com.hubspot.blazar.test.base.service.BlazarTestModule;

public class QueueProcessorTest extends BlazarTestBase {
  private EventBus eventBus;
  private List<Object> received;

  @Before
  public void setup() throws Exception {
    synchronized (injector) {
      if (injector.get() == null) {
        injector.set(
            Guice.createInjector(
                new BlazarTestModule(),
                new BlazarDataModule(),
                new BlazarEventBusModule(),
                new BlazarQueueProcessorModule()
            )
        );
        runSql("schema.sql");
        getFromGuice(EventBus.class).register(new Object() {

          @Subscribe
          public void handleEvent(Object event) {
            received.add(event);
          }
        });
      }
    }

    eventBus = getFromGuice(EventBus.class);
    received = new ArrayList<>();

    QueueProcessor queueProcessor = getFromGuice(QueueProcessor.class);
    queueProcessor.start();
    queueProcessor.isLeader();
  }

  @After
  public void cleanup() throws Exception {
    QueueProcessor queueProcessor = getFromGuice(QueueProcessor.class);
    queueProcessor.stop();
    queueProcessor.notLeader();

    runSql("schema.sql");
  }

  @Test
  public void itProcessesEvents() {
    BuildTrigger event = new BuildTrigger(Type.PUSH, "abc");
    eventBus.post(event);

    waitForEvent();
    assertThat(received).containsExactly(event);
  }

  @Test
  public void itRetriesFailedEvents() {
    eventBus.register(new Object() {
      AtomicBoolean threw = new AtomicBoolean();

      @Subscribe
      public void handleEvent(Object event) {
        if (threw.compareAndSet(false, true)) {
          throw new RuntimeException();
        }
      }
    });

    BuildTrigger event = new BuildTrigger(Type.PUSH, "abc");
    eventBus.post(event);

    waitForEvent();
    waitForEvent(21200, TimeUnit.MILLISECONDS);
    assertThat(received).containsExactly(event, event);
  }

  private void waitForEvent() {
    waitForEvent(1200, TimeUnit.MILLISECONDS);
  }

  private void waitForEvent(long amount, TimeUnit unit) {
    int events = received.size();
    long start = System.nanoTime();
    long stop = start + unit.toNanos(amount);

    while (received.size() == events && System.nanoTime() < stop) {
      Uninterruptibles.sleepUninterruptibly(25, TimeUnit.MILLISECONDS);
    }

    if (received.size() == events) {
      throw new RuntimeException(new TimeoutException("No event received"));
    }
  }
}

@@ -134,24 +134,68 @@
 .util.List;%0A
+import java.util.concurrent.CountDownLatch;%0A
 import java.
@@ -574,54 +574,8 @@
 ce;%0A
-import com.hubspot.blazar.base.BranchSetting;%0A
 impo
@@ -2752,17 +2752,17 @@
 orEvent(
-2
+1
 1200, Ti
@@ -2836,32 +2836,744 @@
 t, event);%0A  %7D%0A%0A
+  @Test%0A  public void itDoesntDoubleProcessItems() %7B%0A    CountDownLatch latch = new CountDownLatch(1);%0A%0A    eventBus.register(new Object() %7B%0A      AtomicBoolean slept = new AtomicBoolean();%0A%0A      @Subscribe%0A      public void handleEvent(Object event) throws InterruptedException %7B%0A        if (slept.compareAndSet(false, true)) %7B%0A          Uninterruptibles.sleepUninterruptibly(3, TimeUnit.SECONDS);%0A          latch.countDown();%0A        %7D%0A      %7D%0A    %7D);%0A%0A    BuildTrigger event = new BuildTrigger(Type.PUSH, %22abc%22);%0A    eventBus.post(event);%0A%0A    Uninterruptibles.awaitUninterruptibly(latch);%0A    Uninterruptibles.sleepUninterruptibly(2, TimeUnit.SECONDS);%0A    assertThat(received).containsExactly(event);%0A  %7D%0A%0A
   private void w

Add another test