Moved "provide codecs via singleton map" commit from ydb repo

Author: ildar-khisambeev

src/client/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 std::vector<ECodec>& GetDefaultCodecs() {
     static const std::vector<ECodec> codecs = {};
     return codecs;

src/client/persqueue_core/impl/read_session.ipp

@@ -2556,23 +2556,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()));
-                            std::string 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()));
+                        std::string 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()));
-                            std::string decompressed = codecImpl->Decompress(data.data());
-                            data.set_data(decompressed);
-                        }
+                        const NYdb::NTopic::ICodec* codecImpl = NYdb::NTopic::TCodecMap::GetTheCodecMap().GetOrThrow(static_cast<ui32>(batch.codec()));
+                        std::string decompressed = codecImpl->Decompress(data.data());
+                        data.set_data(decompressed);
                     }
                 }
 

src/client/persqueue_core/impl/write_session_impl.cpp

@@ -906,7 +906,8 @@ TMemoryUsageChange TWriteSessionImpl::OnMemoryUsageChangedImpl(i64 diff) {
 
 TBuffer CompressBuffer(std::shared_ptr<TPersQueueClient::TImpl> client, std::vector<std::string_view>& 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());
     }

src/client/topic/codecs/codecs.h

@@ -7,8 +7,9 @@
 #include <ydb-cpp-sdk/util/stream/str.h>
 #include <src/util/stream/zlib.h>
 #include <ydb-cpp-sdk/util/stream/output.h>
+#include <ydb-cpp-sdk/util/system/spinlock.h>
 
-#define CODECS_ALREADY_DEFINED
+#include <unordered_map>
 
 namespace NYdb::NTopic {
 
@@ -77,4 +78,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

src/client/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))

src/client/topic/impl/write_session_impl.cpp

@@ -1096,7 +1096,8 @@ TMemoryUsageChange TWriteSessionImpl::OnMemoryUsageChangedImpl(i64 diff) {
 
 TBuffer CompressBuffer(std::shared_ptr<TTopicClient::TImpl> client, std::vector<std::string_view>& 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());
     }

src/util/random/entropy.cpp

@@ -9,6 +9,7 @@
 #include <ydb-cpp-sdk/util/stream/mem.h>
 #include <src/util/stream/zlib.h>
 #include <src/util/stream/buffer.h>
+#include <src/util/stream/buffered.h>
 
 #include <src/util/system/fs.h>
 #include <src/util/system/info.h>

src/util/stream/zlib.h

@@ -3,7 +3,6 @@
 #include <ydb-cpp-sdk/util/stream/fwd.h>
 #include <ydb-cpp-sdk/util/stream/input.h>
 #include <ydb-cpp-sdk/util/stream/output.h>
-#include "buffered.h"
 
 #include <ydb-cpp-sdk/util/system/defaults.h>
 #include <ydb-cpp-sdk/util/generic/ptr.h>