init

Author: kindywu

modules/jooby-kafka/pom.xml

@@ -0,0 +1,49 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd">
+
+  <parent>
+    <groupId>io.jooby</groupId>
+    <artifactId>modules</artifactId>
+    <version>2.9.3-SNAPSHOT</version>
+  </parent>
+
+  <modelVersion>4.0.0</modelVersion>
+  <artifactId>jooby-kafka</artifactId>
+
+  <dependencies>
+    <dependency>
+      <groupId>com.google.code.findbugs</groupId>
+      <artifactId>jsr305</artifactId>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>io.jooby</groupId>
+      <artifactId>jooby</artifactId>
+      <version>${jooby.version}</version>
+    </dependency>
+
+    <!--add kafka dependencies-->
+    <dependency>
+      <groupId>org.apache.kafka</groupId>
+      <artifactId>kafka-clients</artifactId>
+      <version>2.6.0</version>
+    </dependency>
+
+    <!-- Test dependencies -->
+    <dependency>
+      <groupId>org.junit.jupiter</groupId>
+      <artifactId>junit-jupiter-engine</artifactId>
+      <scope>test</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.jacoco</groupId>
+      <artifactId>org.jacoco.agent</artifactId>
+      <classifier>runtime</classifier>
+      <scope>test</scope>
+    </dependency>
+  </dependencies>
+</project>

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

@@ -0,0 +1,119 @@
+/**
+ * Jooby https://jooby.io
+ * Apache License Version 2.0 https://jooby.io/LICENSE.txt
+ * Copyright 2014 Edgar Espina
+ */
+package io.jooby.kafka;
+
+import com.typesafe.config.Config;
+import io.jooby.Extension;
+import io.jooby.Jooby;
+import io.jooby.ServiceRegistry;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.producer.KafkaProducer;
+
+import javax.annotation.Nonnull;
+import java.util.Properties;
+
+/**
+ * Redis module: https://jooby.io/modules/redis.
+ * <p>
+ * Usage:
+ *
+ * <pre>{@code
+ * {
+ *   install(new RedisModule());
+ *
+ *   get("/", ctx -> {
+ *     StatefulRedisConnection redis = require(StatefulRedisConnection.class);
+ *     // work with redis
+ *   });
+ * }
+ * }</pre>
+ * <p>
+ * application.conf:
+ *
+ * <pre>
+ *   redis = "redis://localhost:6379"
+ * </pre>
+ * <p>
+ * Module is built on top of <a href="https://lettuce.io/">lettuce</a>. Once installed you are able
+ * to work with:
+ *
+ * <ul>
+ *   <li>io.lettuce.core.RedisClient</li>
+ *   <li>io.lettuce.core.api.StatefulRedisConnection</li>
+ *   <li>io.lettuce.core.pubsub.StatefulRedisPubSubConnection</li>
+ * </ul>
+ * <p>
+ * Alternative you can pass a redis URI:
+ * <pre>
+ *   install(new RedisModule("redis://localhost:6379"));
+ * </pre>
+ *
+ * @author edgar
+ * @since 2.8.5
+ */
+public class KafkaModule implements Extension {
+  Properties producerProps;
+  Properties consumerProps;
+
+  /**
+   * Creates a new kafka module. Value must be:
+   * - Valid redis URI; or
+   * - Property name
+   *
+   * @param producerProps kafka producer properties.
+   * @param consumerProps kafka consumer properties.
+   */
+  public KafkaModule(@Nonnull Properties producerProps, @Nonnull Properties consumerProps) {
+    this.producerProps = producerProps;
+    this.consumerProps = consumerProps;
+  }
+
+  /**
+   * Create a new redis module. The application configuration file must have a redis property, like:
+   * <pre>
+   *   redis = "redis://localhost:6379"
+   * </pre>
+   */
+  public KafkaModule() {
+  }
+
+  @Override
+  public void install(@Nonnull Jooby application) {
+    Config config = application.getConfig();
+
+    if (this.producerProps == null) {
+      this.producerProps = properties(config, "kafka.producer");
+    }
+
+    if (this.consumerProps == null) {
+      this.consumerProps = properties(config, "kafka.consumer");
+    }
+
+    ServiceRegistry registry = application.getServices();
+
+    if (this.producerProps != null) {
+      KafkaProducer producer = new KafkaProducer(producerProps);
+      application.onStop(producer::close);
+      registry.putIfAbsent(KafkaProducer.class, producer);
+    }
+
+    if (this.consumerProps != null) {
+      KafkaConsumer consumer = new KafkaConsumer(consumerProps);
+      application.onStop(consumer::close);
+      registry.putIfAbsent(KafkaConsumer.class, consumer);
+    }
+  }
+
+  private static Properties properties(final Config config, final String key) {
+    Properties props = new Properties();
+    if (config.hasPath(key)) {
+      // dump
+      config.getConfig(key).entrySet().forEach(
+          e -> props.setProperty(e.getKey(), e.getValue().unwrapped().toString()));
+    }
+    return props;
+  }
+}