---
metadata:
  - name: generator
    content: Diplodoc Platform v5.63.3
alternate:
  - https://ytsaurus.tech/docs/en/flow/cpp/examples/async_request.md
  - https://ytsaurus.tech/docs/ru/flow/cpp/examples/async_request.md
  - href: https://ytsaurus.tech/docs/en/flow/cpp/examples/async_request.md
    type: text/markdown
    title: Markdown version
  - href: https://ytsaurus.tech/docs/en/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65
    rel: describedby
---
> **Documentation Index:** Fetch the complete configuration index at https://ytsaurus.tech/docs/en/llms.txt?revision=181542d8c2d9f36be925894a517db8203cafae65

<!-- source: en/_includes/flow/cpp/examples/async_request.md -->
# Async Request in YTsaurus Flow (C++)

The [pipeline](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#pipeline) demonstrates a pattern for asynchronous external requests using [process functions](https://ytsaurus.tech/docs/en/flow/cpp/process-functions.md). You send events to the input, convert them into requests, process them with a deterministic function, and accumulate the results in the [state](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#state).

[Source code](https://github.com/ytsaurus/ytsaurus/tree/main/yt/yt/flow/examples/cpp/async_request)

## Pipeline components

### TStateKeeper

`TStateKeeper` implements `IProcessFunction` and is hosted by `TProcessFunctionComputation`. The function uses `TSimpleExternalStateManager` to work with the external state and handles two types of input messages:

- **The `event` stream**: when you receive an event, it creates a `TRequestMessage` with a unique `RequestId` and sends it to the `request` stream.
- **The `response` stream**: when you receive a response, it updates the state — it sums `total_length` from all received responses.

You distinguish the streams using `message->StreamId`.

### TRequestProcessor

`TRequestProcessor` implements `IProcessFunction` and is hosted by `TProcessFunctionSwiftMapComputation`, which doesn’t store input and output messages in YTsaurus. The function receives a `TRequestMessage`, performs processing (in this example, it calculates the request length), and generates a `TResponseMessage`.

You use the SwiftMap adapter because the request processing is a pure function: the same input data always generates the same result.

## Message types

- **TEventMessage** — a descendant of `TYsonMessage`. It contains the `Key` and `Data` fields.
- **TRequestMessage** — a descendant of `TYsonMessage`. It contains the `RequestId`, `Key`, and `Request` fields.
- **TResponseMessage** — a descendant of `TYsonMessage`. It contains the `RequestId`, `Key`, and `Length` fields.

You register all message types using the `YT_FLOW_DEFINE_YSON_MESSAGE` macro.

## Key pattern: request-response cycle

The main idea of this example is to build a request-response cycle within the pipeline using multiple streams:

1. **events** → `TStateKeeper` → **request** (request generation)
2. **request** → `TRequestProcessor` → **response** (request processing)
3. **response** → `TStateKeeper` → state (result accumulation)

`TStateKeeper` is both an event consumer and a response consumer. It uses `input_stream_ids = ["event", "response"]` and determines the input message type by `StreamId`.

The spec binds each process function to its adapter explicitly:

- `state` uses `computation_class_name = "NYT::NFlow::TProcessFunctionComputation"` and `processing_function = "NYT::NFlow::NExample::TStateKeeper"`.
- `processor` uses `computation_class_name = "NYT::NFlow::TProcessFunctionSwiftMapComputation"` and `processing_function = "NYT::NFlow::NExample::TRequestProcessor"`.

## State management

`TStateKeeper` uses `TSimpleExternalStateManager` to store the sum of lengths of all processed requests. You bind the state client (`TMutableStateKeyClient<TSimpleExternalState>`) in `Init(const IRuntimeInitContextPtr&)` via `initContext->InitExternalStateClient(StateClient_, "/state")`. You declare the state parameters (the `path` to the table, etc.) in the `external_state_managers` section of the computation [spec](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#spec-and-dynamic-spec).

## The main function

In `main`, you register three streams; the process functions are registered via `YT_FLOW_DEFINE_PROCESS_FUNCTION`:

- `RegisterStream<TEventMessage>("event")` — input events
- `RegisterStream<TRequestMessage>("request")` — requests to the processor
- `RegisterStream<TResponseMessage>("response")` — responses from the processor

## Source code

### TRequestProcessor

```cpp
void TRequestProcessor::ProcessMessage(
    const TInputMessageConstPtr& message,
    const IOutputCollectorPtr& output,
    const IRuntimeContextPtr& context)
{
    auto request = context->ConvertToYsonMessage<TRequestMessage>(message);
    auto response = New<TResponseMessage>();
    response->RequestId = request->RequestId;
    response->Key = request->Key;
    response->Length = std::ssize(request->Request);
    output->AddMessage(context->ConvertToMessage(response));
    YT_TLOG_DEBUG("Processed request")
        .With("RequestId", request->RequestId)
        .With("FailedAttempts", 0);
}
```

### TStateKeeper

```cpp
void TStateKeeper::Init(const IRuntimeInitContextPtr& initContext)
{
    initContext->InitExternalStateClient(StateClient_, "/state");
}

void TStateKeeper::ProcessMessage(
    const TInputMessageConstPtr& message,
    const IOutputCollectorPtr& output,
    const IRuntimeContextPtr& context)
{
    if (message->StreamId == "event") {
        auto event = context->ConvertToYsonMessage<TEventMessage>(message);
        auto request = New<TRequestMessage>();
        request->RequestId = RandomNumber<ui64>();
        request->Key = event->Key;
        request->Request = event->Data;
        output->AddMessage(context->ConvertToMessage(request));
        YT_TLOG_DEBUG("Send request")
            .With("RequestId", request->RequestId);
    } else if (message->StreamId == "response") {
        auto response = context->ConvertToYsonMessage<TResponseMessage>(message);
        YT_TLOG_DEBUG("Received response")
            .With("RequestId", response->RequestId);
        auto state = StateClient_.GetState(message->Key);
        i64 totalLength = state->GetColumnValue<std::optional<i64>>("total_length").value_or(0);
        totalLength += response->Length;
        TPayloadBuilder builder(state->Schema);
        builder.Set(totalLength, "total_length");
        state->Payload = builder.Finish();
    } else {
        THROW_ERROR_EXCEPTION("Unexpected stream_id")
            .With("stream_id", message->StreamId);
    }
}
```
<!-- endsource: en/_includes/flow/cpp/examples/async_request.md -->

<!-- source: en/_includes/flow/cpp/examples/async_request_also.md -->
## See also

- [Getting started (C++)](https://ytsaurus.tech/docs/en/flow/cpp/getting-started.md)
- [Computation (C++)](https://ytsaurus.tech/docs/en/flow/cpp/computation.md)
<!-- endsource: en/_includes/flow/cpp/examples/async_request_also.md -->