Итоговый проект: чат и файловый сервис

grpc проект-финал: весь курс разом#

В финальном блоке прошлого урока была строка-обещание: двадцать пятый урок — итоговый проект, двусторонний чат и файловый сервис на bidi-стримах и стриминговых загрузках. Уроки, которые обещают проект, обязаны его показать целиком — от схемы до теста, без пропусков «домашним заданием». Вот что у нас уже есть по частям: контракт в proto3 из второго урока, генерация из третьего, сервер и клиент из четвёртого и пятого, все три стриминговые модели из седьмого, восьмого и девятого, метаданные из десятого, перехватчики из одиннадцатого, TLS из двенадцатого, статусы из тринадцатого, дедлайны и отмены из четырнадцатого, health из семнадцатого, reflection из восемнадцатого, версионирование из двадцатого, bufconn-тесты из двадцать первого, лимиты и keepalive из двадцать второго. Итоговый grpc проект не вводит ни одного нового API — это и есть проверка: если всё было понято вовремя, здесь нечему быть сложным. Строим приложение из двух половин. Первая — общение: ChatService с комнатами, presence и двунаправленным обменом сообщениями. Вторая — файлы: FileService, который отдаёт файлы потоком чанков и принимает их обратно, считая размер и хеш на лету. Обе половины живут в одном пакете shop.v1, регистрируются на одном сервере и защищаются одним перехватчиком — так и выглядит типовой монолит, который потом легко разъезжается на сервисы.

Техническое задание: что должно уметь приложение#

Прежде чем писать схему, зафиксируем требования — без ТЗ итоговый проект превращается в демо. Запрос «чат grpc» на практике — это почти всегда приложение с вот таким списком. Общение: пользователи входят в комнаты по имени, комната создаётся первым участником; каждый видит события presence — кто вошёл, кто вышел, сколько людей online; сообщения доставляются всем участникам комнаты, включая автора, — эхо-подтверждение дешевле отдельного механизма. Файлы: Download принимает имя и отдаёт содержимое чанками фиксированного размера; Upload принимает поток чанков и возвращает ровно один итог — суммарный размер и sha256, посчитанный по мере приёма; файл сохраняется в хранилище на диске. Доступ: Join и Upload требуют Bearer-токен в metadata, Download открыт анонимно — чтение разрешено всем, запись и присутствие только своим. Ошибки: пустая комната или пустое имя файла — InvalidArgument, несуществующий файл — NotFound, плохой токен — PermissionDenied, отсутствие токена — Unauthenticated; никаких паник и голых ошибок. Сроки: файловые вызовы получают дедлайн, долгий чат живёт на отмене, а не на сроке. Проверки: интеграционный тест на bufconn гоняет полный круг Upload → Download и сверяет хеш. Всё перечисленное реализуется тремя RPC: один bidi и два стрима в разные стороны. Обратите внимание, чего в ТЗ нет: unary-вызова — он остался за бортом сознательно, и позже разберём, почему это честный выбор схемы, а не недоработка.

Карта курса: какой урок что дал проекту#

Урок	Что там было	Что забираем в проект
01 Введение	Модель RPC поверх HTTP/2, четыре кардинальности вызова	Каркас вызова: проект построен на трёх стриминговых кардинальностях из четырёх
02 proto3	message, enum, oneof, service, номера полей	oneof ChatEvent: сообщение и presence в одном типе; правила оформления схемы
03 Генерация	protoc и buf generate, стабы _grpc.pb.go	Одна генерация даёт стабы обоих сервисов разом
04 Сервер	grpc.NewServer, Register*, Serve	Оба сервиса регистрируются на одном сервере и одном порту
05 Клиент	grpc.NewClient, New*ServiceClient	Одно соединение — два клиента: ChatServiceClient и FileServiceClient
06 Unary	Запрос-ответ, простейший вызов	Единственная модель, которой в проекте нет, — полезно помнить, чем она отличается
07 Server streaming	WatchOrders, цикл Recv до io.EOF	Download: файл уезжает потоком чанков
08 Client streaming	UploadOrderEvents, Send плюс CloseAndRecv	Upload: чанки на вход, агрегированный итог на выход
09 Bidi	ChatTickets, Send, Recv, CloseSend, горутины	Join: каркас двунаправленного чата целиком
10 Metadata	AppendToOutgoingContext, FromIncomingContext	Токен authorization едет вне сообщений; ключи metadata строчные
11 Перехватчики	Unary- и Stream-перехватчики, цепочки	authStream: проверка токена в одном месте по FullMethod
12 TLS	TransportCredentials, mTLS	В bufconn-тестах insecure, в бою — креды из двенадцатого урока
13 Статусы	codes.*, status.Error, детали Any	InvalidArgument, NotFound, PermissionDenied — язык ошибок проекта
14 Дедлайны	WithTimeout, grpc-timeout, отмена	Срок на файловые вызовы; ctx.Done() будит хендлеры чата
15 Ретраи	retryPolicy, идемпотентность	Повтор стрима политикой не делается: переподключение чата — работа клиента
16 Балансировка	pick_first и round_robin	Долгая bidi-сессия живёт на одном соединении — липкость по умолчанию
17 Health	grpc.health.v1, Check и Watch	Health-сервис рядом: пробы k8s и прокси на оба сервиса
18 Reflection	reflection.Register, grpcurl	grpcurl видит ChatService и FileService без схемы под рукой
19 proto-оптимизация	Wire-формат, varint, LEN, reserved	bytes для чанков: LEN-тип без varint-накладной на каждый байт
20 Версионирование	package shop.v1, совместимость	Два сервиса в одной версии пакета растут вместе и аддитивно
21 Тестирование	bufconn, t.Cleanup, табличные тесты	Интеграционный roundtrip Upload → Download со сверкой хеша
22 Тюнинг	keepalive, message size	Чанки 64КБ против лимита 4 МиБ; keepalive держит долгий чат
23 Микросервисы	Gateway, mesh, канарейки	Начинаем с модульного монолита: два сервиса — один процесс
24 Проксирование	Envoy, gRPC-Web, ingress	За прокси код не меняется; браузеру bidi через gRPC-Web не по силам

Проектная схема: один proto, два сервиса#

Оба сервиса живут в одном файле api/shop/v1/app.proto и одном пакете shop.v1 — это не лень, а решение из двадцатого урока: версия пакета объявляет совместимость, и чат с файлами выпускаются одной версией, потому что меняются вместе. Порядок буквы v здесь тот же, что у OrderService из всех прошлых уроков: полный путь метода складывается из пакета и имени сервиса — /shop.v1.ChatService/Join, /shop.v1.FileService/Download, /shop.v1.FileService/Upload. Именно эти строки увидите в логах перехватчика из одиннадцатого урока, в grpcurl из восемнадцатого и в маршруте Envoy из двадцать четвёртого — по ним перехватчик различает, чей вызов пришёл. Почему оба сервиса рядом, а не в разных репозиториях: у них общая судьба — один процесс, один health-статус, один деплой. Когда файловый сервис захочет свою команду и свою частоту релизов, его переедет выносить двадцать третий урок — но контракт уже спроектирован так, что съезжает без правок: имена сообщений уникальны в пакете, пересечений с OrderService нет, номера полей у каждого типа свои. В схеме три RPC и четыре сообщения, и каждое имя здесь выбрано под работу: Join описывает, что делает клиент, Chunk — что едет в файловом стриме, ChatEvent — что событие может быть разным, потому и oneof.

app.proto: контракт итогового проекта

// api/shop/v1/app.proto — итоговый проект: чат и файлы в одной схеме.
syntax = "proto3";

package shop.v1; // та же версия пакета, что у OrderService: растут вместе

option go_package = "example.com/shop/gen/shop/v1;shopv1";

// ===== ChatService: bidi-чат с комнатами и presence =====
service ChatService {
  // Join — единственный метод сервиса, и он двунаправленный:
  // клиент шлёт ChatEvent с сообщением, сервер отвечает ChatEvent
  // с сообщениями соседей и событиями presence.
  rpc Join (stream ChatEvent) returns (stream ChatEvent);
}

message ChatEvent {
  oneof payload { // ровно одно из двух — как в событиях из второго урока
    ChatMsg msg = 1;       // сообщение в комнату (клиент -> все)
    Presence presence = 2; // событие присутствия (только сервер -> всем)
  }
}

message ChatMsg {
  string room = 1;   // имя комнаты; задаётся первым сообщением клиента
  string author = 2; // кто пишет
  string text = 3;   // текст сообщения
}

message Presence {
  string room = 1;  // комната, в которой произошло событие
  string user = 2;  // кто вошёл или вышел
  bool joined = 3;  // true — вошёл, false — вышел
  int32 online = 4; // сколько человек осталось в комнате
}

// ===== FileService: файлы чанками в обе стороны =====
service FileService {
  // Download — server streaming из седьмого урока: имя на входе,
  // поток чанков на выходе, конец — статус OK.
  rpc Download (DownloadRequest) returns (stream Chunk);

  // Upload — client streaming из восьмого: поток чанков на входе,
  // ровно один итог: размер и sha256.
  rpc Upload (stream Chunk) returns (UploadReply);
}

message DownloadRequest {
  string name = 1; // имя файла в хранилище
}

message Chunk {
  bytes data = 1; // ломтик файла, до 64 КБ
  string name = 2; // имя файла; заполняется в первом чанке Upload
}

message UploadReply {
  int64 size = 1;  // суммарный размер принятых байт
  string hash = 2; // sha256 в hex, посчитан по мере приёма
}

Один файл, один пакет, два service-блока — protoc и buf съедают такую схему за один прогон и кладут оба стаба рядом. Заметьте распределение ролей у полей: Chunk.name заполняется только в Upload, потому что клиенту надо сказать, куда класть файл, а Download передаёт имя в DownloadRequest — поле простаивает в другом направлении, и это нормально для общего типа. Presence.online — int32: счётчик людей не бывает отрицательным и не дотягивает до int64, а varint из девятнадцатого урока сделает маленькие числа дешёвыми на проводе.

Как читать схему: три кардинальности и один oneof#

Схема сознательно собирает все три стриминговые модели курса в один контракт. Join — двунаправленный стрим из девятого урока: stream слева и справа от returns, у клиента Send, Recv и CloseSend по отдельности, у сервера ни req, ни ctx — стрим забрал всё; io.EOF от Recv на сервере значит, что клиент вызвал CloseSend и закончил говорить, а возврат из хендлера закрывает серверную половину. Download — server streaming из седьмого: один запрос, поток ответов, клиент читает Recv в цикле до io.EOF. Upload — client streaming из восьмого: клиент шлёт чанки и закрывает склейкой CloseAndRecv, сервер копит агрегат и отдаёт итог через SendAndClose. Unary не участвует вовсе — и это повод перечитать шестой урок: если бы схеме понадобился GetRoomInfo, он встал бы сюда без перестройки остального. Теперь самое важное проектное решение: почему чат — один метод, а не три разных Join, SendMessage и Subscribe. Такая декомпозиция возможна: SendMessage был бы unary, Subscribe — server streaming, и это даже выглядит привычнее. Но тогда членство в комнате перестаёт быть свойством соединения: Subscribe возвращает поток, а сервер вынужден хранить соответствие «пользователь — комнаты» отдельно от соединений и чистить его таймаутами. В bidi-дизайне состояние равно жизни стрима: соединение оборвалось — участник исчез из комнаты сам, presence посчитался сам, а отмена контекста — одна точка уничтожения. Плата честная: все события клиента идут по одному потоку, а подписаться на вторую комнату нельзя без второго вызова Join. Для чата поддержки это правильный компромисс; для рассылки тысяч комнат одному окну уже нет.

Каркас: два сервиса на одном сервере

// internal/server/app.go — каркас итогового проекта.
type chatServer struct {
        pb.UnimplementedChatServiceServer
        mu    sync.Mutex
        rooms map[string]map[*member]struct{} // имя комнаты -> её участники
}

// member — один подписчик внутри комнаты: очередь исходящего
// и имя для presence.
type member struct {
        send chan *pb.ChatEvent
        name string
        once sync.Once // close(send) ровно один раз
}

func (m *member) shutdown() {
        m.once.Do(func() { close(m.send) }) // двойной close — паника
}

type fileServer struct {
        pb.UnimplementedFileServiceServer
        dir string // корень файлового хранилища
}

func main() {
        lis, err := net.Listen("tcp", ":50051") // порт из четвёртого урока
        if err != nil {
                log.Fatal(err)
        }
        srv := grpc.NewServer(
                // Цепочка из одиннадцатого урока: токен, затем recovery.
                grpc.ChainStreamInterceptor(authStream, recoveryStream),
        )
        // Оба сервиса — один сервер, один порт, один health-статус.
        pb.RegisterChatServiceServer(srv, &chatServer{
                rooms: map[string]map[*member]struct{}{},
        })
        pb.RegisterFileServiceServer(srv, &fileServer{dir: "store"})
        reflection.Register(srv) // grpcurl видит оба сервиса
        // Health из семнадцатого урока: отдельный статус на каждый сервис.
        healthSrv := health.NewServer()
        grpc_health_v1.RegisterHealthServer(srv, healthSrv)
        healthSrv.SetServingStatus("shop.v1.ChatService",
                healthpb.HealthCheckResponse_SERVING)
        healthSrv.SetServingStatus("shop.v1.FileService",
                healthpb.HealthCheckResponse_SERVING)
        log.Fatal(srv.Serve(lis))
}

Встроивание UnimplementedChatServiceServer и UnimplementedFileServiceServer — тот самый приём совместимости из четвёртого урока: пока метод не реализован, клиент получает Unimplemented, а схема может расти впереди кода. Health-сервис из семнадцатого урока регистрируется с именами обоих сервисов — пробы k8s и Envoy из двадцать четвёртого урока смогут спросить статус каждого по отдельности. ChainStreamInterceptor — потому что все три метода проекта стриминговые: unary-перехватчик здесь не вызовется ни разу, и держать его просто негде.

ChatService: hub, комнаты и presence#

Сердце чата — hub: map, где ключ — имя комнаты, а значение — множество указателей на member. Member — это очередь исходящих событий с буфером и горутина-писатель, единственная, кому разрешено звать stream.Send. Почему нельзя рассылать прямо из цикла Recv, объясняет flow control из седьмого и двадцать второго уроков: Send блокируется, когда окно потока получателя переполнено, — если слать в цикле, один медленный клиент остановит обработку входящих у этого участника, а с ними и его presence-рассылку для всех. Буфер канала сглаживает пики, а если он переполнен, вступает правило select с default: медленный подписчик выпадает из комнаты немедленно — лучше потерять отставшего, чем тормозить остальных. Вторая тонкость — закрытие очереди. Close на канале паникует при повторе, а закрыть его нужно и при выходе участника, и при выбрасывании из broadcast, поэтому у member есть sync.Once: shutdown срабатывает ровно один раз, кто бы его ни позвал первым. Presence — это те же ChatEvent, только с Presence внутри oneof: enter шлёт «вошёл, теперь нас N», leave — «вышел, осталось N», и оба события получают все участники, включая самого героя события. Комната создаётся автоматически при первом enter — отдельного «создать комнату» в контракте нет, и это осознанная простота: имя комнаты — просто ключ в map.

Join: вход, чтение, рассылка, выход

// internal/server/chat.go — bidi-хендлер из девятого урока, выросший в чат.
func (s *chatServer) Join(stream pb.ChatService_JoinServer) error {
        // Первое сообщение выбирает комнату: это и есть «войти».
        first, err := stream.Recv()
        if err != nil {
                return err // клиент умер до входа: статус уже записан
        }
        msg := first.GetMsg() // nil-безопасный геттер из второго урока
        if msg == nil || msg.GetRoom() == "" {
                return status.Error(codes.InvalidArgument, "укажите комнату")
        }
        room := msg.GetRoom()
        m := &member{send: make(chan *pb.ChatEvent, 16), name: msg.GetAuthor()}
        s.enter(room, m)          // внутри — presence «вошёл»
        defer s.leave(room, m)    // внутри — presence «вышел»

        // Писатель: единственный владелец Send. Горутина гаснет,
        // когда очередь закрыта или стрим сломан.
        writerDone := make(chan struct{})
        go func() {
                defer close(writerDone)
                for ev := range m.send {
                        if stream.Send(ev) != nil {
                                return // получатель отстал или ушёл
                        }
                }
        }()

        for {
                in, err := stream.Recv()
                switch {
                case err == io.EOF:
                        // Клиент вызвал CloseSend: закончил говорить.
                        m.shutdown()  // дописываем остаток очереди
                        <-writerDone  // дожидаемся писателя
                        return nil    // return nil — статус OK, как в 09
                case err != nil:
                        // Обрыв, отмена, дедлайн: статус записан библиотекой.
                        m.shutdown()
                        <-writerDone
                        return err
                }
                if in.GetMsg() != nil {
                        s.broadcast(room, in) // presence от клиента игнорируем
                }
        }
}

// enter/leave меняют состав комнаты и объявляют об этом всем.
func (s *chatServer) enter(room string, m *member) {
        s.mu.Lock()
        defer s.mu.Unlock()
        if s.rooms[room] == nil {
                s.rooms[room] = make(map[*member]struct{})
        }
        s.rooms[room][m] = struct{}{}
        s.broadcast(room, presence(room, m.name, true, len(s.rooms[room])))
}

func (s *chatServer) leave(room string, m *member) {
        s.mu.Lock()
        defer s.mu.Unlock()
        if _, ok := s.rooms[room][m]; ok {
                delete(s.rooms[room], m)
                m.shutdown()
                s.broadcast(room, presence(room, m.name, false, len(s.rooms[room])))
        }
}

// broadcast кладёт событие всем; отставший с переполненной
// очередью выпадает из комнаты, а не тормозит остальных.
func (s *chatServer) broadcast(room string, ev *pb.ChatEvent) {
        s.mu.Lock()
        defer s.mu.Unlock()
        for m := range s.rooms[room] {
                select {
                case m.send <- ev:
                default:
                        delete(s.rooms[room], m)
                        m.shutdown()
                }
        }
}

Это хендлер ChatTickets из девятого урока, у которого появилось состояние. Обратите внимание на порядок разборов: io.EOF — не ошибка, а вежливое «я закончил»; прочая ошибка Recv уже несёт записанный библиотекой статус — возвращаем как есть. Два вызова shutdown не конфликтуют: once гарантирует один close. Заметьте и честное ограничение: выброшенный за медлительность участник остаётся подключённым, но ничего не получает; в бою у member хранят ещё и cancel-функцию стрима, чтобы тут же оборвать его вызов — оставляем это домашним упражнением.

Проверка: медленный подписчик

В первой версии чата сервер рассылал события прямо из цикла Recv, без очередей и писателей. Что произойдёт, когда один из участников повиснет, и что именно чинит отдельная очередь у каждого member?

FileService: Download чанками#

Файловая половина начинается с вопроса, который задаёт каждый, кто впервые шлёт файл по gRPC: почему нельзя просто отправить bytes размером с файл? Потому что лимит. В исходниках grpc-go сервер по умолчанию объявлен так: defaultServerMaxReceiveMessageSize = 1024 1024 4 — четыре мебибайта на сообщение; попытка прислать больше закончится ResourceExhausted со словами о превышении максимума. Чанк в 64 КБ — 65536 байт — берёт запас в шестьдесят четыре раза от лимита: заголовки, обвязка и будущее росту полей помещаются свободно, память на буфер фиксирована, а больших сообщений просто не возникает. Это ручка из двадцать второго урока, которой мы сознательно не пользуемся: поднять лимит проще, чем объяснить OOM на объекте в гигабайт. Второй вопрос — чем кончается поток. Конец файла — это конец стрима: сервер возвращает nil из хендлера, статус OK, клиент видит io.EOF из Recv — ровно семантика седьмого урока с WatchOrders. Клиент даже не знает размер заранее: он читает, пока читается, и складывает байты сам — если нужен прогресс-бар, сервер может досылать размер отдельным полем, но базовый протокол об этом не спрашивает. И последняя деталь, на которой спотываются: буфер чтения переиспользуется, поэтому в Chunk уходит копия — append([]byte(nil), buf[:n]...). Документация не обещает, что библиотека скопирует ваши байты при Send; копия на месте стоит копейки и снимает вопрос навсегда.

Download: поток чанков по 64 КБ

// internal/server/file.go — server streaming из седьмого урока.
const chunkSize = 64 * 1024 // 65536 байт: запас от лимита 4 МиБ

func (s *fileServer) Download(req *pb.DownloadRequest,
        stream pb.FileService_DownloadServer) error {
        if req.GetName() == "" {
                // Пустое имя — вина клиента: код из тринадцатого урока.
                return status.Error(codes.InvalidArgument, "пустое имя файла")
        }
        f, err := os.Open(filepath.Join(s.dir, req.GetName()))
        if os.IsNotExist(err) {
                return status.Errorf(codes.NotFound,
                        "файла %q нет в хранилище", req.GetName())
        }
        if err != nil {
                return status.Error(codes.Internal, "хранилище недоступно")
        }
        defer f.Close()

        buf := make([]byte, chunkSize)
        for {
                n, rerr := f.Read(buf)
                if n > 0 {
                        // Копия обязательна: buf переиспользуется
                        // на следующей итерации чтения.
                        chunk := &pb.Chunk{Data: append([]byte(nil), buf[:n]...)}
                        if serr := stream.Send(chunk); serr != nil {
                                return serr // клиент ушёл — стрим закончен
                        }
                }
                if rerr == io.EOF {
                        return nil // весь файл ушёл: клиент увидит io.EOF
                }
                if rerr != nil {
                        return status.Error(codes.Internal, "чтение не удалось")
                }
        }
}

Три статус-кода на двадцать строк: InvalidArgument на пустое имя — ошибка запроса, NotFound на отсутствующий файл — ошибка данных, Internal на сбой хранилища — ошибка сервера; это ровно словарь тринадцатого урока, где коды разделены на семейства приложения и инфраструктуры. Цикл чтения переживает неудобный случай «n > 0 при err != nil»: сначала отправляем прочитанное, потом разбираемся с ошибкой. Ошибка Send означает, что получатель исчез — хендлер обязан выйти; продолжать читать файл без получателя бессмысленно.

Проверка: сколько придёт чанков

Прочитайте код и выберите, что он выведет:

// Клиент Download. На сервере лежит файл в 130000 байт,
// сервер шлёт чанками по 65536, как в уроке.
var chunks, last int
for {
        chunk, err := stream.Recv()
        if err == io.EOF {
                break
        }
        if err != nil {
                log.Fatal(err)
        }
        chunks++
        last = len(chunk.GetData())
}
fmt.Println(chunks, last)

FileService: Upload и агрегация на лету#

Upload — зеркало Download и прямой наследник восьмого урока: клиент шлёт чанки Send-ами, закрывает сторону склейкой CloseAndRecv и получает ровно один ответ. Серверная сторона — учебник по агрегации в потоке: вместо накопления файла в памяти размер и хеш считаются по мере прихода байт. size += int64(len(chunk.GetData())) — счётчик; h.Write(chunk.GetData()) — байты капают в sha256.New(), который и не подозревает, что его зовут из gRPC-хендлера: hash.Hash — обычный io.Writer, и стриминговый приём из восьмого урока склеивается со стандартной библиотекой без единого адаптера. К моменту io.EOF итог готов целиком: SendAndClose уезжает с размером и hex-строкой хеша. Сравните с UploadOrderEvents из восьмого урока: там агрегат был счётчиком событий, здесь — криптографический хеш, но скелет один: Recv-цикл, io.EOF как легальный финал, ровно одно ответное сообщение. Ранняя валидация тоже на месте: имя файла прилетает в первом чанке — если до io.EOF дошёл поток без имени, клиент получает InvalidArgument, не успев создать мусорный файл. Хеш здесь не украшение: он станет ключом проверки в bufconn-тесте — загрузили, скачали, сверили — и честным контрактом для боевого клиента, который может проверить целостность без повторного чтения.

Upload: размер и sha256 без памяти на файл

// internal/server/file.go — client streaming из восьмого урока.
func (s *fileServer) Upload(stream pb.FileService_UploadServer) error {
        var name string
        var f *os.File
        h := sha256.New() // hash.Hash — обычный io.Writer
        var size int64
        defer func() {
                if f != nil {
                        f.Close()
                }
        }()
        for {
                chunk, err := stream.Recv()
                switch {
                case err == io.EOF:
                        // Клиент закрыл поток склейкой CloseAndRecv.
                        if f == nil {
                                return status.Error(codes.InvalidArgument,
                                        "имя файла не задано")
                        }
                        // Итог ровно один: SendAndClose из восьмого урока.
                        return stream.SendAndClose(&pb.UploadReply{
                                Size: size,
                                Hash: hex.EncodeToString(h.Sum(nil)),
                        })
                case err != nil:
                        return err // обрыв, отмена, дедлайн
                }
                if chunk.GetName() != "" && f == nil {
                        name = chunk.GetName()
                        var cerr error
                        if f, cerr = os.Create(filepath.Join(s.dir, name)); cerr != nil {
                                return status.Error(codes.Internal,
                                        "не удалось создать файл")
                        }
                }
                if f == nil {
                        return status.Error(codes.InvalidArgument,
                                "первый чанк без имени")
                }
                size += int64(len(chunk.GetData()))
                h.Write(chunk.GetData())
                if _, werr := f.Write(chunk.GetData()); werr != nil {
                        return status.Error(codes.Internal, "запись не удалась")
                }
        }
}

Переменная name нужна только для читаемости логов — файл создаётся в первом чанке с именем и живёт до конца вызова. Частичная загрузка после обрыва останется на диске — это честная дыра учебной версии; боевой приём — писать во временный файл и переименовывать после успешного io.EOF, потому что rename на той же файловой системе атомарен. Заметьте: если клиент прервёт загрузку, defer закроет дескриптор, а хеш и размер просто пропадут — ответа у вызова не будет, есть только статус.

Проверка: что вернёт UploadReply

Прочитайте код и выберите, что он выведет:

// Клиент Upload, как в восьмом уроке.
up, err := files.Upload(ctx)
if err != nil {
        log.Fatal(err)
}
up.Send(&pb.Chunk{Name: "report.bin", Data: []byte("hello")})
up.Send(&pb.Chunk{Data: []byte("world!")})
reply, err := up.CloseAndRecv()
if err != nil {
        log.Fatal(err)
}
fmt.Println(reply.GetSize())

Проверка: где живёт политика доступа

В проекте Download открыт анонимам, а Join и Upload требуют Bearer-токен. Где этому правилу правильнее всего жить в коде?

Метаданные и перехватчик: токен в одном месте#

Токен едет вне сообщений — в metadata, как учил десятый урок: клиент добавляет его через metadata.AppendToOutgoingContext с ключом authorization, сервер достаёт через metadata.FromIncomingContext, и оба помнят, что ключи metadata всегда строчные. Приятная деталь: официальный пример запроса из спеки gRPC несёт ровно нашу пару — grpc-timeout = 1S и authorization = Bearer y235.wef315yfh138vh31hv93hv8h3v. Токен и срок — заголовки рядом с сообщениями, и спека задумывала их соседями с самого начала. Проверка — стримовый перехватчик из одиннадцатого урока: сигнатура с StreamServerInfo, внутри metadata, решение по info.FullMethod и вызов handler(srv, ss) в конце — или статус вместо него. Разница между Unauthenticated и PermissionDenied из тринадцатого урока работает как надо: нет токена — Unauthenticated, «кто ты, мы не знаем»; токен есть, но не принят — PermissionDenied, «мы тебя знаем, но нельзя». Это не педантизм: по первому коду разумный клиент предлагает логин, по второму — не предлагает ничего. И заметьте асимметрию политики: перехватчик пускает Download без токена, потому что чтение файлов публично; полный адрес метода — /shop.v1.FileService/Download — отличает его от Upload того же сервиса, и никакой другой маркер для этого не нужен.

authStream: токен проверяет один перехватчик

// internal/server/auth.go — стримовый перехватчик из одиннадцатого урока.
func authStream(srv interface{}, ss grpc.ServerStream,
        info *grpc.StreamServerInfo, handler grpc.StreamHandler) error {
        // Чтение файлов публично: полный адрес метода решает всё.
        if info.FullMethod == "/shop.v1.FileService/Download" {
                return handler(srv, ss)
        }
        md, ok := metadata.FromIncomingContext(ss.Context())
        if !ok {
                return status.Error(codes.Unauthenticated, "нет metadata")
        }
        vals := md.Get("authorization") // ключи metadata всегда строчные
        if len(vals) == 0 || !strings.HasPrefix(vals[0], "Bearer ") {
                return status.Error(codes.Unauthenticated, "нет Bearer-токена")
        }
        if vals[0] != "Bearer secret" {
                // Токен есть, но не принят: другой код, другая реакция клиента.
                return status.Error(codes.PermissionDenied, "токен не принят")
        }
        return handler(srv, ss) // пускаем хендлер дальше по цепочке
}

Строковый сверочный токен — учебный: в бою здесь стоит проверка подписи JWT или поход к провайдеру, но форма перехватчика не изменится ни на символ. Ключевой ход урока — решение по info.FullMethod: адрес метода это уникальный ключ, который gRPC и так знает, и политика доступа становится одной читаемой функцией. Recovery-перехватчик из одиннадцатого урока — второй в ChainStreamInterceptor: паника хендлера чата не уронит процесс, клиент получит Internal, а комната останется жить.

Дедлайны и отмена: у стрима тоже есть срок#

Четырнадцатый урок учил дедлайны на unary — здесь они встречаются со стримами, и результат не всегда очевиден. Дедлайн, поставленный клиентом до вызова, едет на проводе заголовком grpc-timeout: спека описывает его как Timeout → grpc-timeout TimeoutValue TimeoutUnit, где значение — «positive integer as ASCII string of at most 8 digits», и добавляет дословно: «If Timeout is omitted a server should assume an infinite timeout» — если срок не задан, сервер должен считать его бесконечным. Для Upload и Download бесконечный срок — путь к зависшему вызову: диск застрял, сеть молчит — клиент ждёт вечно. Поэтому файловые вызовы получают context.WithTimeout до вызова — тридцать секунд на типовой файл, и остаток срока честно сжимается на сервере. С Join всё наоборот: чат по своей природе долгоживущий, дедлайн в пять минут убьёт разговор посреди фразы. Решение — срок не ставить вовсе (спека это прямо разрешает), а управлять жизнью через отмену: клиент закрыл соединение или context.WithCancel погасили — сервер увидит ошибку на Recv, presence посчитает выход, комната опустеет. На серверной стороне тот же контекст живёт в stream.Context(): боевой хендлер чата выбирает в select между Recv и ctx.Done(), чтобы не висеть на Recv, когда горутина-писатель уже упала. Отмена посреди Upload — отдельная история: частичный файл остаётся на диске, и лучшее, что можно сделать, — писать во временный файл с переименованием после успеха; это домашняя работа по этому уроку.

Проверка: отмена посреди загрузки

Клиент вызвал cancel() посреди Upload, три чанка из пяти уже уехали. Что увидит серверный хендлер и что станет с частичным файлом в учебной версии?

Соберите клиент чата: токен, срок, вход

Клиентская сторона Join: поставьте срок контекста из четырнадцатого урока, добавьте токен в metadata из десятого и завершите разговор как учит девятый.

// cmd/chat/main.go — клиент чата: токен, срок, вход в комнату. ctx, cancel := context.(context.Background(), 30*time.Second) defer cancel() // Ключи metadata всегда строчные — правило десятого урока. callCtx := metadata.AppendToOutgoingContext(ctx, "", "Bearer "+token) stream, err := client.Join(callCtx) if err != nil { log.Fatal(err) } // Первое сообщение выбирает комнату и объявляет нас. stream.Send(&pb.ChatEvent{Payload: &pb.ChatEvent_Msg{ Msg: &pb.ChatMsg{Room: "общий", Author: "аня", Text: "привет"}, }}) // ... цикл чтения и отправки ... // Закончили говорить: закрыть свою сторону стрима. stream.()

bufconn: интеграционные тесты проекта#

Итоговый проект без теста — просто рабочая версия. Двадцать первый урок оставил нам bufconn: слушатель, у которого «Listen returns a Listener that can only be contacted by its own Dialers and creates buffered connections between the two», — сеть без сети, по которой едет настоящий HTTP/2 со всей сериализацией и статусами. Каркас теста тот же: сервер поднимается на bufconn.Listen, клиент коннектится через grpc.NewClient с passthrough:///bufnet и WithContextDialer, уборка живёт в t.Cleanup — и добавляется одна проектная деталь: на тестовом сервере регистрируются оба сервиса разом, потому что тестируем не методы поодиночке, а их совместную жизнь в одном процессе. Самый ценный тест проекта — roundtrip: загрузить файл двумя чанками, сверить ответ Upload с ожидаемым размером, скачать обратно, склеить байты и сравнить с исходной строкой. Этот тест ловит сразу весь класс ошибок, который не виден поодиночке: перепутанный порядок полей Chunk, потерянную копию буфера, сломанную агрегацию, забытый SendAndClose. Чего bufconn не проверит — и не притворяется: TLS-рукопожатие из двенадцатого урока, настоящую сеть с её обрывами, лимиты транспорта из двадцать второго и балансировку из шестнадцатого. Для дымовой проверки живого сервера есть grpcurl из восемнадцатого: reflection видит оба сервиса, и вызов Download с дедлайном через -d делается за минуту без единой строки клиента.

Roundtrip-тест: Upload → Download со сверкой

// internal/server/app_test.go — bufconn из двадцать первого урока.
const bufSize = 1024 * 1024

func newTestConn(t *testing.T) *grpc.ClientConn {
        t.Helper()
        lis := bufconn.Listen(bufSize)
        srv := grpc.NewServer(grpc.ChainStreamInterceptor(authStream))
        pb.RegisterChatServiceServer(srv, &chatServer{
                rooms: map[string]map[*member]struct{}{},
        })
        pb.RegisterFileServiceServer(srv, &fileServer{dir: t.TempDir()})
        go srv.Serve(lis) // без go тест зависнет навсегда
        t.Cleanup(srv.Stop)
        conn, err := grpc.NewClient(
                "passthrough:///bufnet", // схема passthrough обязательна
                grpc.WithContextDialer(func(ctx context.Context, _ string) (net.Conn, error) {
                        return lis.DialContext(ctx)
                }),
                grpc.WithTransportCredentials(insecure.NewCredentials()),
        )
        if err != nil {
                t.Fatal(err)
        }
        t.Cleanup(func() { conn.Close() })
        return conn
}

func TestUploadDownloadRoundtrip(t *testing.T) {
        conn := newTestConn(t)
        files := pb.NewFileServiceClient(conn)
        ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
        defer cancel()

        // Загружаем файл двумя чанками — только Upload требует токен.
        upCtx := metadata.AppendToOutgoingContext(ctx,
                "authorization", "Bearer secret")
        up, err := files.Upload(upCtx)
        if err != nil {
                t.Fatal(err)
        }
        up.Send(&pb.Chunk{Name: "report.bin", Data: []byte("hello")})
        up.Send(&pb.Chunk{Data: []byte("world!")})
        reply, err := up.CloseAndRecv()
        if err != nil {
                t.Fatal(err)
        }
        if reply.GetSize() != 11 {
                t.Fatalf("size = %d; want 11", reply.GetSize())
        }

        // Скачиваем и склеиваем: Download анонимен.
        stream, err := files.Download(ctx, &pb.DownloadRequest{Name: "report.bin"})
        if err != nil {
                t.Fatal(err)
        }
        var got []byte
        for {
                chunk, err := stream.Recv()
                if err == io.EOF {
                        break
                }
                if err != nil {
                        t.Fatal(err)
                }
                got = append(got, chunk.GetData()...)
        }
        if string(got) != "helloworld!" {
                t.Fatalf("got %q; want %q", got, "helloworld!")
        }
}

Тест показывает и асимметрию политики: upCtx несёт токен для Upload, Download идёт с голым ctx — и перехватчик его пускает. fileServer указывает на t.TempDir() из двадцать первого урока: хранилище создаётся и убирается само. Чанки «hello» и «world!» могут вернуться одним сообщением или двумя — протокол не обещает сохранить границы Send, и тест этого не проверяет: склеиваем байты, а не коробки.

Соберите проект: от схемы до прокси

Расставьте шаги в том порядке, в котором итоговый проект собирается и уезжает в бой.

Строки

Ваша программа

Нажимайте на строки слева, чтобы собрать программу

Чек-лист итогового проекта#

Схема: один пакет shop.v1, один файл app.proto, oneof для событий чата, bytes для чанков
Join — единственный bidi-метод: io.EOF от клиента не ошибка, return из хендлера закрывает серверную половину
Каждому подписчику — своя очередь и своя горутина-писатель: Send к медленному клиенту не тормозит комнату
Медленный подписчик с переполненной очередью выпадает по select-default; close(send) — через sync.Once
Чанки по 65536 байт — запас от лимита 4 МиБ по умолчанию; последний чанк — остаток, конец — io.EOF у клиента
Upload агрегирует на лету: size += len и h.Write — файл никогда не лежит в памяти целиком
Токен — в metadata с строчным ключом, проверка — в стримовом перехватчике по FullMethod; Download анонимен
Дедлайны на Upload и Download, для долгого Join — отмена вместо срока; grpc-timeout едет на проводе
Статусы вместо паник: InvalidArgument на пустые поля, NotFound на чужие имена, Unauthenticated и PermissionDenied — раздельно
bufconn-тест с roundtrip Upload → Download и сверкой хеша — страховка после каждого изменения схемы

Выпускник курса: что теперь по силам#

Курс grpc-go завершён: двадцать пять уроков, от первого net.Listen до Envoy за спиной. Что это значит практически. Вы умеете читать чужой proto-файл и видеть в нём кардинальности, эволюционные риски и цену полей на проводе. Вы пишете сервер с перехватчиками, health и reflection, клиент с токенами, дедлайнами и отменами — и покрываете это bufconn-тестами, которые гоняются быстрее сборки. Стриминговые модели в чистом виде — server streaming и bidi — вы уже применяли по отдельности; итоговый проект показал их в одном контракте с настоящим состоянием. Развертывание — двадцать четвёртый урок: конфиг прокси, ingress и health-пробы принимают ваш проект без правок кода. Куда дальше: весь курс целиком теперь работает как справочник — возвращайтесь к нему, когда заводите новый сервис. Если встанет браузерный клиент, пересмотрите связку gRPC-Web в connect-курсе — там тот же контракт живёт в другом стеке с тремя протоколами на одном хендлере. И последнее: итоговый grpc проект этого урока — каркас, а не витрина. Замените фиксированный токен на JWT-провайдера, in-memory хаб на Redis или NATS, диск на объектное хранилище — и кардинальности, перехватчики, дедлайны и тесты останутся теми же. Это и есть главный итог: механика выучена, скелет собран, и он переживёт замену всех деталей.

Карточки финального урока

Термин

bidi-чат одним методом

нажмите, чтобы перевернуть

Определение

Join (stream ChatEvent) returns (stream ChatEvent): клиентские сообщения — SendMessage, серверный поток — Subscribe. Членство в комнате равно жизни стрима: оборвалось соединение — участник исчез, presence пересчитался сам. io.EOF от Recv — клиент вызвал CloseSend; return из хендлера закрывает серверную половину.

1 / 6

Песочница урока

Редактируйте код и запускайте его прямо здесь — интерпретатор работает в браузере, ничего устанавливать не нужно. За каждый запуск — +2 XP.

Вывод

Запустите код, чтобы увидеть вывод.

Коротко: что запомнить

Итоговый проект собрал курс в один proto: ChatService на bidi-стриме из девятого урока и FileService на server- и client-streaming из седьмого и восьмого — три кардинальности из четырёх, unary не понадобился.
bidi-чат живёт одним методом Join: членство в комнате равно жизни стрима, у каждого подписчика своя очередь с горутиной-писателем, медленные читатели выпадают по select-default, а не тормозят комнату.
Файлы едут чанками по 65536 байт — запас от лимита 4 МиБ: Download отдаёт поток, Upload агрегирует размер и sha256 на лету, итог уезжает одним SendAndClose.
Токен живёт в metadata, проверка — в стримовом перехватчике по FullMethod: хендлеры не знают про authorization, политика видна в одном месте, Unauthenticated и PermissionDenied различаются.
Дедлайны и отмена из четырнадцатого урока накрывают стримы: срок на файловые вызовы, отмена вместо срока для долгого чата, частичная загрузка — повод для временного файла с атомарным переименованием.
bufconn-тест с roundtrip Upload → Download и сверкой хеша — самая дешёвая страховка проекта; дальше — прокси из двадцать четвёртого урока, и grpc проект уезжает в прод без правок бизнес-кода.

Домашнее задание

Доведите проект урока до состояния «можно показывать». Первое: научите чат признавать пользователей — токен несёт имя, enter пишет его в presence, а перехватчик отклоняет повторный вход под тем же именем в ту же комнату кодом PermissionDenied. Второе: добавьте истории последних ста сообщений на комнату и досылайте их новому участнику сразу после его presence — проследите, чтобы это не столкнуло enter и broadcast в гонке. Третье: реализуйте атомарную загрузку — Upload пишет во временный файл и переименовывает его после успешного io.EOF; добавьте bufconn-тест, что после отмены посреди загрузки старый файл не повреждён. Четвёртое: напишите табличный тест отказов: Upload без токена — Unauthenticated, Download несуществующего файла — NotFound, Upload с пустым именем — InvalidArgument. В конце выпишите, что произойдёт с клиентами при добавлении полей mime в Chunk и room_created в Presence, и какой урок курса отвечает за этот ответ.

Вопросы и ответы по уроку

Почему чат сделан одним bidi-методом, а не SendMessage и Subscribe отдельно?

Это два легальных дизайна. Раздельный — SendMessage unary и Subscribe server streaming — привычнее, но членство в комнате перестаёт быть свойством соединения: серверу приходится хранить карту «пользователь — комнаты» отдельно и чистить её таймаутами. Bidi-дизайн делает состояние равным жизни стрима: обрыв соединения сам убирает участника, presence сам пересчитывается, отмена контекста — одна точка уничтожения. Плата: события клиента идут по одному потоку и вторая комната требует второго вызова Join. Для чата поддержки и мессенджера это правильный компромисс; для массовой подписки одного клиента на тысячи комнат — нет.

Почему чанки именно 64 КБ и что будет с файлом больше четырёх мебибайт?

64 КБ — это 65536 байт, в 64 раза меньше дефолтного лимита сообщения grpc-go в 4 МиБ: запас покрывает обвязку и будущее полей, буфер чтения фиксирован, больших сообщений не возникает. Файл любого размера — просто большее количество чанков: сто мегабайт — около 1600 вызовов Send, и лимит сообщения тут ни при чём. Поднимать лимит из двадцать второго урока стоит только когда чанк-модель неудобна, а большие сообщения осознаны: они держат буферы и окно потока дольше.

Что делать клиенту чата при обрыве соединения?

Ретраи из пятнадцатого урока стрим не переиграют: retryPolicy работает на unary, а bidi-вызов — это долгоживущее соединение. Клиент переподключается новым вызовом Join, сервер видит обрыв старого, presence объявляет выход и вход — участники переживут небольшую деградацию. Идемпотентность на стороне клиента: повторная отправка сообщения после обрыва может продублировать его соседям, поэтому боевые чаты держат идентификаторы сообщений и дедупликацию на приёме. Липкость из шестнадцатого урока помогает, пока жив соединённый под, но от обрыва сети не спасает.

Чем этот проект отличается от боевого мессенджера?

Честный список: комнаты и история живут в памяти и умрут с процессом; токен сверяется строкой, а не подписью JWT; загрузка без отмены оставляет частичные файлы; доставка — at-most-once, без подтверждений и дедупликации; масштабирование в несколько подов потребует вынести хаб из памяти в шину вроде Redis или NATS. Каркас при этом тот же: контракты, кардинальности, перехватчик, дедлайны, статусы и bufconn-тесты не изменятся при замене ни одной из этих деталей — именно поэтому проект и строился вокруг схемы, а не вокруг хранилищ.

Термины урока

Ключевые понятия из этого урока — короткие определения в глоссарии. Наведите на термин, чтобы вспомнить, откройте карточку для подробностей и ссылки на практику.

Сравните с альтернативами

Практический вопрос после первых уроков: а почему именно эта библиотека? Мы разобрали конкурентов по одним критериям, с кодом одной задачи рядом и персональным подбором под ваши приоритеты.