ydb-platform / ydb-platform/ydb-cpp-sdk
Refactor TopicClient event loop
Open
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
- Read the whole issue, then the project's contributing guide.
- Comment on the issue to say you are picking it up — it saves two people doing the same work.
- Fork the repository and make your change on a branch.
- 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