WIP

diff --git a/src/main/java/de/juplo/kafka/sumup/gateway/Application.java b/src/main/java/de/juplo/kafka/sumup/gateway/Application.java
new file mode 100644 (file)
index 0000000..abe0685
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/sumup/gateway/Application.java
@@ -0,0 +1,36 @@
+package de.juplo.kafka.wordcount.recorder;
+
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.serialization.StringSerializer;
+import org.springframework.boot.SpringApplication;
+import org.springframework.boot.autoconfigure.SpringBootApplication;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
+import org.springframework.context.annotation.Bean;
+import org.springframework.util.Assert;
+
+import java.util.Properties;
+
+
+@SpringBootApplication
+@EnableConfigurationProperties(RecorderApplicationProperties.class)
+public class RecorderApplication
+{
+       @Bean(destroyMethod = "close")
+       KafkaProducer<String, String> producer(RecorderApplicationProperties properties)
+       {
+               Assert.hasText(properties.getBootstrapServer(), "juplo.wordcount.recorder.bootstrap-server must be set");
+
+               Properties props = new Properties();
+               props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServer());
+               props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+               props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
+
+               return new KafkaProducer<>(props);
+       }
+
+       public static void main(String[] args)
+       {
+               SpringApplication.run(RecorderApplication.class, args);
+       }
+}

diff --git a/src/main/java/de/juplo/kafka/sumup/gateway/ApplicationController.java b/src/main/java/de/juplo/kafka/sumup/gateway/ApplicationController.java
new file mode 100644 (file)
index 0000000..5fe69ad
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/sumup/gateway/ApplicationController.java
@@ -0,0 +1,80 @@
+package de.juplo.kafka.wordcount.recorder;
+
+import org.apache.kafka.clients.producer.KafkaProducer;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.springframework.http.HttpStatus;
+import org.springframework.http.ResponseEntity;
+import org.springframework.util.MimeTypeUtils;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.context.request.async.DeferredResult;
+
+import javax.validation.constraints.NotEmpty;
+
+
+@RestController
+public class RecorderController
+{
+  private final String topic;
+  private final KafkaProducer<String, String> producer;
+
+
+  public RecorderController(RecorderApplicationProperties properties, KafkaProducer<String,String> producer)
+  {
+    this.topic = properties.getTopic();
+    this.producer = producer;
+  }
+
+  @PostMapping(
+      path = "/{username}",
+      consumes = {
+          MimeTypeUtils.TEXT_PLAIN_VALUE,
+          MimeTypeUtils.APPLICATION_JSON_VALUE
+      },
+      produces = MimeTypeUtils.APPLICATION_JSON_VALUE)
+  DeferredResult<ResponseEntity<RecordingResult>> speak(
+      @PathVariable
+      @NotEmpty(message = "A username must be provided")
+      String username,
+      @RequestBody
+      @NotEmpty(message = "The spoken sentence must not be empty!")
+      String sentence)
+  {
+    DeferredResult<ResponseEntity<RecordingResult>> result = new DeferredResult<>();
+
+    ProducerRecord<String, String> record = new ProducerRecord<>(topic, username, sentence);
+    producer.send(record, (metadata, exception) ->
+    {
+      if (metadata != null)
+      {
+        result.setResult(
+            ResponseEntity.ok(RecordingResult.of(
+                username,
+                sentence,
+                topic,
+                metadata.partition(),
+                metadata.offset(),
+                null,
+                null)));
+      }
+      else
+      {
+        result.setErrorResult(
+            ResponseEntity
+                .internalServerError()
+                .body(RecordingResult.of(
+                    username,
+                    sentence,
+                    topic,
+                    null,
+                    null,
+                    HttpStatus.INTERNAL_SERVER_ERROR.value(),
+                    exception.toString())));
+      }
+    });
+
+    return result;
+  }
+}

diff --git a/src/main/java/de/juplo/kafka/sumup/gateway/ApplicationProperties.java b/src/main/java/de/juplo/kafka/sumup/gateway/ApplicationProperties.java
new file mode 100644 (file)
index 0000000..552ebaf
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/sumup/gateway/ApplicationProperties.java
@@ -0,0 +1,18 @@
+package de.juplo.kafka.wordcount.recorder;
+
+
+import lombok.Getter;
+import lombok.Setter;
+import lombok.ToString;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+
+@ConfigurationProperties("juplo.wordcount.recorder")
+@Getter
+@Setter
+@ToString
+public class RecorderApplicationProperties
+{
+  private String bootstrapServer = "localhost:9092";
+  private String topic = "recordings";
+}

diff --git a/src/main/java/de/juplo/kafka/sumup/gateway/Result.java b/src/main/java/de/juplo/kafka/sumup/gateway/Result.java
new file mode 100644 (file)
index 0000000..939b1d4
--- /dev/null
+++ b/src/main/java/de/juplo/kafka/sumup/gateway/Result.java
@@ -0,0 +1,19 @@
+package de.juplo.kafka.wordcount.recorder;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import lombok.Value;
+
+import static com.fasterxml.jackson.annotation.JsonInclude.Include.NON_NULL;
+
+
+@Value(staticConstructor = "of")
+public class RecordingResult
+{
+  @JsonInclude(NON_NULL) private final String username;
+  @JsonInclude(NON_NULL) private final String sentence;
+  @JsonInclude(NON_NULL) private final String topic;
+  @JsonInclude(NON_NULL) private final Integer partition;
+  @JsonInclude(NON_NULL) private final Long offset;
+  @JsonInclude(NON_NULL) private final Integer status;
+  @JsonInclude(NON_NULL) private final String error;
+}

diff --git a/src/main/java/de/juplo/kafka/wordcount/recorder/RecorderApplication.java b/src/main/java/de/juplo/kafka/wordcount/recorder/RecorderApplication.java
deleted file mode 100644 (file)
index abe0685..0000000
--- a/src/main/java/de/juplo/kafka/wordcount/recorder/RecorderApplication.java
+++ /dev/null
@@ -1,36 +0,0 @@
-package de.juplo.kafka.wordcount.recorder;
-
-import org.apache.kafka.clients.producer.KafkaProducer;
-import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.common.serialization.StringSerializer;
-import org.springframework.boot.SpringApplication;
-import org.springframework.boot.autoconfigure.SpringBootApplication;
-import org.springframework.boot.context.properties.EnableConfigurationProperties;
-import org.springframework.context.annotation.Bean;
-import org.springframework.util.Assert;
-
-import java.util.Properties;
-
-
-@SpringBootApplication
-@EnableConfigurationProperties(RecorderApplicationProperties.class)
-public class RecorderApplication
-{
-       @Bean(destroyMethod = "close")
-       KafkaProducer<String, String> producer(RecorderApplicationProperties properties)
-       {
-               Assert.hasText(properties.getBootstrapServer(), "juplo.wordcount.recorder.bootstrap-server must be set");
-
-               Properties props = new Properties();
-               props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, properties.getBootstrapServer());
-               props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
-               props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
-
-               return new KafkaProducer<>(props);
-       }
-
-       public static void main(String[] args)
-       {
-               SpringApplication.run(RecorderApplication.class, args);
-       }
-}

diff --git a/src/main/java/de/juplo/kafka/wordcount/recorder/RecorderApplicationProperties.java b/src/main/java/de/juplo/kafka/wordcount/recorder/RecorderApplicationProperties.java
deleted file mode 100644 (file)
index 552ebaf..0000000
--- a/src/main/java/de/juplo/kafka/wordcount/recorder/RecorderApplicationProperties.java
+++ /dev/null
@@ -1,18 +0,0 @@
-package de.juplo.kafka.wordcount.recorder;
-
-
-import lombok.Getter;
-import lombok.Setter;
-import lombok.ToString;
-import org.springframework.boot.context.properties.ConfigurationProperties;
-
-
-@ConfigurationProperties("juplo.wordcount.recorder")
-@Getter
-@Setter
-@ToString
-public class RecorderApplicationProperties
-{
-  private String bootstrapServer = "localhost:9092";
-  private String topic = "recordings";
-}

diff --git a/src/main/java/de/juplo/kafka/wordcount/recorder/RecorderController.java b/src/main/java/de/juplo/kafka/wordcount/recorder/RecorderController.java
deleted file mode 100644 (file)
index 5fe69ad..0000000
--- a/src/main/java/de/juplo/kafka/wordcount/recorder/RecorderController.java
+++ /dev/null
@@ -1,80 +0,0 @@
-package de.juplo.kafka.wordcount.recorder;
-
-import org.apache.kafka.clients.producer.KafkaProducer;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.springframework.http.HttpStatus;
-import org.springframework.http.ResponseEntity;
-import org.springframework.util.MimeTypeUtils;
-import org.springframework.web.bind.annotation.PathVariable;
-import org.springframework.web.bind.annotation.PostMapping;
-import org.springframework.web.bind.annotation.RequestBody;
-import org.springframework.web.bind.annotation.RestController;
-import org.springframework.web.context.request.async.DeferredResult;
-
-import javax.validation.constraints.NotEmpty;
-
-
-@RestController
-public class RecorderController
-{
-  private final String topic;
-  private final KafkaProducer<String, String> producer;
-
-
-  public RecorderController(RecorderApplicationProperties properties, KafkaProducer<String,String> producer)
-  {
-    this.topic = properties.getTopic();
-    this.producer = producer;
-  }
-
-  @PostMapping(
-      path = "/{username}",
-      consumes = {
-          MimeTypeUtils.TEXT_PLAIN_VALUE,
-          MimeTypeUtils.APPLICATION_JSON_VALUE
-      },
-      produces = MimeTypeUtils.APPLICATION_JSON_VALUE)
-  DeferredResult<ResponseEntity<RecordingResult>> speak(
-      @PathVariable
-      @NotEmpty(message = "A username must be provided")
-      String username,
-      @RequestBody
-      @NotEmpty(message = "The spoken sentence must not be empty!")
-      String sentence)
-  {
-    DeferredResult<ResponseEntity<RecordingResult>> result = new DeferredResult<>();
-
-    ProducerRecord<String, String> record = new ProducerRecord<>(topic, username, sentence);
-    producer.send(record, (metadata, exception) ->
-    {
-      if (metadata != null)
-      {
-        result.setResult(
-            ResponseEntity.ok(RecordingResult.of(
-                username,
-                sentence,
-                topic,
-                metadata.partition(),
-                metadata.offset(),
-                null,
-                null)));
-      }
-      else
-      {
-        result.setErrorResult(
-            ResponseEntity
-                .internalServerError()
-                .body(RecordingResult.of(
-                    username,
-                    sentence,
-                    topic,
-                    null,
-                    null,
-                    HttpStatus.INTERNAL_SERVER_ERROR.value(),
-                    exception.toString())));
-      }
-    });
-
-    return result;
-  }
-}

diff --git a/src/main/java/de/juplo/kafka/wordcount/recorder/RecordingResult.java b/src/main/java/de/juplo/kafka/wordcount/recorder/RecordingResult.java
deleted file mode 100644 (file)
index 939b1d4..0000000
--- a/src/main/java/de/juplo/kafka/wordcount/recorder/RecordingResult.java
+++ /dev/null
@@ -1,19 +0,0 @@
-package de.juplo.kafka.wordcount.recorder;
-
-import com.fasterxml.jackson.annotation.JsonInclude;
-import lombok.Value;
-
-import static com.fasterxml.jackson.annotation.JsonInclude.Include.NON_NULL;
-
-
-@Value(staticConstructor = "of")
-public class RecordingResult
-{
-  @JsonInclude(NON_NULL) private final String username;
-  @JsonInclude(NON_NULL) private final String sentence;
-  @JsonInclude(NON_NULL) private final String topic;
-  @JsonInclude(NON_NULL) private final Integer partition;
-  @JsonInclude(NON_NULL) private final Long offset;
-  @JsonInclude(NON_NULL) private final Integer status;
-  @JsonInclude(NON_NULL) private final String error;
-}

diff --git a/src/test/java/de/juplo/kafka/sumup/gateway/ApplicationTests.java b/src/test/java/de/juplo/kafka/sumup/gateway/ApplicationTests.java
new file mode 100644 (file)
index 0000000..885a408
--- /dev/null
+++ b/src/test/java/de/juplo/kafka/sumup/gateway/ApplicationTests.java
@@ -0,0 +1,13 @@
+package de.juplo.kafka.wordcount.recorder;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.boot.test.context.SpringBootTest;
+
+@SpringBootTest
+class ApplicationTests
+{
+       @Test
+       void contextLoads()
+       {
+       }
+}

diff --git a/src/test/java/de/juplo/kafka/wordcount/recorder/ApplicationTests.java b/src/test/java/de/juplo/kafka/wordcount/recorder/ApplicationTests.java
deleted file mode 100644 (file)
index 885a408..0000000
--- a/src/test/java/de/juplo/kafka/wordcount/recorder/ApplicationTests.java
+++ /dev/null
@@ -1,13 +0,0 @@
-package de.juplo.kafka.wordcount.recorder;
-
-import org.junit.jupiter.api.Test;
-import org.springframework.boot.test.context.SpringBootTest;
-
-@SpringBootTest
-class ApplicationTests
-{
-       @Test
-       void contextLoads()
-       {
-       }
-}