Json kafka serializer & deserializer

resolves #241
Tinkoff · Aug 7, 2023 · d40ca77 · d40ca77
1 parent 5ed088b
commit d40ca77
Show file tree

Hide file tree

Showing 3 changed files with 60 additions and 0 deletions.
diff --git a/json/json-module/src/main/java/ru/tinkoff/kora/json/module/JsonModule.java b/json/json-module/src/main/java/ru/tinkoff/kora/json/module/JsonModule.java
@@ -10,6 +10,8 @@
 import ru.tinkoff.kora.json.module.http.server.JsonReaderHttpServerRequestMapper;
 import ru.tinkoff.kora.json.module.http.server.JsonStringParameterReader;
 import ru.tinkoff.kora.json.module.http.server.JsonWriterHttpServerResponseMapper;
+import ru.tinkoff.kora.json.module.kafka.JsonKafkaDeserializer;
+import ru.tinkoff.kora.json.module.kafka.JsonKafkaSerializer;
 
 public interface JsonModule extends JsonCommonModule {
     @Json
@@ -41,4 +43,14 @@ default <T> JsonStringParameterConverter<T> jsonStringParameterConverter(JsonWri
     default <T> JsonStringParameterReader<T> jsonStringParameterReader(JsonReader<T> reader) {
         return new JsonStringParameterReader<>(reader);
     }
+
+    @Json
+    default <T> JsonKafkaDeserializer<T> jsonKafkaDeserializer(JsonReader<T> reader) {
+        return new JsonKafkaDeserializer<>(reader);
+    }
+
+    @Json
+    default <T> JsonKafkaSerializer<T> jsonKafkaSerializer(JsonWriter<T> writer) {
+        return new JsonKafkaSerializer<>(writer);
+    }
 }
diff --git a/json/json-module/src/main/java/ru/tinkoff/kora/json/module/kafka/JsonKafkaDeserializer.java b/json/json-module/src/main/java/ru/tinkoff/kora/json/module/kafka/JsonKafkaDeserializer.java
@@ -0,0 +1,24 @@
+package ru.tinkoff.kora.json.module.kafka;
+
+import org.apache.kafka.common.errors.SerializationException;
+import org.apache.kafka.common.serialization.Deserializer;
+import ru.tinkoff.kora.json.common.JsonReader;
+
+import java.io.IOException;
+
+public final class JsonKafkaDeserializer<T> implements Deserializer<T> {
+    private final JsonReader<T> reader;
+
+    public JsonKafkaDeserializer(JsonReader<T> reader) {
+        this.reader = reader;
+    }
+
+    @Override
+    public T deserialize(String topic, byte[] data) {
+        try {
+            return this.reader.read(data);
+        } catch (IOException e) {
+            throw new SerializationException("Unable to deserialize from json", e);
+        }
+    }
+}
diff --git a/json/json-module/src/main/java/ru/tinkoff/kora/json/module/kafka/JsonKafkaSerializer.java b/json/json-module/src/main/java/ru/tinkoff/kora/json/module/kafka/JsonKafkaSerializer.java
@@ -0,0 +1,24 @@
+package ru.tinkoff.kora.json.module.kafka;
+
+import org.apache.kafka.common.errors.SerializationException;
+import org.apache.kafka.common.serialization.Serializer;
+import ru.tinkoff.kora.json.common.JsonWriter;
+
+import java.io.IOException;
+
+public final class JsonKafkaSerializer<T> implements Serializer<T> {
+    private final JsonWriter<T> writer;
+
+    public JsonKafkaSerializer(JsonWriter<T> writer) {
+        this.writer = writer;
+    }
+
+    @Override
+    public byte[] serialize(String topic, T data) {
+        try {
+            return this.writer.toByteArray(data);
+        } catch (IOException e) {
+            throw new SerializationException("Unable to serialize into json", e);
+        }
+    }
+}