ydb-platform / ydb-platform/ydb-cpp-sdk

Refactor TopicClient event loop

Open
#359 0 comments 0 reactions 0 assignees View on GitHub

Nobody has claimed this yet.

Dominant language
C++
Stars
19
Forks
31
Avg merge
2d 5h
Merged PRs (30d)
10

Description

Current example for topic reader:

// Create topic client.

NYdb::NTopic::TTopicClient topicClient(driver);

// Create read session.
NYdb::NTopic::TReadSessionSettings settings;
settings
    .ConsumerName(opts.ConsumerName)
    .AppendTopics(opts.TopicPath);

ReadSession = topicClient.CreateReadSession(settings);

std::cerr << "Session was created" << std::endl;

// [BEGIN read session process events]
// Event loop
while (true) {
    auto future = ReadSession->WaitEvent();
    // Wait for next event or ten seconds
    future.Wait(TDuration::Seconds(10));
    // Get event
    std::optional<NYdb::NTopic::TReadSessionEvent::TEvent> event = ReadSession->GetEvent(true/*block - will block if no event received yet*/);
    std::cerr << "Got new read session event: " << DebugString(*event) << std::endl;

    if (auto* dataEvent = std::get_if<NYdb::NTopic::TReadSessionEvent::TDataReceivedEvent>(&*event)) {
        for (const auto& message : dataEvent->GetMessages()) {
            std::cerr << "Data message: \"" << message.GetData() << "\"" << std::endl;
        }

        if (opts.CommitAfterProcessing) {
            dataEvent->Commit();
        }
    } else if (auto* startPartitionSessionEvent = std::get_if<NYdb::NTopic::TReadSessionEvent::TStartPartitionSessionEvent>(&*event)) {
        startPartitionSessionEvent->Confirm();
    } else if (auto* stopPartitionSessionEvent = std::get_if<NYdb::NTopic::TReadSessionEvent::TStopPartitionSessionEvent>(&*event)) {
        stopPartitionSessionEvent->Confirm();
    } else if (auto* endPartitionSessionEvent = std::get_if<NYdb::NTopic::TReadSessionEvent::TEndPartitionSessionEvent>(&*event)) {
        endPartitionSessionEvent->Confirm();
    } else if (auto* closeSessionEvent = std::get_if<NYdb::NTopic::TSessionClosedEvent>(&*event)) {
        break;
    }
}

It's unfriendly API, we should design more simple and friendly API like ISimpleWriteSession

Contributor guide

No contributing guide indexed for this repository

First steps

  1. Read the whole issue, then the project's contributing guide.
  2. Comment on the issue to say you are picking it up — it saves two people doing the same work.
  3. Fork the repository and make your change on a branch.
  4. Open a pull request that references the issue number.

Research direction

Start with the topic reader example in the issue and trace the TTopicClient, TReadSession, WaitEvent, and GetEvent entry points. Compare the requested design with the existing ISimpleWriteSession API. Done means a decided, simpler read-session event-loop API with its behavior and scope specified.

Written by the indexing model from the issue text.

Assessment

Tech stack
cpp
Domain
api
Issue type
Refactor
Difficulty
5/5
Estimated time
Over a week
Activity status
Stale
Clarity
Needs clarification
Newbie friendliness
25/100

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.