Skip to content

Migrate cloud/blockstore/libs/logbroker to standalone YDB C++ SDK #6854

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Draft
kobzonega wants to merge 1 commit into
base: codex/add-ydb-cpp-sdk
Choose a base branch
from
Draft

Migrate cloud/blockstore/libs/logbroker to standalone YDB C++ SDK #6854

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion cloud/blockstore/libs/logbroker/iface/credentials_provider.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -2,7 +2,7 @@

#include <cloud/blockstore/libs/logbroker/iface/config.h>

#include <contrib/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/iam/iam.h>
#include <ydb-cpp-sdk/client/iam/iam.h>

#include <util/string/cast.h>

Expand Down
2 changes: 1 addition & 1 deletion cloud/blockstore/libs/logbroker/iface/ya.make
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,7 +14,7 @@ PEERDIR(
library/cpp/monlib/service/pages
library/cpp/threading/future

contrib/ydb/public/sdk/cpp/src/library/jwt
contrib/libs/ydb-cpp-sdk/src/client/iam
)

END()
Expand Down
54 changes: 38 additions & 16 deletions cloud/blockstore/libs/logbroker/topic_api_impl/topic_api.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,9 +7,9 @@
#include <cloud/storage/core/libs/common/error.h>
#include <cloud/storage/core/libs/diagnostics/logging.h>

#include <contrib/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/iam/iam.h>
#include <contrib/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/topic/client.h>
#include <contrib/ydb/public/sdk/cpp/include/ydb-cpp-sdk/client/types/status_codes.h>
#include <ydb-cpp-sdk/client/iam/iam.h>
#include <ydb-cpp-sdk/client/topic/client.h>
#include <ydb-cpp-sdk/client/types/status_codes.h>

#include <util/generic/overloaded.h>
#include <util/stream/file.h>
Expand Down Expand Up @@ -91,16 +91,19 @@ class TService final

std::shared_ptr<NYdbICredentialsProviderFactory>
CredentialsProviderFactory;
TWriteSessionFactory WriteSessionFactory;

public:
TService(
TLogbrokerConfigPtr config,
ILoggingServicePtr logging,
std::shared_ptr<NYdbICredentialsProviderFactory>
credentialsProviderFactory)
credentialsProviderFactory,
TWriteSessionFactory writeSessionFactory)
: Config(std::move(config))
, Logging(std::move(logging))
, CredentialsProviderFactory(std::move(credentialsProviderFactory))
, WriteSessionFactory(std::move(writeSessionFactory))
{}

TFuture<NProto::TError> Write(
Expand All @@ -119,15 +122,21 @@ class TService final
Batch = std::make_unique<TBatch>(std::move(messages), now);

if (!Session) {
TTopicClient client{GetDriver()};

Session = client.CreateWriteSession(
TWriteSessionSettings()
.Path(Config->GetTopic())
.ProducerId(Config->GetSourceId())
.MessageGroupId(Config->GetSourceId())
.RetryPolicy(
NYdb::NTopic::IRetryPolicy::GetNoRetryPolicy()));
if (WriteSessionFactory) {
Session = WriteSessionFactory();
} else {
TTopicClient client{GetDriver()};

Session = client.CreateWriteSession(
TWriteSessionSettings()
.Path(Config->GetTopic())
.ProducerId(Config->GetSourceId())
.MessageGroupId(Config->GetSourceId())
.RetryPolicy(
NYdb::NTopic::IRetryPolicy::GetNoRetryPolicy()));
}

Y_ABORT_UNLESS(Session);
} else if (ContinuationToken.has_value()) {
TContinuationToken token{std::move(ContinuationToken.value())};
ContinuationToken.reset();
Expand Down Expand Up @@ -314,13 +323,26 @@ class TService final
IServicePtr CreateTopicAPIService(
TLogbrokerConfigPtr config,
ILoggingServicePtr logging,
std::shared_ptr<NYdbICredentialsProviderFactory>
credentialsProviderFactory)
std::shared_ptr<NYdbICredentialsProviderFactory> credentialsProviderFactory,
TWriteSessionFactory writeSessionFactory)
{
return std::make_shared<TService>(
std::move(config),
std::move(logging),
std::move(credentialsProviderFactory));
std::move(credentialsProviderFactory),
std::move(writeSessionFactory));
}

IServicePtr CreateTopicAPIService(
TLogbrokerConfigPtr config,
ILoggingServicePtr logging,
std::shared_ptr<NYdbICredentialsProviderFactory> credentialsProviderFactory)
{
return CreateTopicAPIService(
std::move(config),
std::move(logging),
std::move(credentialsProviderFactory),
{});
}

IServicePtr CreateTopicAPIService(
Expand Down
15 changes: 15 additions & 0 deletions cloud/blockstore/libs/logbroker/topic_api_impl/topic_api.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -5,10 +5,25 @@

#include <cloud/storage/core/libs/diagnostics/public.h>

#include <functional>

namespace NYdb::inline V3::NTopic {
class IWriteSession;
} // namespace NYdb::inline V3::NTopic

namespace NCloud::NBlockStore::NLogbroker {

////////////////////////////////////////////////////////////////////////////////

using TWriteSessionFactory =
std::function<std::shared_ptr<NYdb::NTopic::IWriteSession>()>;

IServicePtr CreateTopicAPIService(
TLogbrokerConfigPtr config,
ILoggingServicePtr logging,
std::shared_ptr<NYdbICredentialsProviderFactory> credentialsProviderFactory,
TWriteSessionFactory writeSessionFactory);

IServicePtr CreateTopicAPIService(
TLogbrokerConfigPtr config,
ILoggingServicePtr logging,
Expand Down
Loading
Loading