Async Request in YTsaurus Flow (C++)

The pipeline demonstrates a pattern for asynchronous external requests using process functions. You send events to the input, convert them into requests, process them with a deterministic function, and accumulate the results in the state.

Source code

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.

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

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

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);
    }
}

See also

• Getting started (C++)
• Computation (C++)