Kafka module cleanup

Author: jknack

docs/asciidoc/modules/kafka.adoc

@@ -0,0 +1,77 @@
+== Kafka
+
+Kafka module for https://kafka.apache.org/documentation
+
+=== Usage
+
+1) Add the dependency:
+
+[dependency, artifactId="jooby-kafka"]
+.
+
+2) Setup properties: 
+
+.application.conf
+[source, properties]
+----
+kafka.producer.bootstrap.servers = "localhost:9092"
+kafka.producer.key.serializer = "org.apache.kafka.common.serialization.StringSerializer"
+kafka.producer.value.serializer = "org.apache.kafka.common.serialization.StringSerializer"
+
+kafka.consumer.bootstrap.servers = "localhost:9092"
+kafka.consumer.key.deserializer = "org.apache.kafka.common.serialization.StringDeserializer"
+kafka.consumer.value.deserializer = "org.apache.kafka.common.serialization.StringDeserializer"
+----
+
+3) Install
+
+.Java
+[source, java, role="primary"]
+----
+import io.jooby.kafka.KafkaModule;
+
+{
+  install(new KafkaModule());                              <1>
+
+  get("/", ctx -> {
+    KafkaProducer producer = require(KafkaProducer.class); <2>
+    // work with producer
+
+    KafkaConsumer consumer = require(KafkaConsumer.class); <3>
+    // work with consumer
+
+    ...
+  });
+}
+----
+
+.Kotlin
+[source, kt, role="secondary"]
+----
+import io.jooby.kafka.KafkaModule
+
+{
+  install(KafkaModule())                                   <1>
+
+  get("/") {
+    val producer = require(KafkaProducer::class);          <2>
+    // work with producer
+
+    val consumer = require(KafkaConsumer::class);          <3>
+    // work with consumer
+    ...
+  }
+}
+----
+
+<1> Install module
+<2> Use KafkaProducer
+<3> Use KafkaConsumer
+
+Optionally, you can install just a consumer:
+
+    install(new KafkaConsumerModule());
+
+Or producer:
+
+    install(new KafkaProducerModule());

docs/asciidoc/modules/modules.adoc

@@ -23,6 +23,7 @@ Available modules are listed next.
    * link:/modules/hikari[HikariCP]: A high-performance JDBC connection pool.
    * link:/modules/hibernate[Hibernate]: Hibernate ORM module.
    * link:/modules/jdbi[Jdbi]: Jdbi module.
+   * link:/modules/kafka[Kafka]: Kafka module.
    * link:/modules/redis[Redis]: Redis module.
 
 === Development Tools

modules/jooby-bom/pom.xml

@@ -63,6 +63,7 @@
   <jsonwebtoken.version>0.11.2</jsonwebtoken.version>
   <jsr305.version>3.0.2</jsr305.version>
   <junit.version>5.7.0</junit.version>
+  <kafka.version>2.6.0</kafka.version>
   <kotlin.version>1.4.10</kotlin.version>
   <kotlinx-coroutines-core.version>1.4.1</kotlinx-coroutines-core.version>
   <lettuce.version>6.0.0.RELEASE</lettuce.version>
@@ -184,6 +185,12 @@
       <version>${jooby.version}</version>
       <type>jar</type>
     </dependency>
+    <dependency>
+      <groupId>io.jooby</groupId>
+      <artifactId>jooby-kafka</artifactId>
+      <version>${jooby.version}</version>
+      <type>jar</type>
+    </dependency>
     <dependency>
       <groupId>io.jooby</groupId>
       <artifactId>jooby-hikari</artifactId>
@@ -982,6 +989,12 @@
       <version>${lettuce.version}</version>
       <type>jar</type>
     </dependency>
+    <dependency>
+      <groupId>org.apache.kafka</groupId>
+      <artifactId>kafka-clients</artifactId>
+      <version>${kafka.version}</version>
+      <type>jar</type>
+    </dependency>
     <dependency>
       <groupId>org.codehaus.mojo</groupId>
       <artifactId>exec-maven-plugin</artifactId>

modules/jooby-kafka/pom.xml

@@ -29,7 +29,6 @@
     <dependency>
       <groupId>org.apache.kafka</groupId>
       <artifactId>kafka-clients</artifactId>
-      <version>2.6.0</version>
     </dependency>
 
     <!-- Test dependencies -->

modules/jooby-kafka/src/main/java/io/jooby/kafka/KafkaConsumerModule.java

@@ -0,0 +1,68 @@
+/**
+ * Jooby https://jooby.io
+ * Apache License Version 2.0 https://jooby.io/LICENSE.txt
+ * Copyright 2014 Edgar Espina
+ */
+package io.jooby.kafka;
+
+import javax.annotation.Nonnull;
+
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+
+import io.jooby.Extension;
+import io.jooby.Jooby;
+
+/**
+ * Kafka consumer module: https://jooby.io/modules/kafka.
+ * <p>
+ * Usage:
+ * </p>
+ *
+ * <pre>{@code
+ * {
+ *   install(new KafkaConsumerModule());
+ *
+ *   get("/", ctx -> {
+ *     KafkaConsumer producer = require(KafkaConsumer.class);
+ *     // work with consumer
+ *   });
+ * }
+ * }</pre>
+ *
+ * application.conf:
+ *  Creates a new kafka consumer module using the <code>kafka.consumer</code> property key.
+ * This key must be present in the application configuration file, like:
+ *
+ * <pre>{@code
+ *  kafka.consumer.bootstrap.servers = "localhost:9092"
+ *  kafka.consumer.group.id = "group A"
+ *  kafka.consumer.key.deserializer = "org.apache.kafka.common.serialization.StringDeserializer"
+ *  kafka.consumer.value.deserializer = "org.apache.kafka.common.serialization.StringDeserializer"
+ * }</pre>
+ *
+ * @author edgar
+ * @since 2.9.3
+ */
+public class KafkaConsumerModule implements Extension {
+  private final String key;
+
+  /**
+   * Creates a new kafka consumer module.
+   *
+   * @param key Kafka key.
+   */
+  public KafkaConsumerModule(@Nonnull String key) {
+    this.key = key;
+  }
+
+  /**
+   * Creates a new kafka consumer module. Uses the default key: <code>kafka.consumer</code>.
+   */
+  public KafkaConsumerModule() {
+    this("kafka.consumer");
+  }
+
+  @Override public void install(@Nonnull Jooby application) {
+    KafkaHelper.install(application, key, KafkaConsumer::new);
+  }
+}

modules/jooby-kafka/src/main/java/io/jooby/kafka/KafkaHelper.java

@@ -0,0 +1,33 @@
+/**
+ * Jooby https://jooby.io
+ * Apache License Version 2.0 https://jooby.io/LICENSE.txt
+ * Copyright 2014 Edgar Espina
+ */
+package io.jooby.kafka;
+
+import java.util.Properties;
+import java.util.function.Function;
+
+import io.jooby.Environment;
+import io.jooby.Jooby;
+import io.jooby.ServiceKey;
+import io.jooby.ServiceRegistry;
+
+class KafkaHelper {
+
+  public static void install(Jooby application, String key,
+      Function<Properties, AutoCloseable> factory) {
+    Environment environment = application.getEnvironment();
+
+    Properties properties = new Properties();
+    properties.putAll(environment.getProperties(key, null));
+
+    AutoCloseable service = factory.apply(properties);
+    ServiceRegistry registry = application.getServices();
+
+    application.onStop(service::close);
+    Class serviceType = service.getClass();
+    registry.putIfAbsent(serviceType, service);
+    registry.put(ServiceKey.key(serviceType, key), service);
+  }
+}