userver: TCP half-duplex server with static configs validation
No Matches
TCP half-duplex server with static configs validation

Before you start

Make sure that you can compile and run core tests as described at Configure, Build and Install.

Step by step guide

Let's write a simple TCP server that accepts incoming connections, and as long as the client sends "hi" responds with greeting from configuration file.

TCP server

Derive from components::TcpAcceptorBase and override the ProcessSocket function to get the new sockets:

namespace samples::tcp {
class Hello final : public components::TcpAcceptorBase {
static constexpr std::string_view kName = "tcp-hello";
// Component is valid after construction and is able to accept requests
Hello(const components::ComponentConfig& config,
: TcpAcceptorBase(config, context),
greeting_(config["greeting"].As<std::string>("hi")) {}
void ProcessSocket(engine::io::Socket&& sock) override;
static yaml_config::Schema GetStaticConfigSchema();
const std::string greeting_;
} // namespace samples::tcp
ProcessSocket functions are invoked concurrently on the same instance of the class. Use synchronization primitives or do not modify shared data in ProcessSocket.

Static config

Our new "tcp-hello" component should support the options of the components::TcpAcceptorBase and the "greeting" option. To achieve that we would need the following implementation of the GetStaticConfigSchema function:

yaml_config::Schema Hello::GetStaticConfigSchema() {
return yaml_config::MergeSchemas<TcpAcceptorBase>(R"(
type: object
description: |
Component for accepting incoming TCP connections and responding with some
greeting as long as the client sends 'hi'.
additionalProperties: false
type: string
description: greeting to send to client
defaultDescription: hi

Now lets configure our component in the components section:

# yaml
task_processor: main-task-processor # Run socket accepts on CPU bound task processor
sockets_task_processor: main-task-processor # Run ProcessSocket() for each new socket on CPU bound task processor
port: 8180
greeting: hello


It's time to deal with new sockets. The code is quite straightforward:

void Hello::ProcessSocket(engine::io::Socket&& sock) {
std::string data;
const auto read_bytes = sock.ReadAll(, 2, {});
if (read_bytes != 2 || data != "hi") {
const auto sent_bytes =
sock.SendAll(, greeting_.size(), {});
if (sent_bytes != greeting_.size()) {

int main()

Finally, add the component to the components::MinimalComponentList(), and start the server with static configuration file passed from command line.

int main(int argc, const char* const argv[]) {
const auto component_list =
return utils::DaemonMain(argc, argv, component_list);

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-tcp_service

The sample could be started by running make start-userver-samples-tcp_service. The command would invoke testsuite start target that sets proper paths in the configuration files and starts the service.

To start the service manually run ./samples/tcp_service/userver-samples-tcp_service -c </path/to/static_config.yaml>.

Now you can send a request to your server from another terminal:

$ nc localhost 8180

Functional testing

Functional tests for the service could be implemented using the testsuite in the following way:

import socket
import pytest
async def test_basic(service_client, loop, tcp_service_port):
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.connect(('localhost', tcp_service_port))
await loop.sock_sendall(sock, b'hi')
hello = await loop.sock_recv(sock, 5)
assert hello == b'hello'
await loop.sock_sendall(sock, b'whats up?')
with pytest.raises(ConnectionResetError):
await loop.sock_recv(sock, 1)

Note that in this case testsuite requires some help to detect that the service is ready to accept requests. To do that, override the pytest_userver.plugins.service.service_non_http_health_checks :

import pytest
from pytest_userver.utils import net
pytest_plugins = ['pytest_userver.plugins.core']
@pytest.fixture(name='tcp_service_port', scope='session')
def _tcp_service_port(service_config) -> int:
components = service_config['components_manager']['components']
tcp_hello = components.get('tcp-hello')
assert tcp_hello, 'No "tcp-hello" component found'
return int(tcp_hello['port'])
def service_non_http_health_checks(
service_config, tcp_service_port,
) -> net.HealthChecks:
checks = net.get_health_checks_info(service_config)
checks.tcp.append(net.HostPort(host='localhost', port=tcp_service_port))
return checks

Full sources

See the full example at: