provide codecs via singleton map (#3070)

Author: ildar-khisambeev

ydb/public/sdk/cpp/client/ydb_persqueue_core/impl/persqueue.cpp

@@ -11,6 +11,15 @@
 
 namespace NYdb::NPersQueue {
 
+class TCommonCodecsProvider {
+public:
+    TCommonCodecsProvider() {
+        NYdb::NTopic::TCodecMap::GetTheCodecMap().Set((ui32)NYdb::NPersQueue::ECodec::GZIP, MakeHolder<NYdb::NTopic::TGzipCodec>());
+        NYdb::NTopic::TCodecMap::GetTheCodecMap().Set((ui32)NYdb::NPersQueue::ECodec::ZSTD, MakeHolder<NYdb::NTopic::TZstdCodec>());
+    }
+};
+TCommonCodecsProvider COMMON_CODECS_PROVIDER;
+
 const TVector<ECodec>& GetDefaultCodecs() {
     static const TVector<ECodec> codecs = {};
     return codecs;

ydb/public/sdk/cpp/client/ydb_persqueue_core/impl/read_session.ipp

@@ -2569,23 +2569,19 @@ void TDataDecompressionInfo<UseMigrationProtocol>::TDecompressionTask::operator(
                         && data.codec() != Ydb::PersQueue::V1::CODEC_RAW
                         && data.codec() != Ydb::PersQueue::V1::CODEC_UNSPECIFIED
                     ) {
-                        if (auto session = Parent->CbContext->LockShared()) {
-                            const NYdb::NTopic::ICodec* codecImpl = session->GetCodecImplOrThrow(static_cast<ECodec>(data.codec()));
-                            TString decompressed = codecImpl->Decompress(data.data());
-                            data.set_data(decompressed);
-                            data.set_codec(Ydb::PersQueue::V1::CODEC_RAW);
-                        }
+                        const NYdb::NTopic::ICodec* codecImpl = NYdb::NTopic::TCodecMap::GetTheCodecMap().GetOrThrow(static_cast<ui32>(data.codec()));
+                        TString decompressed = codecImpl->Decompress(data.data());
+                        data.set_data(decompressed);
+                        data.set_codec(Ydb::PersQueue::V1::CODEC_RAW);
                     }
                 } else {
                     if (Parent->DoDecompress
                         && static_cast<Ydb::Topic::Codec>(batch.codec()) != Ydb::Topic::CODEC_RAW
                         && static_cast<Ydb::Topic::Codec>(batch.codec()) != Ydb::Topic::CODEC_UNSPECIFIED
                     ) {
-                        if (auto session = Parent->CbContext->LockShared()) {
-                            const NYdb::NTopic::ICodec* codecImpl = session->GetCodecImplOrThrow(static_cast<NTopic::ECodec>(batch.codec()));
-                            TString decompressed = codecImpl->Decompress(data.data());
-                            data.set_data(decompressed);
-                        }
+                        const NYdb::NTopic::ICodec* codecImpl = NYdb::NTopic::TCodecMap::GetTheCodecMap().GetOrThrow(static_cast<ui32>(batch.codec()));
+                        TString decompressed = codecImpl->Decompress(data.data());
+                        data.set_data(decompressed);
                     }
                 }
 

ydb/public/sdk/cpp/client/ydb_persqueue_core/impl/write_session_impl.cpp

@@ -892,7 +892,8 @@ TMemoryUsageChange TWriteSessionImpl::OnMemoryUsageChangedImpl(i64 diff) {
 
 TBuffer CompressBuffer(std::shared_ptr<TPersQueueClient::TImpl> client, TVector<TStringBuf>& data, ECodec codec, i32 level) {
     TBuffer result;
-    THolder<IOutputStream> coder = client->GetCodecImplOrThrow(codec)->CreateCoder(result, level);
+    Y_UNUSED(client);
+    THolder<IOutputStream> coder = NYdb::NTopic::TCodecMap::GetTheCodecMap().GetOrThrow((ui32)codec)->CreateCoder(result, level);
     for (auto& buffer : data) {
         coder->Write(buffer.data(), buffer.size());
     }

ydb/public/sdk/cpp/client/ydb_topic/codecs/codecs.h

@@ -8,8 +8,9 @@
 #include "util/stream/str.h"
 #include <util/stream/zlib.h>
 #include <util/stream/output.h>
+#include <util/system/spinlock.h>
 
-#define CODECS_ALREADY_DEFINED
+#include <unordered_map>
 
 namespace NYdb::NTopic {
 
@@ -78,4 +79,42 @@ class TUnsupportedCodec final : public ICodec {
     }
 };
 
+class TCodecMap {
+public:
+    static TCodecMap& GetTheCodecMap() {
+        static TCodecMap instance;
+        return instance;
+    }
+
+    void Set(ui32 codecId, THolder<ICodec>&& codecImpl) {
+        with_lock(Lock) {
+            Codecs[codecId] = std::move(codecImpl);
+        }
+    }
+
+    const ICodec* GetOrThrow(ui32 codecId) const {
+        with_lock(Lock) {
+            if (!Codecs.contains(codecId)) {
+                throw yexception() << "codec with id " << ui32(codecId) << " not provided";
+            }
+            return Codecs.at(codecId).Get();
+        }
+    }
+
+
+    TCodecMap(const TCodecMap&) = delete;
+    TCodecMap(TCodecMap&&) = delete;
+    TCodecMap& operator=(const TCodecMap&) = delete;
+    TCodecMap& operator=(TCodecMap&&) = delete;
+
+private:
+    TCodecMap() = default;
+
+private:
+    std::unordered_map<ui32, THolder<ICodec>> Codecs;
+    TAdaptiveLock Lock;
+};
+
+#define CODEC_MAP_ALREADY_PROVIDED
+
 } // namespace NYdb::NTopic

ydb/public/sdk/cpp/client/ydb_topic/impl/topic.cpp

@@ -12,6 +12,15 @@
 
 namespace NYdb::NTopic {
 
+class TCommonCodecsProvider {
+public:
+    TCommonCodecsProvider() {
+        NYdb::NTopic::TCodecMap::GetTheCodecMap().Set((ui32)NYdb::NPersQueue::ECodec::GZIP, MakeHolder<NYdb::NTopic::TGzipCodec>());
+        NYdb::NTopic::TCodecMap::GetTheCodecMap().Set((ui32)NYdb::NPersQueue::ECodec::ZSTD, MakeHolder<NYdb::NTopic::TZstdCodec>());
+    }
+};
+TCommonCodecsProvider COMMON_CODECS_PROVIDER;
+
 TDescribeTopicResult::TDescribeTopicResult(TStatus&& status, Ydb::Topic::DescribeTopicResult&& result)
     : TStatus(std::move(status))
     , TopicDescription_(std::move(result))

ydb/public/sdk/cpp/client/ydb_topic/impl/write_session_impl.cpp

@@ -1073,7 +1073,8 @@ TMemoryUsageChange TWriteSessionImpl::OnMemoryUsageChangedImpl(i64 diff) {
 
 TBuffer CompressBuffer(std::shared_ptr<TTopicClient::TImpl> client, TVector<TStringBuf>& data, ECodec codec, i32 level) {
     TBuffer result;
-    THolder<IOutputStream> coder = client->GetCodecImplOrThrow(codec)->CreateCoder(result, level);
+    Y_UNUSED(client);
+    THolder<IOutputStream> coder = NYdb::NTopic::TCodecMap::GetTheCodecMap().GetOrThrow((ui32)codec)->CreateCoder(result, level);
     for (auto& buffer : data) {
         coder->Write(buffer.data(), buffer.size());
     }