Streaming в Connect

Всё начинается в схеме — proto-first никто не отменял. В protobuf потоковые вызовы описываются тем же rpc, но перед типом появляется слово stream, и позиция этого слова определяет жанр: stream перед типом ответа — сервер шлёт поток (server streaming), перед типом запроса — клиент шлёт поток (client streaming), перед обоими — двунаправленный (bidi). Дополним eliza.proto из третьего урока:rpc Introduce — сервер-консультант представляется и шлёт несколько фраз-знакомств по одной;rpc Converse — диалог, где клиент и сервер обмениваются репликами в произвольном порядке. Генерация ничего нового не требует: buf generate из урока про установку соберёт те же два плагина, protoc-gen-connect-go просто добавит в интерфейс ElizaServiceHandler новые методы — со своими, стриминговыми сигнатурами. Заглушка UnimplementedElizaServiceHandler, о которой мы говорили в третьем уроке, ответит unimplemented на незанятые методы, пока вы не допишете реализацию, — код останется зелёным в любой момент эволюции схемы.

Сгенерированный интерфейс выглядит так: для server streaming хендлер получает контекст, обычный конверт connect.Request с запросом и новый для вас тип — connect.ServerStream[Res], поток, в который хендлер отправляет ответы. Сигнатура: func Introduce(ctx context.Context, req connect.Request[elizav1.IntroduceRequest], stream connect.ServerStream[elizav1.IntroduceResponse]) error. Обратите внимание на возврат — просто error, без объекта ответа: ответов много, и каждый уходит в поток через метод Send, который принимает указатель на message-тип. Цикл сервера предельно честный: шлём сообщения, пока есть что шлёшь, проверяя по дороге контекст — если клиент отключился или истёк дедлайн, ctx.Err() вернёт отмену, и хендлер должен остановиться. Дедлайны здесь работают так же, как в unary: клиент задаёт их через контекст, о чём мы говорили в уроке про клиентов, — разница в том, что у стрима больше точек, где отмена перехватывается. Ошибка, возвращённая из хендлера в любой момент, закрывает поток: сообщения, ушедшие до неё, клиент получит, а код ошибки придёт после них — частичные результаты в стриминге законны и полезны, это одно из главных отличий жанра от unary.

Клиентская сторона зеркальна. Сгенерированный метод Introduce принимает конверт запроса и возвращает пару (connect.ServerStreamForClient[elizav1.IntroduceResponse], error) — та же идея типов-двойников, что и с UnaryFunc у перехватчиков: серверная ServerStream для записи, клиентская ServerStreamForClient для чтения. Ошибка из первого return — это ошибка старта вызова: маршрут не найден, авторизация не прошла, сервер отклонил запрос до начала потока. Дальше — цикл чтения, самая важная идиома этого урока: stream.Receive() возвращает следующее сообщение или ошибку, и конец потока выражается значением io.EOF. Это та же семантика, что у io.Reader из стандартной библиотеки: EOF — не провал, а нормальный сигнал «данные кончились». Ошибка реальная — это connect.Error, и здесь работает весь арсенал из урока про ошибки: connect.CodeOf, errors.As, разбор кодов. Итого цикл чтения всегда выглядит одинаково: Receive, проверка io.EOF — выход, проверка ошибки — обработка, иначе — работа с сообщением. Заучите эту форму: она встретится в каждом стриминговом клиенте вашего кода.

Двунаправленный стриминг — жанр, где обе стороны шлют сообщения независимо: диалог, чат, живая синхронизация. Здесь у Connect один тип на сервере — connect.BidiStream[Req, Res] и один у клиента — connect.BidiStreamForClient[Req, Res]; хендлер получает только контекст и поток, без конверта запроса, потому что запросов много, и все они приходят через stream.Receive(). Как и в gRPC, bidi на уровне API — это два независимых полудуплекса: вы шлёте через Send и читаете через Receive, библиотека разруливает мультиплексирование. А если нужен жанр client streaming — клиент шлёт поток, сервер отвечает один раз? Отдельного типа в Connect нет, и это осознанное упрощение: client streaming выражается через тот же BidiStream, просто хендлер читает циклом, а отвечает одним Send. Клиенту в этом жанре доступен метод CloseRequest — закрыть свою половину, сообщив серверу «я закончил слать», — а CloseResponse закрывает приёмную. Двунаправленные потоки требуют HTTP/2: ограничение протокола, о котором мы говорили ещё в первом уроке; server streaming живёт и на HTTP/1.1. Внутри горутин следите за завершением: одна сторона читает, другая шлёт, и обе обязаны увидеть закрытие потока — иначе утечка горутин, тема, которую мы разбирали в gRPC-курсе на чате.

Теперь закрываем обещание из урока про перехватчики: как выглядит сквозная логика на стримах. Методы WrapStreamingClient и WrapStreamingHandler работают не с request-объектом, а с соединением: next получает StreamingClientFunc вида func(ctx, Spec) StreamingClientConn — и ваш перехватчик возвращает собственную обёртку над conn. У соединения есть всё то же, что было у AnyRequest: conn.Spec() с Procedure и StreamType, conn.RequestHeader() для чтения и записи заголовков, conn.Send и conn.Receive с обобщёнными данными сообщения, плюс CloseRequest и CloseResponse у клиента. Классический серверный сценарий — авторизация стрима: перехватчик читает Authorization из conn.RequestHeader() до запуска next и на провал возвращает connect.NewError с CodeUnauthenticated — как в unary. Клиентский — навешивание токена: обёртка выставляет заголовок на соединении один раз, и все сообщения стрима едут с ним. Заметьте одну тонкость: интерсептор оборачивает соединение целиком, а не каждое сообщение — «до/после» сработает один раз на вызов, а не на каждое Receive. Если нужна статистика по сообщениям, обёртка декорирует методы conn.Send и conn.Receive — счётчик сообщений пишется именно там.

Отдельный разговор — что происходит, когда всё пошло не так посреди потока. Сервер возвращает connect.NewError в любой момент: сообщения, ушедшие до ошибки, клиент уже получил, код придёт следом — и клиентский Receive вернёт *connect.Error с этим кодом вместо io.EOF. Это принципиально отличается от unary, где ошибка обнуляет весь вызов: в стриминге частичный успех — легитимный исход, и API следует проектировать так, чтобы клиент мог понять, сколько данных он получил. Коды работают те же, что и в unary: resource_exhausted на превышение лимитов, deadline_exceeded при истёкшем дедлайне, canceled при отмене контекста клиентом — отмена распространяется вниз по стеку, и сервер обязан это заметить через ctx.Err(). Таймаут ставится клиентом на весь вызов, как мы разбирали в пятом уроке: стрим длиной в час с дедлайном в минуту завершится deadline_exceeded, даже если сообщения шли исправно. И ещё одна дисциплина: проверяйте ошибку каждого Send — после первого сбоя поток мёртв, дальнейшие отправки бесполезны. Хендлер, игнорирующий ошибку Send и продолжающий цикл, напрасно тратит CPU и маскирует реальную проблему.

Стриминг встал в ряд: schema-side stream-типы, серверные Send и клиентские Receive, io.EOF, ошибки в середине потока, перехватчики соединений. Восьмой урок закрывает второй жанр RPC — и готовит сцену для следующего шага: браузер. Клиенты, которые мы писали до сих пор, живут в Go-процессах, а в девятом уроке выйдет в браузер TypeScript-клиент: connect-web, fetch-транспорт, CORS и асинхронные итераторы вместо циклов Receive — всё то, ради чего Connect и создавался. Для сравнения полезно держать под рукой, как те же жанры устроены в gRPC: server streaming мы разбирали в gRPC-уроке про серверные потоки, а дуплексный чат — в уроке про bidi — API другое, а дисциплина горутин и завершения потоков та же. Программа всех двадцати пяти уроков — на странице курса.