Added more methods.

Author: sauroter

src/main/java/com/hivemq/extensions/kafka/api/builders/KafkaRecordBuilder.java

@@ -19,6 +19,7 @@
 import com.hivemq.extensions.kafka.api.model.KafkaRecord;
 
 import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
 
 /**
  * @author Christoph Schäbel
@@ -33,16 +34,24 @@ public interface KafkaRecordBuilder {
 
     @NotNull KafkaRecordBuilder header(@NotNull String key, @NotNull String value);
 
+    @NotNull KafkaRecordBuilder header(@NotNull String key, @NotNull String value, @NotNull Charset charset);
+
     @NotNull KafkaRecordBuilder key(@NotNull ByteBuffer key);
 
     @NotNull KafkaRecordBuilder key(@NotNull byte[] key);
 
     @NotNull KafkaRecordBuilder key(@NotNull String key);
 
+    @NotNull KafkaRecordBuilder key(@NotNull String key, @NotNull Charset charset);
+
     @NotNull KafkaRecordBuilder value(@NotNull ByteBuffer value);
 
     @NotNull KafkaRecordBuilder value(@NotNull byte[] value);
 
+    @NotNull KafkaRecordBuilder value(@NotNull String value);
+
+    @NotNull KafkaRecordBuilder value(@NotNull String value, @NotNull Charset charset);
+
     @NotNull KafkaRecordBuilder timestamp(long timestamp);
 
     @NotNull KafkaRecordBuilder partition(int partition);

src/main/java/com/hivemq/extensions/kafka/api/model/KafkaCluster.java

@@ -15,14 +15,16 @@
  */
 package com.hivemq.extensions.kafka.api.model;
 
+import com.hivemq.extension.sdk.api.annotations.NotNull;
+
 /**
  * @author Christoph Schäbel
  */
 public
 interface KafkaCluster {
 
-    String getId();
+    @NotNull String getId();
 
-    String getBootstrapServers();
+    @NotNull String getBootstrapServers();
 
 }

src/main/java/com/hivemq/extensions/kafka/api/model/KafkaHeader.java

@@ -15,19 +15,24 @@
  */
 package com.hivemq.extensions.kafka.api.model;
 
+import com.hivemq.extension.sdk.api.annotations.NotNull;
+
 import java.nio.ByteBuffer;
+import java.nio.charset.Charset;
 
 /**
  * @author Christoph Schäbel
  */
 public interface KafkaHeader {
 
-    String getKey();
+    @NotNull String getKey();
+
+    @NotNull ByteBuffer getValue();
 
-    ByteBuffer getValue();
+    @NotNull String getValueAsString();
 
-    String getValueAsString();
+    @NotNull String getValueAsString(@NotNull Charset charset);
 
-    byte[] getValueAsByteArray();
+    @NotNull byte[] getValueAsByteArray();
 
 }

src/main/java/com/hivemq/extensions/kafka/api/model/KafkaHeaders.java

@@ -26,7 +26,7 @@
  */
 public interface KafkaHeaders {
 
-    List<KafkaHeader> asList();
+    @NotNull List<KafkaHeader> asList();
 
     /**
      * @param name The name of the user property to get.

src/main/java/com/hivemq/extensions/kafka/api/model/KafkaRecord.java

@@ -33,11 +33,11 @@ public interface KafkaRecord {
 
     @NotNull Optional<byte[]> getKeyAsByteArray();
 
-    @NotNull ByteBuffer getValue();
+    @NotNull Optional<ByteBuffer> getValue();
 
-    @NotNull byte[] getValueAsByteArray();
+    @NotNull Optional<byte[]> getValueAsByteArray();
 
-    @NotNull Long getTimestamp();
+    @NotNull Optional<Long> getTimestamp();
 
     @NotNull Optional<Integer> getPartition();
 

src/main/java/com/hivemq/extensions/kafka/api/services/KafkaTopicService.java

@@ -0,0 +1,20 @@
+package com.hivemq.extensions.kafka.api.services;
+
+import com.hivemq.extension.sdk.api.annotations.NotNull;
+
+import java.util.List;
+
+public interface KafkaTopicService {
+
+    @NotNull KafkaTopicState getKafkaTopicState(@NotNull String topic);
+
+    @NotNull List<@NotNull KafkaTopicState> getKafkaTopicStates(@NotNull List<@NotNull String> topics);
+
+    @NotNull KafkaTopicState createKafkaTopic(@NotNull String topic);
+
+    @NotNull List<@NotNull KafkaTopicState> createKafkaTopics(@NotNull List<@NotNull String> topics);
+
+    enum KafkaTopicState {
+        FAILURE, EXISTS, CREATED, MISSING,
+    }
+}

src/main/java/com/hivemq/extensions/kafka/api/transformers/Transformer.java

@@ -20,7 +20,7 @@
 /**
  * @author Christoph Schäbel
  */
-public interface Transformer {
-    default void init(@NotNull TransformerInitInput input) {
+public interface Transformer<I extends TransformerInitInput> {
+    default void init(@NotNull I input) {
     }
 }

src/main/java/com/hivemq/extensions/kafka/api/transformers/mqtttokafka/MqttToKafkaInitInput.java

@@ -0,0 +1,13 @@
+package com.hivemq.extensions.kafka.api.transformers.mqtttokafka;
+
+import com.hivemq.extension.sdk.api.annotations.NotNull;
+import com.hivemq.extensions.kafka.api.services.KafkaTopicService;
+import com.hivemq.extensions.kafka.api.transformers.TransformerInitInput;
+
+/**
+ * @author Georg Held
+ */
+public interface MqttToKafkaInitInput extends TransformerInitInput {
+
+    @NotNull KafkaTopicService getKafkaTopicService();
+}

src/main/java/com/hivemq/extensions/kafka/api/transformers/mqtttokafka/MqttToKafkaInput.java

@@ -18,12 +18,15 @@
 import com.hivemq.extension.sdk.api.annotations.NotNull;
 import com.hivemq.extension.sdk.api.packets.publish.PublishPacket;
 import com.hivemq.extensions.kafka.api.model.KafkaCluster;
+import com.hivemq.extensions.kafka.api.services.KafkaTopicService;
 
 /**
  * @author Christoph Schäbel
  */
 public interface MqttToKafkaInput {
 
+    @NotNull KafkaTopicService getKafkaTopicService();
+
     @NotNull PublishPacket getPublishPacket();
 
     @NotNull KafkaCluster getKafkaCluster();

src/main/java/com/hivemq/extensions/kafka/api/transformers/mqtttokafka/MqttToKafkaOutput.java

@@ -28,5 +28,5 @@ public interface MqttToKafkaOutput {
 
     @NotNull KafkaRecordBuilder newKafkaRecordBuilder();
 
-    void setKafkaRecords(@NotNull List<KafkaRecord> kafkaRecord);
+    void setKafkaRecords(@NotNull List<@NotNull KafkaRecord> kafkaRecord);
 }