Before you start
Make sure that you can compile and run core tests and read a basic example Writing your first HTTP server.
It is recommended to read YDB service first, as it covers YDB table operations and topic consumption.
Step by step guide
In this tutorial we will write a service that publishes messages to a YDB topic using ydb::TopicWriter. The service would have the following REST API:
- HTTP POST by path /write with the message body in the request body writes the message to the configured YDB topic.
WriteHandler declaration
Like in Writing your first HTTP server we create a component for handling HTTP requests. The component holds references to ydb::TopicWriterManager and ydb::TopicWriter obtained from ydb::TopicWriterComponent:
class WriteHandler final : public server::handlers::HttpHandlerBase {
public:
static constexpr std::string_view kName{"handler-write"};
WriteHandler(const components::ComponentConfig& config, const components::ComponentContext& context);
std::string HandleRequest(server::http::HttpRequest& request, server::request::RequestContext& context)
const override;
private:
ydb::TopicWriterManager& writer_manager_;
ydb::TopicWriter& writer_;
};
WriteHandler constructor
The topic writer is looked up by name configured in static config (messages in this sample):
WriteHandler::WriteHandler(const components::ComponentConfig& config, const components::ComponentContext& context)
: server::handlers::HttpHandlerBase(config, context),
writer_manager_{context.FindComponent<
ydb::TopicWriterComponent>().GetTopicWriterManager()},
writer_(writer_manager_.GetTopicWriter(kWriterName))
{}
WriteHandler::HandleRequest
To publish a message, call ydb::TopicWriter::WriteMessage(). The method returns ydb::TopicWriteResult with a status that indicates whether the message was accepted into the writer queue:
request.GetHttpResponse().
SetContentType(http::content_type::kTextPlain);
const auto result = writer_.WriteMessage(request.
RequestBody(), {});
switch (result.GetStatus()) {
return "Message was written to topic\n";
return "Topic writer queue is full\n";
return "Failed to write message to topic\n";
}
return "Unexpected topic writer status\n";
}
ydb::TopicWriteStatus::kResourceExhausted means the internal queue is full and the caller should retry later. ydb::TopicWriteStatus::kOk means the message was accepted for delivery.
Static config
Static configuration of the service is quite close to the configuration from Writing your first HTTP server except for the YDB component, topic writer and handler:
# yaml
ydb:
operation-settings:
client-timeout: 1100ms
retries: 3
databases:
sampledb:
database: sampledb
max_pool_size: 10
min_pool_size: 5
Topic writer settings:
# yaml
ydb-topic-writer:
topics:
messages:
topic: messages
database: sampledb
HTTP handler:
# yaml
handler-write:
path: /write
method: POST
task_processor: main-task-processor
There are more static options for the topic writer component configuration, all of them are described at ydb::TopicWriterComponent.
See YDB for YDB hints and more usage samples.
int main()
Add the handler, ydb::YdbComponent and ydb::TopicWriterComponent to the components::MinimalServerComponentList() and run utils::DaemonRun:
int main(int argc, char* argv[]) {
const auto component_list =
.Append<server::handlers::TestsControl>()
.Append<components::DefaultSecdistProvider>()
.Append<samples::ydb_topic_writer::WriteHandler>()
.Append<ydb::TopicWriterComponent>();
}
Build and Run
To build the sample, execute the following build steps at the userver root directory:
mkdir build_release
cd build_release
cmake -DCMAKE_BUILD_TYPE=Release ..
make userver-samples-ydb_topic_writer_service
The sample could be started by running make start-userver-samples-ydb_topic_writer_service. The command would invoke testsuite start target that sets proper paths in the configuration files, prepares and starts YDB, and starts the service.
To start the service manually start the YDB server and run ./samples/ydb_topic_writer_service/userver-samples-ydb_topic_writer_service -c </path/to/static_config.yaml>.
Now you can send a request to your service from another terminal:
bash
$ curl -X POST 'http://localhost:8080/write' -d 'hello topic writer'
Message was written to topic
Functional testing
Functional tests for the service could be implemented using the testsuite. To do that you have to:
- Turn on the pytest_userver.plugins.ydb plugin and create the topic before each test:
import pytest
pytest_plugins = ['pytest_userver.plugins.core', 'pytest_userver.plugins.ydb']
@pytest.fixture(scope='session')
def ydb_topic_path() -> str:
return 'messages'
@pytest.fixture(scope='session')
def ydb_topic_consumer_name() -> str:
return 'test-consumer'
@pytest.fixture(autouse=True, scope='function')
def ydb_create_topic(ydb, ydb_topic_path, ydb_topic_consumer_name):
topic_client = ydb.topic_client
try:
describe_result = topic_client.describe_topic(ydb_topic_path)
if hasattr(describe_result, 'result'):
describe_result = describe_result.result()
_ = describe_result
except Exception:
create_result = topic_client.create_topic(
ydb_topic_path,
consumers=[ydb_topic_consumer_name],
)
if hasattr(create_result, 'result'):
create_result.result()
- Write the test that sends a message via HTTP and reads it back from the topic using the YDB topic client:
async def test_topic_writer_cycle(service_client, ydb, ydb_topic_path, ydb_topic_consumer_name):
message = 'hello topic writer'
response = await service_client.post('/write', data=message)
assert response.status == 200
assert response.text == 'Message was written to topic\n'
topic_client = ydb.topic_client
reader = topic_client.reader(
ydb_topic_path,
consumer=ydb_topic_consumer_name,
)
deadline = asyncio.get_running_loop().time() + 10.0
received_messages = []
while asyncio.get_running_loop().time() < deadline:
message_batch = reader.receive_message(timeout=1.0)
if hasattr(message_batch, 'result'):
message_batch = message_batch.result()
if not message_batch:
await asyncio.sleep(0.2)
continue
if isinstance(message_batch, (list, tuple)):
received_messages.extend(message_batch)
else:
received_messages.append(message_batch)
for received in received_messages:
data = getattr(received, 'data', None)
if callable(data):
data = data()
if data == message.encode():
if hasattr(reader, 'commit'):
commit_result = reader.commit(received)
if hasattr(commit_result, 'result'):
commit_result.result()
if hasattr(reader, 'close'):
close_result = reader.close()
if hasattr(close_result, 'result'):
close_result.result()
return
if hasattr(reader, 'close'):
close_result = reader.close()
if hasattr(close_result, 'result'):
close_result.result()
rendered = []
for received in received_messages:
data = getattr(received, 'data', None)
if callable(data):
data = data()
rendered.append(data)
raise AssertionError(f'Message `{message}` was not found in topic messages: {rendered}')
Full sources
See the full example:
