---
metadata:
  - name: generator
    content: Diplodoc Platform v5.63.3
alternate:
  - https://ytsaurus.tech/docs/en/flow/go/examples/async_request.md
  - https://ytsaurus.tech/docs/ru/flow/go/examples/async_request.md
  - href: https://ytsaurus.tech/docs/en/flow/go/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/go/examples/async_request.md -->
# Async Request in YTsaurus Flow (Go)

An example of a [pipeline](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#pipeline) that makes an asynchronous call to an external service: one [computation](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#stream-and-computation) turns events into requests and accumulates the responses in an external [state](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#state), and another one serves the requests without any state. This is a Go implementation of the same scenario as the [C++ example](https://ytsaurus.tech/docs/en/flow/cpp/examples/async_request.md).

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

## Structure {#structure}

The companion serves two computations; `injector` stays a native source declared in the [spec](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#spec-and-dynamic-spec):

1. **`state`** (`stateKeeper`) — a stateful computation grouped by `key`, which:
   - takes events from the `event` stream and produces a request in the `request` stream with a random `request_id`;
   - takes responses from the `response` stream and adds up their total length (`total_length`) in the external state `/state`.

2. **`processor`** (`requestProcessor`) — a stateless computation grouped by `request_id`: it takes requests from the `request` stream and immediately answers with the length of the request string in the `response` stream.

The `event → request → response → state` cycle closes between the two computations. An event is answered with a request rather than with the result right away, so the serving side never holds up processing: the response arrives later, as a separate [message](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#message), and only then does the state of the key move on.

## `main.go` {#main-go}

The entry point: creating the pipeline and registering both computations.

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

```go

type eventMessage struct {
	flow.YSONMessage
	Key  uint64 `yson:"key"`
	Data string `yson:"data"`
}

type requestMessage struct {
	flow.YSONMessage
	RequestID uint64 `yson:"request_id"`
	Key       uint64 `yson:"key"`
	Request   string `yson:"request"`
}

type responseMessage struct {
	flow.YSONMessage
	RequestID uint64 `yson:"request_id"`
	Key       uint64 `yson:"key"`
	Length    int64  `yson:"length"`
}

func main() {
	pipeline := flow.NewPipeline()
	pipeline.AddStreams(
		flow.NewYSONStream[eventMessage](eventStreamID),
		flow.NewYSONStream[requestMessage](requestStreamID),
		flow.NewYSONStream[responseMessage](responseStreamID),
	)
	pipeline.Add(
		flow.NewRowComputation("state", &stateKeeper{}),
		flow.NewRowComputation("processor", &requestProcessor{}),
	)

	if err := pipeline.Run(); err != nil {
		fmt.Fprintf(os.Stderr, "async_request: %v\n", err)
		os.Exit(1)
	}
}
```

## `state_keeper.go` {#state-keeper-go}

Routing of the input streams (`event` / `response`) and work with the external state.

```go

type totalLengthState struct {
	TotalLength int64 `yson:"total_length"`
}

type stateKeeper struct{}

var _ flow.RowFunction = (*stateKeeper)(nil)

func (k *stateKeeper) OnMessage(
	ctx context.Context,
	rt flow.Runtime,
	msg flow.ExtendedMessage,
	out flow.OutputCollector,
) error {
	switch msg.StreamID {
	case eventStreamID:
		return k.sendRequest(rt, msg, out)
	case responseStreamID:
		return k.accumulate(rt, msg)
	default:
		return fmt.Errorf("unhandled stream %q", msg.StreamID)
	}
}

func (*stateKeeper) sendRequest(rt flow.Runtime, msg flow.ExtendedMessage, out flow.OutputCollector) error {
	var event eventMessage
	if err := msg.ConvertTo(&event); err != nil {
		return err
	}

	request := flow.NewYSONMessage[requestMessage](requestStreamID)
	request.RequestID = rand.Uint64()
	request.Key = event.Key
	request.Request = event.Data

	encoded, err := flow.ConvertFrom(rt, request)
	if err != nil {
		return err
	}
	out.AddMessage(encoded)
	return nil
}

func (*stateKeeper) accumulate(rt flow.Runtime, msg flow.ExtendedMessage) error {
	var response responseMessage
	if err := msg.ConvertTo(&response); err != nil {
		return err
	}

	state, err := flow.OpenExternalState(rt, totalLengthStateName, msg)
	if err != nil {
		return err
	}

	var total totalLengthState
	if _, err := state.ConvertTo(&total); err != nil {
		return err
	}
	total.TotalLength += response.Length
	return state.ConvertFrom(&total)
}
```

## `request_processor.go` {#request-processor-go}

The stateless request handler: it computes the length of the request string and returns a response.

```go

type requestProcessor struct{}

var _ flow.RowFunction = (*requestProcessor)(nil)

func (*requestProcessor) OnMessage(
	ctx context.Context,
	rt flow.Runtime,
	msg flow.ExtendedMessage,
	out flow.OutputCollector,
) error {
	var request requestMessage
	if err := msg.ConvertTo(&request); err != nil {
		return err
	}

	response := flow.NewYSONMessage[responseMessage](responseStreamID)
	response.RequestID = request.RequestID
	response.Key = request.Key
	response.Length = int64(len(request.Request))

	encoded, err := flow.ConvertFrom(rt, response)
	if err != nil {
		return err
	}
	out.AddMessage(encoded)
	return nil
}
```

## Key patterns {#key-patterns}

- **Routing by `msg.StreamID`**: a `switch` on the input stream identifier lets one computation handle several inputs with different logic. An unknown stream is an error rather than a silent skip.
- **A random `request_id`**: `rand.Uint64()` links the request to the response. The request carries the key of the original event, so the response, partitioned by `request_id`, comes back to the state the call belongs to.
- **External state** through `flow.OpenExternalState(rt, "/state", msg)`: the row is turned into `totalLengthState` by `ConvertTo`, changed as a structure, and saved through `ConvertFrom`.
- **A stateless computation**: `requestProcessor` uses no state and is grouped by `request_id` rather than by the event key, so the requests of one key spread over all [partitions](https://ytsaurus.tech/docs/en/flow/concepts/glossary.md#partition) and scale independently.
- **Stream dependency**: `streams_dependency` in the spec declares that `request` is produced from `event` — the worker takes this into account when advancing [watermarks](https://ytsaurus.tech/docs/en/flow/concepts/watermarks.md).
<!-- endsource: en/_includes/flow/go/examples/async_request.md -->

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

- [Getting started (Go)](https://ytsaurus.tech/docs/en/flow/go/getting-started.md)
- [Computation (Go)](https://ytsaurus.tech/docs/en/flow/go/computation.md)
- [External State (Go)](https://ytsaurus.tech/docs/en/flow/go/external-state.md)
- [Similar example in C++](https://ytsaurus.tech/docs/en/flow/cpp/examples/async_request.md)
- [Similar example in Java](https://ytsaurus.tech/docs/en/flow/java/examples/async_request.md)
<!-- endsource: en/_includes/flow/go/examples/async_request_also.md -->