package com.github.charithe.kafka;

import com.google.common.util.concurrent.Futures;
import org.junit.jupiter.api.extension.AfterEachCallback;
import org.junit.jupiter.api.extension.BeforeAllCallback;
import org.junit.jupiter.api.extension.BeforeEachCallback;
import org.junit.jupiter.api.extension.ExtensionContext;
import org.junit.jupiter.api.extension.ParameterContext;
import org.junit.jupiter.api.extension.ParameterResolver;

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * <p>{@code KafkaJunitExtension} provides a kafka broker that is started and stopped for each test</p>
 * <p>
 * It is configured by the optional annotation @{@link KafkaJunitExtensionConfig} and provides
 * dependency injection for constructors and methods for the classes {@link KafkaHelper} and {@link EphemeralKafkaBroker}
 * </p>
 * <br>
 * <b>Usage:</b>
 * <pre>
 * {@code @ExtendWith(KafkaJunitExtension.class)
 *  @literal @KafkaJunitExtensionConfig(startupMode = StartupMode.WAIT_FOR_STARTUP)
 *   class MyTestClass {
 *
 *      @literal @BeforeEach
 *       void setUp(KafkaHelper kafkaHelper, EphemeralKafkaBroker broker) {
 *
 *       }
 *  }
 * }
 * </pre>
 */
@KafkaJunitExtensionConfig
public class KafkaJunitExtension implements BeforeAllCallback, AfterEachCallback, BeforeEachCallback, ParameterResolver {

    private static final ExtensionContext.Namespace KAFKA_JUNIT = ExtensionContext.Namespace.create("kafka-junit");
    private static final KafkaJunitExtensionConfig DEFAULT_CONFIG = KafkaJunitExtension.class.getAnnotation(KafkaJunitExtensionConfig.class);

    @Override
    public void beforeAll(ExtensionContext extensionContext) {
        KafkaJunitExtensionConfig kafkaConfig = getKafkaConfig(extensionContext);
        EphemeralKafkaBroker broker = EphemeralKafkaBroker.create(kafkaConfig.kafkaPort(), kafkaConfig.zooKeeperPort());
        extensionContext.getStore(KAFKA_JUNIT).put(EphemeralKafkaBroker.class, broker);
        extensionContext.getStore(KAFKA_JUNIT).put(StartupMode.class, kafkaConfig.startupMode());
        extensionContext.getStore(KAFKA_JUNIT).put(KafkaHelper.class, KafkaHelper.createFor(broker));
    }

    private static KafkaJunitExtensionConfig getKafkaConfig(ExtensionContext extensionContext) {
        return extensionContext.getElement().map(annotatedElement -> {
            if (annotatedElement.isAnnotationPresent(KafkaJunitExtensionConfig.class)) {
                return annotatedElement.getAnnotation(KafkaJunitExtensionConfig.class);
            } else {
                return DEFAULT_CONFIG;
            }
        }).orElse(DEFAULT_CONFIG);
    }

    @Override
    public void beforeEach(ExtensionContext extensionContext) throws Exception {
        CompletableFuture<Void> startFuture = getBroker(extensionContext).start();
        if (getStartupMode(extensionContext) == StartupMode.WAIT_FOR_STARTUP) {
            Futures.getUnchecked(startFuture);
        }
    }

    private static EphemeralKafkaBroker getBroker(ExtensionContext extensionContext) {
        return extensionContext.getStore(KAFKA_JUNIT).get(EphemeralKafkaBroker.class, EphemeralKafkaBroker.class);
    }

    private static StartupMode getStartupMode(ExtensionContext extensionContext) {
        return extensionContext.getStore(KAFKA_JUNIT).get(StartupMode.class, StartupMode.class);
    }

    @Override
    public void afterEach(ExtensionContext extensionContext) {
        try {
            getBroker(extensionContext).stop();
        } catch (ExecutionException | InterruptedException e) {
            throw new AssertionError(e);
        }
    }

    @Override
    public boolean supportsParameter(ParameterContext parameterContext, ExtensionContext extensionContext) {
        return parameterContext.getParameter().getType().equals(KafkaHelper.class) || parameterContext.getParameter().getType().equals(EphemeralKafkaBroker.class);
    }

    @Override
    public Object resolveParameter(ParameterContext parameterContext, ExtensionContext extensionContext) {
        Class<?> parameterType = parameterContext.getParameter().getType();
        return extensionContext.getStore(KAFKA_JUNIT).get(parameterType, parameterType);
    }

}
@@ -530,19 +530,16 @@
 %0A/**%0A * 
-%3Cp%3E
 %7B@code K
@@ -627,22 +627,9 @@
 test
-%3C/p%3E%0A * %3Cp%3E%0A *
+.
  It 
@@ -644,16 +644,19 @@
 gured by
+%0A *
  the opt
@@ -719,19 +719,16 @@
 provides
-%0A *
  depende
@@ -761,16 +761,19 @@
 tors and
+%0A *
  methods
@@ -848,38 +848,18 @@
 %7D%0A *
- %3C/p%3E%0A * %3Cbr%3E
 %0A * 
-%3Cb%3E
 Usage:
-%3C/b%3E
 %0A * 
@@ -871,16 +871,34 @@
 %0A * 
-%7B@code @
+    %3Ccode%3E%0A * %7B@literal @%7D
 Exte
@@ -934,17 +934,17 @@
 ass)%0A * 
- 
+%7B
 @literal
@@ -945,16 +945,17 @@
 iteral @
+%7D
 KafkaJun
@@ -1018,18 +1018,16 @@
 RTUP)%0A *
-  
  class M
@@ -1039,27 +1039,24 @@
 Class %7B%0A
- *%0A
  *      
 @literal
@@ -1047,16 +1047,17 @@
  *      
+%7B
 @literal
@@ -1058,16 +1058,17 @@
 iteral @
+%7D
 BeforeEa
@@ -1166,16 +1166,25 @@
 %7D%0A *
- 
  %7D%0A * 
-%7D
+    %3C/code%3E
 %0A * 
@@ -1253,16 +1253,20 @@
 xtension
+%0A   
  impleme
@@ -1531,16 +1531,24 @@
 CONFIG =
+%0A       
  KafkaJu
@@ -3866,34 +3866,60 @@
 tParameter()
-.getType()
+%0A            .getType()%0A            
 .equals(Ephe
@@ -4242,10 +4242,9 @@
 ;%0A    %7D%0A
-%0A
 %7D

Fix Javadoc warnings