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.CountDownLatch;
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.BuildTrigger;
import com.hubspot.blazar.base.BuildTrigger.Type;
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 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(11200, TimeUnit.MILLISECONDS);
    assertThat(received).containsExactly(event, event);
  }

  @Test
  public void itDoesntDoubleProcessItems() {
    CountDownLatch latch = new CountDownLatch(1);

    eventBus.register(new Object() {
      AtomicBoolean slept = new AtomicBoolean();

      @Subscribe
      public void handleEvent(Object event) throws InterruptedException {
        if (slept.compareAndSet(false, true)) {
          Uninterruptibles.sleepUninterruptibly(3, TimeUnit.SECONDS);
          latch.countDown();
        }
      }
    });

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

    Uninterruptibles.awaitUninterruptibly(latch);
    Uninterruptibles.sleepUninterruptibly(2, TimeUnit.SECONDS);
    assertThat(received).containsExactly(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"));
    }
  }
}

@@ -317,16 +317,78 @@
 olean;%0A%0A
+import org.jukito.JukitoRunner;%0Aimport org.jukito.UseModules;%0A
 import o
@@ -450,16 +450,49 @@
 it.Test;
+%0Aimport org.junit.runner.RunWith;
 %0A%0Aimport
@@ -658,21 +658,22 @@
 .inject.
-Guice
+Inject
 ;%0Aimport
@@ -879,68 +879,8 @@
 le;%0A
-import com.hubspot.blazar.test.base.service.BlazarTestBase;%0A
 impo
@@ -941,552 +941,363 @@
 le;%0A
-%0Apublic class QueueProcessorTest extends B
+import com.hubspot.b
 lazar
-TestBase %7B%0A  private EventBus eventBus;%0A  private List%3CObject%3E received;%0A%0A  @Before%0A  public void setup() throws Exception %7B%0A    synchronized (injector) %7B%0A      if (injector.get() == null) %7B%0A        injector.set(%0A            Guice.createInjector(%0A                new BlazarTestModule(),%0A                new BlazarEventBusModule(),%0A                new BlazarQueueProcessorModule()%0A            )%0A        );%0A        runSql(%22schema.sql%22);%0A        getFromGuice(EventBus.class).register(new Object
+.test.base.service.DatabaseBackedTest;%0A%0A@RunWith(JukitoRunner.class)%0A@UseModules(%7BBlazarQueueProcessorModule.class, BlazarEventBusModule.class, BlazarTestModule.class%7D)%0Apublic class QueueProcessorTest extends DatabaseBackedTest %7B%0A  private List%3CObject%3E received = new ArrayList%3C%3E();%0A  private EventBus eventBus = new EventBus
 () %7B%0A
-%0A
     
@@ -1411,187 +1411,97 @@
    %7D
-%0A        %7D);%0A      %7D%0A    %7D%0A%0A    eventBus = getFromGuice(EventBus.class);%0A    received = new ArrayList%3C%3E();%0A%0A    QueueProcessor queueProcessor = getFromGuice(QueueProcessor.class);
+%7D;%0A  @Inject%0A  private QueueProcessor queueProcessor;%0A%0A  @Before%0A  public void before() %7B
 %0A   
@@ -1617,80 +1617,8 @@
 n %7B%0A
-    QueueProcessor queueProcessor = getFromGuice(QueueProcessor.class);%0A
     
@@ -1675,35 +1675,8 @@
 r();
-%0A%0A    runSql(%22schema.sql%22);
 %0A  %7D

fix up this test to use jukito