Change code due to CR

Author: ekuvardin

topic/src/main/java/tech/ydb/topic/TopicClient.java

@@ -8,8 +8,8 @@
 import tech.ydb.core.Result;
 import tech.ydb.core.Status;
 import tech.ydb.core.grpc.GrpcTransport;
+import tech.ydb.topic.description.Codec;
 import tech.ydb.topic.description.ConsumerDescription;
-import tech.ydb.topic.description.CustomTopicCodec;
 import tech.ydb.topic.description.TopicDescription;
 import tech.ydb.topic.impl.GrpcTopicRpc;
 import tech.ydb.topic.impl.TopicClientImpl;
@@ -168,20 +168,11 @@ default CompletableFuture<Result<ConsumerDescription>> describeConsumer(String p
     /**
      * Register custom codec implementation to TopicClient     *
      *
-     * @param codec - codec identifier (must be more than 10000)
-     * @param customTopicCodec - custom implementation
+     * @param codec - custom implementation
      */
-    void registerCodec(int codec, CustomTopicCodec customTopicCodec);
+    void registerCodec(Codec codec);
 
 
-    /**
-     * Unregister custom codec implementation
-     *
-     * @param codec - codec identifier (must be more than 10000)
-     * @return implementation was before
-     */
-    CustomTopicCodec unregisterCodec(int codec);
-
     /**
      * BUILDER
      */

topic/src/main/java/tech/ydb/topic/description/Codec.java

@@ -1,28 +1,83 @@
 package tech.ydb.topic.description;
 
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.OutputStream;
+
 /**
- * @author Nikolay Perfilov
+
+ *
+ * Interface for custom codec implementation.
+ * <p>
  *
- * List of reserved codecs
+ * You can use custom codec as below
+ * 1. Implement interface methods
+ *    Specify getId which return value more than 10000. This value identify codec across others
+ * 2. Use code below to write data
+ *   Codec codecImpl = ....
+ *   Topic client = TopicClient.newClient(ydbTransport).build();
+ * <p>
+ *   client.registerCodec(codecImpl);
+ *   WriterSettings settings = WriterSettings.newBuilder()
+ *                 .setTopicPath(topicName)
+ *                 .setCodec(codecId)
+ *                 .build();
+ * <p>
+ *    SyncWriter writer = client.createSyncWriter(settings);
+ * <p>
+ * 3. Use to read data. Codec should be registered in {@link CodecRegistry}
+ *   Codec codecImpl = ....
+ *   Topic client = TopicClient.newClient(ydbTransport).build();
+ * <p>
+ *   ReaderSettings readerSettings = ReaderSettings.newBuilder()
+ *                  .addTopic(TopicReadSettings.newBuilder().setPath(topicName).build())
+ *                  .setConsumerName(TEST_CONSUMER1)
+ *                  .build();
+ * <p>
+ *   SyncReader reader = client.createSyncReader(readerSettings);
+ *
+ * @author Nikolay Perfilov
  */
-public class Codec {
-    public static final int RAW = 1;
-    public static final int GZIP = 2;
-    public static final int LZOP = 3;
-    public static final int ZSTD = 4;
-    public static final int CUSTOM = 10000;
-
-    private static final Codec INSTANCE = new Codec();
+public interface Codec {
+    int RAW = 1;
+    int GZIP = 2;
+    int LZOP = 3;
+    int ZSTD = 4;
+    int CUSTOM = 10000;
 
-    private Codec() {
+    /**
+     * Check is codec is reserved
+     *
+     * @param codec codec id
+     * @return true - codec id is reserved; false - elsewhere
+     */
+    default boolean isReserved(int codec) {
+        return codec <= CUSTOM;
     }
 
-    public static Codec getInstance() {
-        return INSTANCE;
-    }
+    /**
+     * Get codec identifier
+     * @return codec identifier
+     */
+    int getId();
 
-    public boolean isReserved(int codec) {
-        return codec <= CUSTOM;
-    }
+    /**
+     * Decode data
+     *
+     * @param byteArrayInputStream input stream
+     * @return output stream
+     * @throws IOException throws when error occurs
+     */
+
+    InputStream decode(InputStream byteArrayInputStream) throws IOException;
+
+    /**
+     * Encode data
+     *
+     * @param byteArrayOutputStream output stream
+     * @return output stream
+     * @throws IOException throws when error occurs
+     */
+    OutputStream encode(OutputStream byteArrayOutputStream) throws IOException;
 
 }

topic/src/main/java/tech/ydb/topic/description/CodecRegistry.java

@@ -1,32 +1,67 @@
 package tech.ydb.topic.description;
 
+import java.util.HashMap;
+import java.util.Map;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import tech.ydb.topic.impl.GzipCodec;
+import tech.ydb.topic.impl.LzopCodec;
+import tech.ydb.topic.impl.RawCodec;
+import tech.ydb.topic.impl.ZstdCodec;
+
 /**
- * Interface for register custom codec
+ * Register for custom topic codec. Local to TopicClient
  *
  * @author Evgeny Kuvardin
  **/
-public interface CodecRegistry {
+public class CodecRegistry {
+
+    private static final Logger logger = LoggerFactory.getLogger(CodecRegistry.class);
+
+    final Map<Integer, Codec> customCodecMap;
+
+    public CodecRegistry() {
+        customCodecMap = new HashMap<>();
+        customCodecMap.put(Codec.RAW, RawCodec.getInstance());
+        customCodecMap.put(Codec.GZIP, GzipCodec.getInstance());
+        customCodecMap.put(Codec.LZOP, LzopCodec.getInstance());
+        customCodecMap.put(Codec.ZSTD, ZstdCodec.getInstance());
+    }
 
     /**
      * Register codec implementation
-     * @param codec codec identifier
-     * @param customTopicCodec codec implementation
+     * @param codec codec implementation
      * @return previous implementation with associated codec
      */
-    CustomTopicCodec registerCustomCodec(int codec, CustomTopicCodec customTopicCodec);
+    public Codec registerCodec(Codec codec) {
+        assert codec != null;
+        int codecId = codec.getId();
 
-    /**
-     * Unregister codec implementation
-     * @param codec codec identifier
-     * @return previous implementation with associated codec
-     */
-    CustomTopicCodec unregisterCustomCodec(int codec);
+        if (codec.isReserved(codecId)) {
+            throw new RuntimeException(
+                    "Create custom codec for reserved code not allowed: " + codec + " .Use code more than 10000");
+        }
+        Codec result = customCodecMap.put(codecId, codec);
+
+        if (result != null) {
+            logger.info(
+                    "Replace codec which have already associated with this id. CodecId: {} Codec: {}",
+                    codecId,
+                    result);
+        }
+
+        return result;
+    }
 
     /**
      * Get codec implementation by associated id
-     * @param codec codec identifier
+     * @param codecId codec identifier
      * @return codec implementation
      */
-    CustomTopicCodec getCustomCodec(int codec);
+    public Codec getCodec(int codecId) {
+        return customCodecMap.get(codecId);
+    }
 
 }