Compare commits
3
Commits
11f14108b4
...
b8b55cd7ab
| Author | SHA1 | Date | |
|---|---|---|---|
 syakovlev
|
b8b55cd7ab | ||
 osukhoroslov
|
a5011dd94b | ||
|
syakovlevandosukhoroslov
|
43f49f4b2a |
@@ -0,0 +1,5 @@
|
||||
target
|
||||
**/target
|
||||
**/__pycache__
|
||||
**/.pytest_cache
|
||||
*.pyc
|
||||
@@ -0,0 +1,125 @@
|
||||
x-tests-image: &tests-image
|
||||
image: grpc-solution-tests
|
||||
build:
|
||||
context: .
|
||||
dockerfile: tests/Dockerfile
|
||||
|
||||
services:
|
||||
server:
|
||||
image: grpc-solution-server
|
||||
build:
|
||||
context: ./solution
|
||||
dockerfile: server.dockerfile
|
||||
environment:
|
||||
MESSENGER_SERVER_PORT: 51075
|
||||
ports:
|
||||
- "${MESSENGER_SERVER_HOST_PORT:-51075}:51075"
|
||||
# Start the clients first to exercise their reconnect logic during manual runs.
|
||||
depends_on:
|
||||
- client1
|
||||
- client2
|
||||
|
||||
server-tests:
|
||||
image: grpc-solution-server
|
||||
build:
|
||||
context: ./solution
|
||||
dockerfile: server.dockerfile
|
||||
environment:
|
||||
MESSENGER_SERVER_PORT: 51075
|
||||
|
||||
client1:
|
||||
image: grpc-solution-client
|
||||
build:
|
||||
context: ./solution
|
||||
dockerfile: client.dockerfile
|
||||
environment:
|
||||
MESSENGER_SERVER_ADDR: server:51075
|
||||
MESSENGER_HTTP_PORT: 8080
|
||||
ports:
|
||||
- "${MESSENGER_CLIENT1_HOST_PORT:-8080}:8080"
|
||||
|
||||
client2:
|
||||
image: grpc-solution-client
|
||||
build:
|
||||
context: ./solution
|
||||
dockerfile: client.dockerfile
|
||||
environment:
|
||||
MESSENGER_SERVER_ADDR: server:51075
|
||||
MESSENGER_HTTP_PORT: 8080
|
||||
ports:
|
||||
- "${MESSENGER_CLIENT2_HOST_PORT:-8081}:8080"
|
||||
|
||||
client-test-server:
|
||||
image: "${MESSENGER_CLIENT_TEST_SERVER_IMAGE:-distsys.ru/course/grpc-messenger:latest}"
|
||||
entrypoint:
|
||||
- /usr/local/bin/client-test-server
|
||||
environment:
|
||||
MESSENGER_CLIENT_TEST_SERVER_EXPECTED_STREAMS: 2
|
||||
MESSENGER_CLIENT_TEST_SERVER_PROTO: /submission/proto/messenger.proto
|
||||
MESSENGER_SERVER_PORT: 51075
|
||||
volumes:
|
||||
- type: bind
|
||||
source: "${MESSENGER_CLIENT_TEST_SERVER_PROTO_DIR:-./solution/proto}"
|
||||
target: /submission/proto
|
||||
read_only: true
|
||||
user: "65532:65532"
|
||||
read_only: true
|
||||
cap_drop:
|
||||
- ALL
|
||||
security_opt:
|
||||
- no-new-privileges:true
|
||||
|
||||
client-test1:
|
||||
image: grpc-solution-client
|
||||
build:
|
||||
context: ./solution
|
||||
dockerfile: client.dockerfile
|
||||
environment:
|
||||
MESSENGER_SERVER_ADDR: client-test-server:51075
|
||||
MESSENGER_HTTP_PORT: 8080
|
||||
depends_on:
|
||||
- client-test-server
|
||||
|
||||
client-test2:
|
||||
image: grpc-solution-client
|
||||
build:
|
||||
context: ./solution
|
||||
dockerfile: client.dockerfile
|
||||
environment:
|
||||
MESSENGER_SERVER_ADDR: client-test-server:51075
|
||||
MESSENGER_HTTP_PORT: 8080
|
||||
depends_on:
|
||||
- client-test-server
|
||||
|
||||
tests:
|
||||
<<: *tests-image
|
||||
environment:
|
||||
MESSENGER_TEST_CLIENT1_ADDR: client-test1:8080
|
||||
MESSENGER_TEST_CLIENT2_ADDR: client-test2:8080
|
||||
MESSENGER_TEST_SERVER_ADDR: server-tests:51075
|
||||
depends_on:
|
||||
- server-tests
|
||||
- client-test1
|
||||
- client-test2
|
||||
|
||||
server-test-runner:
|
||||
<<: *tests-image
|
||||
command:
|
||||
- --component
|
||||
- server
|
||||
environment:
|
||||
MESSENGER_TEST_SERVER_ADDR: server-tests:51075
|
||||
depends_on:
|
||||
- server-tests
|
||||
|
||||
client-tests:
|
||||
<<: *tests-image
|
||||
command:
|
||||
- --component
|
||||
- client
|
||||
environment:
|
||||
MESSENGER_TEST_CLIENT1_ADDR: client-test1:8080
|
||||
MESSENGER_TEST_CLIENT2_ADDR: client-test2:8080
|
||||
depends_on:
|
||||
- client-test1
|
||||
- client-test2
|
||||
@@ -0,0 +1,296 @@
|
||||
# Мессенджер на gRPC
|
||||
|
||||
Опишите gRPC-интерфейс и реализуйте **сервер** и **клиент** мессенджера с одним общим чатом. Сервер и клиент общаются по gRPC, пользователь обращается к клиенту по HTTP.
|
||||
|
||||
У сервера два RPC-метода: `SendMessage` отправляет сообщение в чат, `ReadMessages` открывает подписку на новые сообщения. Сервер должен обрабатывать несколько запросов одновременно, в том числе принимать сообщения при открытых подписках.
|
||||
|
||||
Клиент при запуске открывает подписку и сохраняет сообщения от сервера в буфере в порядке получения. Через HTTP пользователь отправляет сообщения и забирает содержимое буфера.
|
||||
|
||||
На схеме показаны сервер и два клиента. Тесты обращаются к клиентам от имени двух пользователей:
|
||||
|
||||
```mermaid
|
||||
flowchart RL
|
||||
subgraph Tests
|
||||
U1{User 1}
|
||||
U2{User 2}
|
||||
end
|
||||
|
||||
subgraph Clients with HTTP interface
|
||||
C1(fa:fa-comments Client 1)
|
||||
C2(fa:fa-comments Client 2)
|
||||
end
|
||||
|
||||
subgraph gRPC server
|
||||
S(fa:fa-server Server)
|
||||
end
|
||||
|
||||
|
||||
C1 -- SendMessage --> S
|
||||
S -. Stream ReadMessages .-> C1
|
||||
|
||||
U1 -- POST /sendMessage --> C1
|
||||
C1 -- Forward messages in /getAndFlushMessages --> U1
|
||||
|
||||
C2 -- SendMessage --> S
|
||||
S -. Stream ReadMessages .-> C2
|
||||
|
||||
U2 -- POST /sendMessage --> C2
|
||||
C2 -- Forward messages in /getAndFlushMessages --> U2
|
||||
```
|
||||
|
||||
## Требования
|
||||
|
||||
### Доставка сообщений
|
||||
|
||||
- Подписка начинается, когда сервер регистрирует вызов `ReadMessages`, и действует до отмены RPC или закрытия соединения. Сообщения, принятые до регистрации, в подписку не попадают.
|
||||
- Сервер передаёт каждое сообщение ровно один раз во все подписки, активные в момент его принятия, включая подписку отправителя. Восстанавливать историю после разрывов и перезапусков не нужно.
|
||||
- Сообщения, общие для нескольких подписчиков, должны идти в одинаковом порядке во всех их потоках. Клиенты сохраняют этот порядок. Для одновременных вызовов `SendMessage` сервер может выбрать любой порядок.
|
||||
- `sendTime` — серверное время принятия сообщения. В течение одного запуска эти значения должны быть уникальны и строго возрастать в порядке рассылки. Ответ `SendMessage` и сообщение во всех подписках содержат одинаковый `sendTime`.
|
||||
- Успешный ответ `SendMessage` означает, что сервер принял сообщение. Это не гарантирует, что все клиенты уже его получили.
|
||||
|
||||
### HTTP-интерфейс клиента
|
||||
|
||||
Пользователи и тесты обращаются к клиенту через два HTTP-метода.
|
||||
|
||||
При успехе оба метода возвращают HTTP `200` и JSON. Поле `sendTime` — строка в [JSON-формате `google.protobuf.Timestamp`](https://protobuf.dev/reference/protobuf/google.protobuf/#timestamp), например `2025-09-20T10:58:42.665193557Z`.
|
||||
|
||||
```
|
||||
POST /sendMessage
|
||||
Отправляет одно сообщение в общий чат.
|
||||
|
||||
Тело запроса:
|
||||
{
|
||||
"author": "Ivan Ivanov",
|
||||
"text": "Hey guys"
|
||||
}
|
||||
|
||||
Тело ответа:
|
||||
{
|
||||
"sendTime": "..."
|
||||
}
|
||||
```
|
||||
```
|
||||
POST /getAndFlushMessages
|
||||
Возвращает накопленные сообщения в порядке получения и очищает буфер.
|
||||
|
||||
Тело запроса: нет
|
||||
|
||||
Тело ответа:
|
||||
[{
|
||||
"author": "Ivan Ivanov",
|
||||
"text": "Hey guys",
|
||||
"sendTime": "..."
|
||||
},{
|
||||
"author": "Petr Petrov",
|
||||
"text": "Hey Ivan",
|
||||
"sendTime": "..."
|
||||
}]
|
||||
```
|
||||
|
||||
Если буфер пуст, `getAndFlushMessages` сразу возвращает `[]`. Чтение и очистка буфера должны быть атомарными: сообщение, пришедшее во время этой операции, попадает в текущий или следующий ответ. Клиент не должен терять сообщения, выдавать их повторно или менять их порядок.
|
||||
|
||||
### gRPC-интерфейс сервера
|
||||
|
||||
Тесты проверяют сервер отдельно от клиента. Соблюдайте требования к интерфейсу:
|
||||
|
||||
- синтаксис — `proto3`, пакет — `mes_grpc`;
|
||||
- gRPC-сервис `MessengerServer` содержит два метода: `SendMessage` и `ReadMessages`;
|
||||
- `SendMessage` — унарный вызов. Запрос содержит одиночные строковые поля `author` и `text`, ответ — одиночное поле `sendTime` типа `google.protobuf.Timestamp`;
|
||||
- `ReadMessages` принимает один пустой запрос и возвращает поток сообщений. Можно описать свой тип пустого сообщения или взять готовый из библиотеки. Каждое сообщение в потоке содержит одиночные поля `author` и `text` типа `string` и `sendTime` типа `google.protobuf.Timestamp`.
|
||||
|
||||
Все перечисленные поля одного сообщения должны допускать одновременное заполнение.
|
||||
|
||||
Имена типов сообщений и номера полей выберите самостоятельно — тесты их не фиксируют.
|
||||
|
||||
## Оценивание
|
||||
|
||||
За задание можно получить 10 баллов:
|
||||
|
||||
- **2 балла** — протокол `messenger.proto`, проверяется в `test_proto.py`.
|
||||
- **4 балла** — сервер, проверяется в `test_server.py`.
|
||||
- **4 балла** — клиент, проверяется в `test_client.py`.
|
||||
|
||||
В отчёте `solution/readme.md` опишите структуру решения, какие компоненты вы реализовали и как работают методы сервера и клиента. Без отчёта тесты запускаются, но защита не проводится и решение не засчитывается — см. [общие правила сдачи](../readme.md#сдача-решения).
|
||||
|
||||
Баллы за компонент начисляются, только если прошли все его тесты: 2 или 0 за протокол, 4 или 0 за сервер, 4 или 0 за клиент. Значение `SCORE` в выводе тестов — предварительная оценка. Итоговую оценку преподаватель выставляет после защиты с учётом штрафов ниже.
|
||||
|
||||
Система собирает и проверяет сервер и клиент независимо. Если сервер не собирается или не запускается, он получает 0 баллов, но клиент всё равно проверяется, и наоборот. Протокол проверяется отдельно.
|
||||
|
||||
При независимом оценивании клиенты работают со служебным gRPC-сервером, построенным по вашему `messenger.proto`. Поэтому для проверки клиента нужен корректный протокол, но ошибки вашего сервера не влияют на баллы за клиент.
|
||||
|
||||
На защите можно потерять баллы за следующие ошибки:
|
||||
|
||||
- Сервер не может обрабатывать несколько запросов одновременно — 2 балла.
|
||||
- При конкурентном доступе сервер может потерять или продублировать сообщения, выдать их в разном порядке в потоках `ReadMessages` либо нарушить требования к `sendTime` — 2 балла.
|
||||
- Клиент теряет, повторно выдаёт или меняет порядок сообщений из потока `ReadMessages` — 2 балла.
|
||||
|
||||
За ошибки сервера снимаются только баллы за сервер, за ошибки клиента — только баллы за клиент, не больше 4 баллов в каждом случае. Баллы за протокол сохраняются. На защите нужно разобрать предложенный преподавателем сценарий конкурентного выполнения и объяснить по своему коду, почему решение работает правильно.
|
||||
|
||||
## Заготовки для клиента
|
||||
|
||||
В `templates` есть заготовки клиента на Python, Python с asyncio и Go. Официальная заготовка — `messenger-py/client`; она проверена для текущего задания. Остальные заготовки относятся к прошлым версиям задания: их можно использовать, но расхождения с условием нужно исправить самостоятельно. Можно выбрать и другой язык — тесты обращаются к решению через HTTP и gRPC.
|
||||
|
||||
## Порядок выполнения задания
|
||||
|
||||
Выполняйте команды из папки `homework/02-grpc-messenger`. Примеры с переменными окружения написаны для Bash. В Windows используйте WSL или задавайте переменные через PowerShell.
|
||||
|
||||
### Подготовка окружения
|
||||
|
||||
Установите Python 3.12 или новее и Docker по [общей инструкции](../readme.md#настройка-окружения). Для Python-заготовки и локального запуска тестов установите зависимости:
|
||||
|
||||
```bash
|
||||
python3 -m pip install -r templates/messenger-py/client/requirements.txt -r tests/requirements.txt
|
||||
grpcurl -version
|
||||
```
|
||||
|
||||
Если `grpcurl` не найден, установите его по инструкции в разделе «Полезные материалы». На Windows используйте `python` вместо `python3`.
|
||||
|
||||
### Структура проекта
|
||||
|
||||
Разместите решение в папке `solution`. Сохраните пути к трём файлам, которые используют тесты и [docker-compose.yml](docker-compose.yml):
|
||||
|
||||
- `client.dockerfile` — сборка и запуск клиента;
|
||||
- `server.dockerfile` — сборка и запуск сервера;
|
||||
- `proto/messenger.proto` — описание gRPC-интерфейса; дополните начальный файл.
|
||||
|
||||
При сдаче отправляется только папка `solution`. Изменения за её пределами не учитываются.
|
||||
|
||||
Для официальной Python-заготовки скопируйте `templates/messenger-py/client` в `solution/client`, а образец `client.dockerfile` — в `solution/client.dockerfile`. Сервер разместите в `solution/server/server.py`. Заготовка использует пакет `solution` и импорты `from solution.proto import messenger_pb2, messenger_pb2_grpc`. В своей реализации можно выбрать другую структуру, сохранив три обязательных пути выше.
|
||||
|
||||
### Описание и компиляция gRPC-интерфейса
|
||||
|
||||
Опишите сообщения и сервис в `solution/proto/messenger.proto`.
|
||||
|
||||
Сгенерируйте код для выбранного языка с помощью `protoc`:
|
||||
|
||||
```bash
|
||||
# Python
|
||||
python3 -m grpc_tools.protoc -I. --python_out=. --pyi_out=. --grpc_python_out=. solution/proto/messenger.proto
|
||||
|
||||
# Go (после установки protoc и плагинов protoc-gen-go и protoc-gen-go-grpc)
|
||||
protoc -I solution/proto --go_out=solution/proto --go_opt=paths=source_relative --go-grpc_out=solution/proto --go-grpc_opt=paths=source_relative messenger.proto
|
||||
```
|
||||
|
||||
Для Go укажите в `option go_package` путь пакета в вашем модуле; пример есть в Go-заготовке. Закрепите версии генераторов, совместимые с вашей версией Go. Сгенерированные файлы включите в решение или генерируйте при сборке образа с закреплёнными версиями инструментов.
|
||||
|
||||
### Реализация сервера и клиента
|
||||
|
||||
Сервер реализуйте с нуля; заготовки для него нет. Можно использовать потоки или асинхронный код. Открытые подписки не должны мешать обработке других запросов.
|
||||
|
||||
Если используете Python-заготовку клиента, заполните места с пометкой TODO. HTTP-сервер в ней уже реализован. При запуске клиент должен дождаться сервера, открыть `ReadMessages` и принимать сообщения независимо от обработки HTTP-запросов.
|
||||
|
||||
Сервер и клиент должны брать настройки из переменных окружения:
|
||||
|
||||
| Переменная | Назначение |
|
||||
| --- | --- |
|
||||
| `MESSENGER_SERVER_PORT` | Порт gRPC-сервера, по умолчанию `51075` |
|
||||
| `MESSENGER_SERVER_ADDR` | Адрес gRPC-сервера для подключения клиента |
|
||||
| `MESSENGER_HTTP_PORT` | Порт HTTP-интерфейса клиента |
|
||||
|
||||
Сервер и HTTP-интерфейс клиента должны слушать на `0.0.0.0`.
|
||||
|
||||
В `solution/server.dockerfile` и `solution/client.dockerfile` опишите сборку и запуск сервера и клиента. Контекст сборки — папка `solution`. Включите в образы все нужные файлы и зависимости. За образец можно взять Dockerfile заготовки.
|
||||
|
||||
## Тестирование решения
|
||||
|
||||
Публичные тесты проверяют [протокол](tests/test_proto.py), [сервер](tests/test_server.py) и [клиент](tests/test_client.py). Сложные конкурентные сценарии вы разберёте на защите.
|
||||
|
||||
### Полная проверка
|
||||
|
||||
Рекомендуемый запуск в окружении тестирующей системы:
|
||||
|
||||
```bash
|
||||
docker run --privileged --pull always --rm -v ./solution:/hw/solution distsys.ru/course/grpc-messenger:latest
|
||||
```
|
||||
|
||||
Сервер и клиент собираются и проверяются независимо: ошибка сборки одного не мешает проверить другой.
|
||||
|
||||
### Отдельные компоненты через Docker Compose
|
||||
|
||||
```bash
|
||||
docker compose build tests
|
||||
|
||||
# Протокол
|
||||
docker compose run --rm --no-deps tests --component proto
|
||||
|
||||
# Сервер
|
||||
docker compose build server-tests
|
||||
docker compose run --rm server-test-runner
|
||||
|
||||
# Клиент со служебным сервером
|
||||
docker compose build client-test1
|
||||
docker compose run --rm client-tests
|
||||
```
|
||||
|
||||
После изменения кода пересоберите соответствующий образ и повторите проверку. После изменения протокола заново сгенерируйте код и выполните:
|
||||
|
||||
```bash
|
||||
docker compose down
|
||||
docker compose build tests server-tests client-test1
|
||||
docker compose run --rm tests
|
||||
```
|
||||
|
||||
Смотрите логи через `docker compose logs`, останавливайте контейнеры командой `docker compose down`. После обновления задания скачайте свежий служебный образ: `docker compose pull client-test-server`.
|
||||
|
||||
### Ручная отладка своей связки (необязательно)
|
||||
|
||||
Запустите свой сервер и два клиента через Compose:
|
||||
|
||||
```bash
|
||||
docker compose build server client1
|
||||
docker compose up -d server client1 client2
|
||||
```
|
||||
|
||||
Сервер доступен на `localhost:51075`, клиенты — на `localhost:8080` и `localhost:8081`. Compose запускает клиентов раньше сервера. После правок пересоберите соответствующий образ и повторите `up`; для логов и остановки используйте команды выше.
|
||||
|
||||
Без Docker запустите компоненты и тестер в отдельных терминалах из папки задания. Для официальной Python-заготовки и сервера в `solution/server/server.py`:
|
||||
|
||||
```bash
|
||||
# Терминал 1
|
||||
python3 -m solution.server.server
|
||||
# Терминал 2
|
||||
python3 -m solution.client.client
|
||||
# Терминал 3
|
||||
MESSENGER_HTTP_PORT=8081 python3 -m solution.client.client
|
||||
# Терминал 4
|
||||
python3 tests/main.py
|
||||
```
|
||||
|
||||
Если вы добавили зависимости, установите и их. Чтобы проверить один компонент, передайте тестеру `--component proto`, `--component server` или `--component client`. По умолчанию проверяются все компоненты (`--component all`).
|
||||
|
||||
Здесь клиенты работают с вашим сервером, поэтому его ошибки могут повлиять на клиентские тесты. Для независимой проверки клиента используйте Compose со служебным сервером.
|
||||
|
||||
Примеры HTTP-запросов к запущенному клиенту:
|
||||
|
||||
```bash
|
||||
curl -X POST localhost:8080/sendMessage -d '{"author": "alice", "text": "hey"}'
|
||||
curl -X POST localhost:8080/getAndFlushMessages
|
||||
```
|
||||
|
||||
### Сдача решения
|
||||
|
||||
Подготовьте `solution/readme.md` и отправьте решение по [общей инструкции](../readme.md#сдача-решения). В журнале проверки будут вывод сборки, результаты тестов (после строки `=== RUN TESTS`) и логи контейнеров.
|
||||
|
||||
## Полезные материалы
|
||||
|
||||
### grpcurl
|
||||
|
||||
[grpcurl](https://github.com/fullstorydev/grpcurl) позволяет вызывать gRPC-методы из терминала. Для Linux и Windows скачайте архив для своей ОС и архитектуры со [страницы релизов](https://github.com/fullstorydev/grpcurl/releases), распакуйте его и добавьте каталог с исполняемым файлом в `PATH`. В macOS: `brew install grpcurl`.
|
||||
|
||||
Для проверки сервера откройте подписку в одном терминале, а в другом отправьте сообщение:
|
||||
|
||||
```bash
|
||||
# Терминал 1: поток остаётся открытым; Ctrl+C отменяет вызов
|
||||
grpcurl -proto solution/proto/messenger.proto -plaintext localhost:51075 mes_grpc.MessengerServer/ReadMessages
|
||||
|
||||
# Терминал 2
|
||||
grpcurl -proto solution/proto/messenger.proto -d '{"author": "alice", "text": "hello"}' -plaintext localhost:51075 mes_grpc.MessengerServer/SendMessage
|
||||
```
|
||||
|
||||
### Конкурентная обработка в Python
|
||||
|
||||
При конкурентном доступе к общим изменяемым данным учитывайте возможные гонки. Структуры данных и способы синхронизации выберите самостоятельно.
|
||||
|
||||
- [gRPC Basics Tutorial](https://grpc.io/docs/languages/python/basics/) и [официальные примеры](https://github.com/grpc/grpc/blob/master/examples).
|
||||
- Документация Python: [`threading`](https://docs.python.org/3/library/threading.html), [`queue`](https://docs.python.org/3/library/queue.html), [`asyncio`](https://docs.python.org/3/library/asyncio.html).
|
||||
- Для работы с несколькими терминалами при желании можно использовать [tmux](https://github.com/tmux/tmux/wiki/Getting-Started).
|
||||
@@ -0,0 +1 @@
|
||||
# TODO: Write Docker file for client
|
||||
@@ -0,0 +1,5 @@
|
||||
syntax = "proto3";
|
||||
|
||||
package mes_grpc;
|
||||
|
||||
// TODO: Add messages and service
|
||||
@@ -0,0 +1 @@
|
||||
# TODO: Write Docker file for server
|
||||
@@ -0,0 +1,11 @@
|
||||
FROM golang:1.23-alpine AS builder
|
||||
|
||||
WORKDIR /grpc-messenger
|
||||
COPY proto proto
|
||||
COPY client client
|
||||
RUN cd client && go mod download && go build .
|
||||
|
||||
FROM alpine:latest
|
||||
WORKDIR /grpc-messenger
|
||||
COPY --from=builder /grpc-messenger .
|
||||
CMD ["./client/client"]
|
||||
@@ -0,0 +1,12 @@
|
||||
module github.com/distsys-course/grpc-messenger/client
|
||||
|
||||
go 1.23
|
||||
|
||||
replace github.com/distsys-course/grpc-messenger/grpc => ../proto
|
||||
|
||||
require (
|
||||
github.com/distsys-course/grpc-messenger/grpc v0.0.0-00010101000000-000000000000
|
||||
github.com/gin-gonic/gin v1.10.1
|
||||
github.com/golang/protobuf v1.5.4
|
||||
google.golang.org/grpc v1.75.0
|
||||
)
|
||||
@@ -0,0 +1,80 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"os"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
mes_grpc "github.com/distsys-course/grpc-messenger/grpc"
|
||||
"github.com/gin-gonic/gin"
|
||||
"github.com/golang/protobuf/jsonpb"
|
||||
"google.golang.org/grpc"
|
||||
)
|
||||
|
||||
type ChatMessage struct {
|
||||
Author string `json:"author"`
|
||||
Text string `json:"text"`
|
||||
SendTime time.Time `json:"sendTime"`
|
||||
}
|
||||
|
||||
type MessengerClient struct {
|
||||
pendingMessages []ChatMessage
|
||||
pendingMutex sync.Mutex
|
||||
grpcClient YourMessengerServerClient
|
||||
}
|
||||
|
||||
func NewMessengerClient(serverAddr string) *MessengerClient {
|
||||
// TODO
|
||||
}
|
||||
|
||||
func (c *MessengerClient) ReadMessages() {
|
||||
// TODO: implement messages consumer here
|
||||
}
|
||||
|
||||
func (c *MessengerClient) GetPending() (messages []ChatMessage) {
|
||||
c.pendingMutex.Lock()
|
||||
result := c.pendingMessages
|
||||
c.pendingMessages = nil
|
||||
c.pendingMutex.Unlock()
|
||||
return result
|
||||
}
|
||||
|
||||
type MessageResponse struct {
|
||||
SendTime *time.Time `json:"sendTime"`
|
||||
Error *string `json:"error"`
|
||||
}
|
||||
|
||||
func main() {
|
||||
r := gin.Default()
|
||||
serverAddr := os.Getenv("MESSENGER_SERVER_ADDR")
|
||||
if serverAddr == "" {
|
||||
serverAddr = "localhost:51075"
|
||||
fmt.Println("Missing MESSENGER_SERVER_ADDR variable, using default value: " + serverAddr)
|
||||
}
|
||||
// TODO: create your grpc client with given address
|
||||
r.POST("/getAndFlushMessages", func(c *gin.Context) {
|
||||
c.JSON(http.StatusOK, client.GetPending())
|
||||
})
|
||||
|
||||
r.POST("/sendMessage", func(c *gin.Context) {
|
||||
// TODO: implement send message here, that parses body into protobuf and sends to the server
|
||||
c.JSON(http.StatusOK, MessageResponse{SendTime: nil}) // TODO: do not forget to fill SendTime
|
||||
return
|
||||
})
|
||||
|
||||
// TODO: run consumer in a goroutine
|
||||
|
||||
addr := os.Getenv("MESSENGER_HTTP_PORT")
|
||||
if addr == "" {
|
||||
addr = "0.0.0.0:8080"
|
||||
fmt.Println("Missing MESSENGER_HTTP_PORT variable, using default value: 8080")
|
||||
} else {
|
||||
addr = "0.0.0.0:" + addr
|
||||
}
|
||||
if err := r.Run(addr); err != nil {
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
module github.com/distsys-course/grpc-messenger/grpc
|
||||
|
||||
go 1.23
|
||||
@@ -0,0 +1,7 @@
|
||||
syntax = "proto3";
|
||||
|
||||
package mes_grpc;
|
||||
|
||||
option go_package = "proto/;mes_grpc";
|
||||
|
||||
// TODO: Add messages and service
|
||||
@@ -0,0 +1,60 @@
|
||||
import asyncio
|
||||
import copy
|
||||
import json
|
||||
import os
|
||||
from typing import List
|
||||
|
||||
from aiohttp import web
|
||||
|
||||
|
||||
# TODO: implement grpc client for messenger service
|
||||
|
||||
class MessengerHandler:
|
||||
_pendingMessages: List[dict] # list of messages, that have not been requested yet via get_messages
|
||||
_pendingMessagesLock: asyncio.Lock
|
||||
_grpcClient = None # grpc client of the messenger service
|
||||
|
||||
def __init__(self):
|
||||
self._pendingMessages = []
|
||||
self._pendingMessagesLock = asyncio.Lock()
|
||||
|
||||
async def send_message(self, request):
|
||||
"""
|
||||
Body should be of the form:
|
||||
{"author": "Ivan", "text": "hey guys"}
|
||||
:return web.json_response of the form {"sendTime": ... }
|
||||
"""
|
||||
j = await request.json() # TODO: use google.protobuf.json_format.ParseDict and raise BadRequest on error
|
||||
if 'author' not in j or 'text' not in j:
|
||||
raise web.HTTPBadRequest
|
||||
print('Got message to send:', json.dumps(j))
|
||||
|
||||
# TODO: your rpc call of the messenger here
|
||||
|
||||
raise NotImplementedError
|
||||
return web.json_response({'sendTime': ""}) # TODO: use google.protobuf.json_format.MessageToDict here
|
||||
|
||||
async def get_messages(self, _):
|
||||
async with self._pendingMessagesLock:
|
||||
res: List[dict] = copy.deepcopy(self._pendingMessages)
|
||||
self._pendingMessages = []
|
||||
return web.json_response(res)
|
||||
|
||||
# TODO: implement message stream consumer in async method, that fills self._pendingMessages
|
||||
# btw, consumption can be lazy and happen on get_messages, implement in any suitable way
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
app = web.Application()
|
||||
grpcServerAddr = os.environ.get('MESSENGER_SERVER_ADDR', 'localhost:51075')
|
||||
|
||||
# TODO: create your grpc client with given address and pass it to MessengerHandler constructor
|
||||
|
||||
handler = MessengerHandler()
|
||||
app.add_routes([web.post('/getAndFlushMessages', handler.get_messages)])
|
||||
app.add_routes([web.post('/sendMessage', handler.send_message)])
|
||||
|
||||
# TODO: run message stream consumer in a background coroutine
|
||||
|
||||
httpPort = os.environ.get('MESSENGER_HTTP_PORT', '8080')
|
||||
web.run_app(app, host='0.0.0.0', port=httpPort)
|
||||
@@ -0,0 +1,3 @@
|
||||
aiohttp==3.12.15
|
||||
grpcio==1.75.0
|
||||
grpcio-tools==1.75.0
|
||||
@@ -0,0 +1,11 @@
|
||||
FROM python:3.12-slim
|
||||
|
||||
WORKDIR /grpc-messenger
|
||||
|
||||
COPY client/requirements.txt .
|
||||
RUN pip install --no-cache-dir -r requirements.txt
|
||||
|
||||
COPY client/client.py solution/client/
|
||||
COPY proto solution/proto/
|
||||
|
||||
ENTRYPOINT ["python", "-m", "solution.client.client"]
|
||||
@@ -0,0 +1,102 @@
|
||||
import copy
|
||||
import json
|
||||
import os
|
||||
import threading
|
||||
from http import HTTPStatus
|
||||
from http.server import HTTPServer, BaseHTTPRequestHandler
|
||||
from typing import List, Dict
|
||||
|
||||
import google.protobuf.empty_pb2 # Empty
|
||||
import google.protobuf.json_format # ParseDict, MessageToDict
|
||||
import grpc
|
||||
|
||||
from solution.proto import messenger_pb2
|
||||
from solution.proto import messenger_pb2_grpc
|
||||
|
||||
|
||||
class PostBox:
|
||||
def __init__(self):
|
||||
self._messages: List[Dict] = []
|
||||
self._lock = threading.Lock()
|
||||
|
||||
def collect_messages(self) -> List[Dict]:
|
||||
with self._lock:
|
||||
messages = copy.deepcopy(self._messages)
|
||||
self._messages = []
|
||||
return messages
|
||||
|
||||
def put_message(self, message: Dict):
|
||||
with self._lock:
|
||||
self._messages.append(message)
|
||||
|
||||
|
||||
class MessageHandler(BaseHTTPRequestHandler):
|
||||
_stub = None
|
||||
_postbox: PostBox
|
||||
|
||||
def _read_content(self):
|
||||
content_length = int(self.headers['Content-Length'])
|
||||
bytes_content = self.rfile.read(content_length)
|
||||
return bytes_content.decode('ascii')
|
||||
|
||||
# noinspection PyPep8Naming
|
||||
def do_POST(self):
|
||||
if self.path == '/sendMessage':
|
||||
response = self._send_message(self._read_content())
|
||||
elif self.path == '/getAndFlushMessages':
|
||||
response = self._get_messages()
|
||||
else:
|
||||
self.send_error(HTTPStatus.NOT_IMPLEMENTED)
|
||||
self.end_headers()
|
||||
return
|
||||
|
||||
response_bytes = json.dumps(response).encode('ascii')
|
||||
self.send_response(HTTPStatus.OK)
|
||||
self.send_header('Content-Length', str(len(response_bytes)))
|
||||
self.end_headers()
|
||||
self.wfile.write(response_bytes)
|
||||
|
||||
def _send_message(self, content: str) -> dict:
|
||||
json_request = json.loads(content)
|
||||
|
||||
# TODO: use google.protobuf.json_format.ParseDict
|
||||
|
||||
# TODO: your rpc call of the messenger here
|
||||
|
||||
# TODO: use google.protobuf.json_format.MessageToDict here
|
||||
return {'sendTime': ''}
|
||||
|
||||
def _get_messages(self) -> List[dict]:
|
||||
return self._postbox.collect_messages()
|
||||
|
||||
|
||||
def main():
|
||||
grpc_server_address = os.environ.get('MESSENGER_SERVER_ADDR', 'localhost:51075')
|
||||
|
||||
# TODO: create your grpc client and wait for the server to become available.
|
||||
# The client may start before the server.
|
||||
stub = None
|
||||
|
||||
# A list of messages obtained from the server-py but not yet requested by the user to be shown
|
||||
# (via the http's /getAndFlushMessages).
|
||||
postbox = PostBox()
|
||||
|
||||
# TODO: Implement and run a messages stream consumer in a background thread here.
|
||||
# It should fetch messages via the grpc client and store them in the postbox.
|
||||
|
||||
# Pass the stub and the postbox to the HTTP server.
|
||||
# Dirty, but this simple http server doesn't provide interface
|
||||
# for passing arguments to the handler c-tor.
|
||||
MessageHandler._stub = stub
|
||||
MessageHandler._postbox = postbox
|
||||
|
||||
http_port = os.environ.get('MESSENGER_HTTP_PORT', '8080')
|
||||
http_server_address = ('0.0.0.0', int(http_port))
|
||||
|
||||
# NB: handler_class is instantiated for every http request. Do not store any inter-request state in it.
|
||||
httpd = HTTPServer(http_server_address, MessageHandler)
|
||||
httpd.serve_forever()
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
main()
|
||||
@@ -0,0 +1,2 @@
|
||||
grpcio==1.75.0
|
||||
grpcio-tools==1.75.0
|
||||
@@ -0,0 +1,5 @@
|
||||
syntax = "proto3";
|
||||
|
||||
package mes_grpc;
|
||||
|
||||
// TODO: Add messages and service
|
||||
@@ -0,0 +1,21 @@
|
||||
# syntax=docker/dockerfile:1
|
||||
FROM python:3.12-alpine
|
||||
|
||||
COPY tests/requirements.txt .
|
||||
RUN --mount=type=cache,id=distsys-course-pip,target=/root/.cache/pip,sharing=locked \
|
||||
pip install -r requirements.txt
|
||||
|
||||
ARG GRPCURL_VERSION=1.9.3
|
||||
ARG GRPCURL_SHA256=a926b62a85787ccf73ef8736b3ae554f1242e39d92bb8767a79d6dd23b11d1d5
|
||||
RUN apk add --no-cache curl tini \
|
||||
&& curl --fail --silent --show-error --location \
|
||||
"https://github.com/fullstorydev/grpcurl/releases/download/v${GRPCURL_VERSION}/grpcurl_${GRPCURL_VERSION}_linux_x86_64.tar.gz" \
|
||||
--output /tmp/grpcurl.tar.gz \
|
||||
&& echo "${GRPCURL_SHA256} /tmp/grpcurl.tar.gz" | sha256sum -c \
|
||||
&& tar -xzf /tmp/grpcurl.tar.gz -C /usr/local/bin \
|
||||
&& rm -f /tmp/grpcurl.tar.gz
|
||||
|
||||
COPY solution/proto/ solution/proto/
|
||||
COPY tests/*.py tests/
|
||||
|
||||
ENTRYPOINT ["/sbin/tini", "--", "python3", "-u", "tests/main.py"]
|
||||
@@ -0,0 +1,109 @@
|
||||
import argparse
|
||||
import os
|
||||
import pathlib
|
||||
import signal
|
||||
import subprocess
|
||||
import sys
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
SCRIPT_DIR = pathlib.Path(__file__).parent.resolve()
|
||||
SUITE_TIMEOUT_S = 120
|
||||
|
||||
|
||||
class PassedCounter:
|
||||
def __init__(self):
|
||||
self.passed = 0
|
||||
|
||||
def pytest_report_teststatus(self, report, config):
|
||||
if report.when == 'call' and report.passed:
|
||||
self.passed += 1
|
||||
|
||||
|
||||
def suite_score(passed, expected, maximum, exit_code=pytest.ExitCode.OK):
|
||||
return maximum if exit_code == pytest.ExitCode.OK and passed == expected else 0
|
||||
|
||||
|
||||
def run_suite(filename, expected, maximum):
|
||||
# Each component has its own deadline so a timeout cannot discard other scores.
|
||||
process = subprocess.Popen(
|
||||
[sys.executable, '-u', str(pathlib.Path(__file__).resolve()),
|
||||
'--suite', str(SCRIPT_DIR / filename), str(expected)],
|
||||
start_new_session=os.name == 'posix',
|
||||
)
|
||||
try:
|
||||
exit_code = process.wait(timeout=SUITE_TIMEOUT_S)
|
||||
except subprocess.TimeoutExpired:
|
||||
if os.name == 'posix':
|
||||
# Include grpcurl and other subprocesses started by this test suite.
|
||||
try:
|
||||
os.killpg(process.pid, signal.SIGKILL)
|
||||
except ProcessLookupError:
|
||||
pass # The suite may have exited just after wait() timed out.
|
||||
else:
|
||||
process.kill()
|
||||
process.wait()
|
||||
print(f'{filename} exceeded {SUITE_TIMEOUT_S} seconds; component score is zero.', flush=True)
|
||||
return 0
|
||||
return maximum if exit_code == 0 else 0
|
||||
|
||||
|
||||
def suite_exit_code(filename, expected):
|
||||
counter = PassedCounter()
|
||||
exit_code = pytest.main(['-vs', filename], plugins=[counter])
|
||||
return 0 if suite_score(counter.passed, expected, 1, exit_code) else 1
|
||||
|
||||
|
||||
def component_enabled(environment_name):
|
||||
return os.environ.get(environment_name, '1') == '1'
|
||||
|
||||
|
||||
def parse_args():
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument(
|
||||
'--component',
|
||||
choices=('all', 'proto', 'server', 'client'),
|
||||
default='all',
|
||||
)
|
||||
parser.add_argument('--suite', nargs=2, metavar=('FILE', 'EXPECTED'), help=argparse.SUPPRESS)
|
||||
return parser.parse_args()
|
||||
|
||||
|
||||
def main():
|
||||
args = parse_args()
|
||||
if args.suite is not None:
|
||||
return suite_exit_code(args.suite[0], int(args.suite[1]))
|
||||
component = args.component
|
||||
score = 0
|
||||
|
||||
if component in ('all', 'proto'):
|
||||
proto_score = run_suite('test_proto.py', expected=1, maximum=2)
|
||||
score += proto_score
|
||||
print(f'Proto: {proto_score}/2')
|
||||
print()
|
||||
|
||||
if component in ('all', 'server'):
|
||||
if component_enabled('MESSENGER_SERVER_TESTS_ENABLED'):
|
||||
server_score = run_suite('test_server.py', expected=4, maximum=4)
|
||||
else:
|
||||
server_score = 0
|
||||
print('Server tests were not run because the server image did not build.')
|
||||
score += server_score
|
||||
print(f'Server: {server_score}/4')
|
||||
print()
|
||||
|
||||
if component in ('all', 'client'):
|
||||
if component_enabled('MESSENGER_CLIENT_TESTS_ENABLED'):
|
||||
client_score = run_suite('test_client.py', expected=3, maximum=4)
|
||||
else:
|
||||
client_score = 0
|
||||
print('Client tests were not run because the client image did not build.')
|
||||
score += client_score
|
||||
print(f'Client: {client_score}/4')
|
||||
|
||||
print(f'\nSCORE: {score}')
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
sys.exit(main())
|
||||
@@ -0,0 +1,5 @@
|
||||
grpcio==1.75.0
|
||||
grpcio-tools==1.75.0
|
||||
protobuf==6.32.1
|
||||
pytest==8.4.2
|
||||
requests==2.32.5
|
||||
@@ -0,0 +1,179 @@
|
||||
import os
|
||||
import time
|
||||
|
||||
import pytest
|
||||
import requests
|
||||
|
||||
HTTP_CONNECT_TIMEOUT_S = 1
|
||||
HTTP_READ_TIMEOUT_S = 5
|
||||
HTTP_READY_TIMEOUT_S = 20
|
||||
HTTP_RETRY_INTERVAL_S = 0.5
|
||||
HTTP_TIMEOUT = (HTTP_CONNECT_TIMEOUT_S, HTTP_READ_TIMEOUT_S)
|
||||
MESSAGE_TIMEOUT_S = 10
|
||||
MESSAGE_POLL_INTERVAL_S = 0.05
|
||||
|
||||
|
||||
def wait_for_http(url):
|
||||
deadline = time.monotonic() + HTTP_READY_TIMEOUT_S
|
||||
last_exception = None
|
||||
while True:
|
||||
remaining = deadline - time.monotonic()
|
||||
if remaining <= 0:
|
||||
pytest.fail(
|
||||
f'timed out waiting for HTTP endpoint {url}: {last_exception}',
|
||||
pytrace=False,
|
||||
)
|
||||
readiness_timeout = (
|
||||
min(HTTP_CONNECT_TIMEOUT_S, remaining),
|
||||
min(HTTP_READ_TIMEOUT_S, remaining),
|
||||
)
|
||||
try:
|
||||
response = requests.get(url, timeout=readiness_timeout)
|
||||
response.close()
|
||||
return
|
||||
except (requests.exceptions.ConnectionError, requests.exceptions.Timeout) as exc:
|
||||
last_exception = exc
|
||||
|
||||
remaining = deadline - time.monotonic()
|
||||
if remaining <= 0:
|
||||
pytest.fail(
|
||||
f'timed out waiting for HTTP endpoint {url}: {last_exception}',
|
||||
pytrace=False,
|
||||
)
|
||||
time.sleep(min(HTTP_RETRY_INTERVAL_S, remaining))
|
||||
|
||||
|
||||
def post_json(url, path, timeout=HTTP_TIMEOUT, **kwargs):
|
||||
endpoint = url + path
|
||||
try:
|
||||
response = requests.post(endpoint, timeout=timeout, **kwargs)
|
||||
except requests.exceptions.Timeout as exc:
|
||||
pytest.fail(f'timed out waiting for HTTP response from {endpoint}: {exc}', pytrace=False)
|
||||
except requests.exceptions.ConnectionError as exc:
|
||||
pytest.fail(f'could not connect to HTTP endpoint {endpoint}: {exc}', pytrace=False)
|
||||
|
||||
try:
|
||||
assert response.status_code == 200, response.text
|
||||
return response.json()
|
||||
finally:
|
||||
response.close()
|
||||
|
||||
|
||||
@pytest.fixture(scope='session')
|
||||
def client1_ready_url():
|
||||
url = 'http://' + os.environ.get('MESSENGER_TEST_CLIENT1_ADDR', '127.0.0.1:8080')
|
||||
wait_for_http(url)
|
||||
return url
|
||||
|
||||
|
||||
@pytest.fixture(scope='session')
|
||||
def client2_ready_url():
|
||||
url = 'http://' + os.environ.get('MESSENGER_TEST_CLIENT2_ADDR', '127.0.0.1:8081')
|
||||
wait_for_http(url)
|
||||
return url
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client1_url(client1_ready_url):
|
||||
url = client1_ready_url
|
||||
get_messages(url) # we need to flush pending messages before and after each tests
|
||||
yield url
|
||||
get_messages(url)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client2_url(client2_ready_url):
|
||||
url = client2_ready_url
|
||||
get_messages(url)
|
||||
yield url
|
||||
get_messages(url)
|
||||
|
||||
|
||||
def send_message(url, mes):
|
||||
return post_json(url, '/sendMessage', json=mes)
|
||||
|
||||
|
||||
def get_messages(url, **kwargs):
|
||||
return post_json(url, '/getAndFlushMessages', **kwargs)
|
||||
|
||||
|
||||
def wait_for_messages(url, expected):
|
||||
deadline = time.monotonic() + MESSAGE_TIMEOUT_S
|
||||
messages = []
|
||||
while len(messages) < len(expected):
|
||||
remaining = deadline - time.monotonic()
|
||||
if remaining <= 0:
|
||||
pytest.fail(
|
||||
f'timed out waiting for messages from {url}: '
|
||||
f'expected {expected!r}, received {messages!r}',
|
||||
pytrace=False,
|
||||
)
|
||||
# Divide the remaining budget between connecting and reading the response.
|
||||
timeout = (
|
||||
min(HTTP_CONNECT_TIMEOUT_S, remaining / 2),
|
||||
min(HTTP_READ_TIMEOUT_S, remaining / 2),
|
||||
)
|
||||
batch = get_messages(url, timeout=timeout)
|
||||
assert isinstance(batch, list), f'expected a message array, received {batch!r}'
|
||||
messages.extend(batch)
|
||||
assert messages == expected[:len(messages)], \
|
||||
f'expected {expected!r}, received {messages!r}'
|
||||
remaining = deadline - time.monotonic()
|
||||
if remaining <= 0:
|
||||
pytest.fail(f'timed out waiting for messages from {url}', pytrace=False)
|
||||
if len(messages) < len(expected):
|
||||
time.sleep(min(MESSAGE_POLL_INTERVAL_S, remaining))
|
||||
return messages
|
||||
|
||||
|
||||
def test_single_client_single_message(client1_url, client2_url):
|
||||
mes = {
|
||||
'author': 'TestSingleClient',
|
||||
'text': 'This is test text'
|
||||
}
|
||||
resp = send_message(client1_url, mes)
|
||||
mes['sendTime'] = resp['sendTime']
|
||||
assert wait_for_messages(client1_url, [mes]) == [mes]
|
||||
# Drain both subscriptions before the next test sends more messages.
|
||||
assert wait_for_messages(client2_url, [mes]) == [mes]
|
||||
|
||||
|
||||
def test_single_client_multiple_messages(client1_url, client2_url):
|
||||
mes = [{
|
||||
'author': 'TestSingleClient1',
|
||||
'text': 'This is test text'
|
||||
}, {
|
||||
'author': 'TestSingleClient2',
|
||||
'text': 'This is test text'
|
||||
}]
|
||||
for m in mes:
|
||||
resp = send_message(client1_url, m)
|
||||
m['sendTime'] = resp['sendTime']
|
||||
assert wait_for_messages(client1_url, mes) == mes
|
||||
assert wait_for_messages(client2_url, mes) == mes
|
||||
|
||||
|
||||
def test_two_clients_multiple_messages(client1_url, client2_url):
|
||||
client1_name = 'TestMultiClient1'
|
||||
client2_name = 'TestMultiClient2'
|
||||
mes = [{
|
||||
'author': client1_name,
|
||||
'text': 'This is test text #1'
|
||||
}, {
|
||||
'author': client1_name,
|
||||
'text': 'This is test text #2'
|
||||
}, {
|
||||
'author': client2_name,
|
||||
'text': 'This is test text #3'
|
||||
}, {
|
||||
'author': client2_name,
|
||||
'text': 'This is test text #4'
|
||||
}]
|
||||
times = set()
|
||||
for m in mes:
|
||||
resp = send_message(client1_url if m['author'] == client1_name else client2_url, m)
|
||||
m['sendTime'] = resp['sendTime']
|
||||
times.add(m['sendTime'])
|
||||
assert len(times) == len(mes)
|
||||
assert wait_for_messages(client1_url, mes) == mes
|
||||
assert wait_for_messages(client2_url, mes) == mes
|
||||
@@ -0,0 +1,140 @@
|
||||
import pathlib
|
||||
import tempfile
|
||||
|
||||
import grpc_tools
|
||||
from google.protobuf import descriptor
|
||||
from google.protobuf import descriptor_pb2
|
||||
from google.protobuf import descriptor_pool
|
||||
from grpc_tools import protoc
|
||||
|
||||
|
||||
SCRIPT_DIR = pathlib.Path(__file__).parent.resolve()
|
||||
PROTO_DIR = SCRIPT_DIR.parent / 'solution' / 'proto'
|
||||
PROTO_FILE = PROTO_DIR / 'messenger.proto'
|
||||
WELL_KNOWN_PROTO_DIR = pathlib.Path(grpc_tools.__file__).parent / '_proto'
|
||||
|
||||
|
||||
def compile_descriptor_set(output_path):
|
||||
result = protoc.main([
|
||||
'grpc_tools.protoc',
|
||||
f'-I{PROTO_DIR}',
|
||||
f'-I{WELL_KNOWN_PROTO_DIR}',
|
||||
f'--descriptor_set_out={output_path}',
|
||||
'--include_imports',
|
||||
str(PROTO_FILE),
|
||||
])
|
||||
assert result == 0, 'messenger.proto must compile successfully'
|
||||
|
||||
descriptor_set = descriptor_pb2.FileDescriptorSet()
|
||||
descriptor_set.ParseFromString(output_path.read_bytes())
|
||||
return descriptor_set
|
||||
|
||||
|
||||
def build_descriptor_pool(descriptor_set):
|
||||
pool = descriptor_pool.DescriptorPool()
|
||||
remaining = list(descriptor_set.file)
|
||||
while remaining:
|
||||
deferred = []
|
||||
for file_descriptor in remaining:
|
||||
try:
|
||||
pool.Add(file_descriptor)
|
||||
except TypeError:
|
||||
deferred.append(file_descriptor)
|
||||
assert len(deferred) < len(remaining), 'messenger.proto imports could not be resolved'
|
||||
remaining = deferred
|
||||
return pool
|
||||
|
||||
|
||||
def require_singular_field(message_type, field_name, field_type, message_type_name=None):
|
||||
assert field_name in message_type.fields_by_name, \
|
||||
f'{message_type.full_name} must contain field {field_name}'
|
||||
field = message_type.fields_by_name[field_name]
|
||||
assert not field.is_repeated, f'{field.full_name} must be a singular field'
|
||||
assert field.type == field_type, f'{field.full_name} has an invalid type'
|
||||
if message_type_name is not None:
|
||||
assert field.message_type is not None
|
||||
assert field.message_type.full_name == message_type_name, \
|
||||
f'{field.full_name} has an invalid message type'
|
||||
|
||||
|
||||
def require_fields_can_coexist(message_type, field_names):
|
||||
fields_by_oneof = {}
|
||||
for field_name in field_names:
|
||||
field = message_type.fields_by_name[field_name]
|
||||
if field.containing_oneof is None:
|
||||
continue
|
||||
previous_field = fields_by_oneof.setdefault(field.containing_oneof.full_name, field_name)
|
||||
assert previous_field == field_name, \
|
||||
f'{message_type.full_name} fields must allow simultaneous values'
|
||||
|
||||
|
||||
def test_proto_contract():
|
||||
with tempfile.TemporaryDirectory() as temporary_directory:
|
||||
descriptor_path = pathlib.Path(temporary_directory) / 'messenger.pb'
|
||||
descriptor_set = compile_descriptor_set(descriptor_path)
|
||||
|
||||
submitted_file = next(
|
||||
(file_descriptor for file_descriptor in descriptor_set.file
|
||||
if pathlib.PurePosixPath(file_descriptor.name).name == PROTO_FILE.name),
|
||||
None,
|
||||
)
|
||||
assert submitted_file is not None, 'messenger.proto descriptor is missing'
|
||||
assert submitted_file.syntax == 'proto3', 'messenger.proto must use proto3 syntax'
|
||||
assert submitted_file.package == 'mes_grpc', 'messenger.proto must use package mes_grpc'
|
||||
|
||||
pool = build_descriptor_pool(descriptor_set)
|
||||
try:
|
||||
messenger = pool.FindServiceByName('mes_grpc.MessengerServer')
|
||||
except KeyError:
|
||||
raise AssertionError('gRPC service must be named mes_grpc.MessengerServer') from None
|
||||
|
||||
assert 'SendMessage' in messenger.methods_by_name, \
|
||||
'MessengerServer must contain method SendMessage'
|
||||
send_message = messenger.methods_by_name['SendMessage']
|
||||
assert not send_message.client_streaming and not send_message.server_streaming, \
|
||||
'SendMessage must be unary'
|
||||
require_singular_field(
|
||||
send_message.input_type,
|
||||
'author',
|
||||
descriptor.FieldDescriptor.TYPE_STRING,
|
||||
)
|
||||
require_singular_field(
|
||||
send_message.input_type,
|
||||
'text',
|
||||
descriptor.FieldDescriptor.TYPE_STRING,
|
||||
)
|
||||
require_fields_can_coexist(send_message.input_type, ('author', 'text'))
|
||||
require_singular_field(
|
||||
send_message.output_type,
|
||||
'sendTime',
|
||||
descriptor.FieldDescriptor.TYPE_MESSAGE,
|
||||
'google.protobuf.Timestamp',
|
||||
)
|
||||
|
||||
assert 'ReadMessages' in messenger.methods_by_name, \
|
||||
'MessengerServer must contain method ReadMessages'
|
||||
read_messages = messenger.methods_by_name['ReadMessages']
|
||||
assert not read_messages.client_streaming and read_messages.server_streaming, \
|
||||
'ReadMessages must be a unary request with a server stream response'
|
||||
assert not read_messages.input_type.fields, \
|
||||
'ReadMessages request must not contain fields'
|
||||
require_singular_field(
|
||||
read_messages.output_type,
|
||||
'author',
|
||||
descriptor.FieldDescriptor.TYPE_STRING,
|
||||
)
|
||||
require_singular_field(
|
||||
read_messages.output_type,
|
||||
'text',
|
||||
descriptor.FieldDescriptor.TYPE_STRING,
|
||||
)
|
||||
require_singular_field(
|
||||
read_messages.output_type,
|
||||
'sendTime',
|
||||
descriptor.FieldDescriptor.TYPE_MESSAGE,
|
||||
'google.protobuf.Timestamp',
|
||||
)
|
||||
require_fields_can_coexist(
|
||||
read_messages.output_type,
|
||||
('author', 'text', 'sendTime'),
|
||||
)
|
||||
@@ -0,0 +1,285 @@
|
||||
import copy
|
||||
import json
|
||||
import os
|
||||
import pathlib
|
||||
import queue
|
||||
import re
|
||||
import socket
|
||||
import subprocess
|
||||
import threading
|
||||
import time
|
||||
from datetime import datetime, timezone
|
||||
from typing import Dict
|
||||
|
||||
import pytest
|
||||
|
||||
test_message = {'author': 'alice', 'text': 'hello'}
|
||||
|
||||
PROTO_DIR = pathlib.Path(__file__).resolve().parent.parent / 'solution' / 'proto'
|
||||
|
||||
SOCKET_CONNECT_TIMEOUT_S = 1
|
||||
SERVICE_READY_TIMEOUT_S = 20
|
||||
SERVICE_RETRY_INTERVAL_S = 0.5
|
||||
GRPC_CALL_TIMEOUT_S = 5
|
||||
GRPC_PROCESS_TIMEOUT_S = 10
|
||||
GRPC_STREAM_TIMEOUT_S = 60
|
||||
PROCESS_STOP_TIMEOUT_S = 5
|
||||
MESSAGE_TIMEOUT_S = 10
|
||||
|
||||
|
||||
def wait_for_socket(host, port):
|
||||
deadline = time.monotonic() + SERVICE_READY_TIMEOUT_S
|
||||
last_exception = None
|
||||
while True:
|
||||
try:
|
||||
with socket.create_connection((host, port), timeout=SOCKET_CONNECT_TIMEOUT_S):
|
||||
pass
|
||||
return
|
||||
except OSError as exc:
|
||||
last_exception = exc
|
||||
|
||||
remaining = deadline - time.monotonic()
|
||||
if remaining <= 0:
|
||||
pytest.fail(
|
||||
f'timed out waiting for TCP endpoint {host}:{port}: {last_exception}',
|
||||
pytrace=False,
|
||||
)
|
||||
time.sleep(min(SERVICE_RETRY_INTERVAL_S, remaining))
|
||||
|
||||
|
||||
@pytest.fixture(scope='session')
|
||||
def server_addr():
|
||||
addr = os.environ.get('MESSENGER_TEST_SERVER_ADDR', '127.0.0.1:51075')
|
||||
host = addr.split(':')[0]
|
||||
port = int(addr.split(':')[1])
|
||||
wait_for_socket(host, port)
|
||||
yield addr
|
||||
|
||||
|
||||
def send_message(server_address, message: Dict[str, str]) -> Dict[str, str]:
|
||||
grpcurl_cmd = ['grpcurl',
|
||||
'-max-time', str(GRPC_CALL_TIMEOUT_S),
|
||||
'-import-path', str(PROTO_DIR),
|
||||
'-proto', 'messenger.proto',
|
||||
'-d',
|
||||
json.dumps(message),
|
||||
'-plaintext',
|
||||
server_address,
|
||||
'mes_grpc.MessengerServer/SendMessage']
|
||||
try:
|
||||
completed = subprocess.run(
|
||||
grpcurl_cmd,
|
||||
capture_output=True,
|
||||
check=False,
|
||||
timeout=GRPC_PROCESS_TIMEOUT_S,
|
||||
)
|
||||
except subprocess.TimeoutExpired:
|
||||
pytest.fail(
|
||||
f'grpcurl did not finish within {GRPC_PROCESS_TIMEOUT_S} seconds',
|
||||
pytrace=False,
|
||||
)
|
||||
assert completed.returncode == 0, completed.stderr
|
||||
assert len(completed.stderr) == 0, completed.stderr
|
||||
output_str = completed.stdout.decode('ascii')
|
||||
output = json.loads(output_str)
|
||||
|
||||
message_with_timestamp = copy.deepcopy(message)
|
||||
message_with_timestamp['sendTime'] = output['sendTime']
|
||||
return message_with_timestamp
|
||||
|
||||
|
||||
class MessageStream:
|
||||
def __init__(self, server_address):
|
||||
grpcurl_cmd = ['grpcurl',
|
||||
'-max-time', str(GRPC_STREAM_TIMEOUT_S),
|
||||
'-import-path', str(PROTO_DIR),
|
||||
'-proto', 'messenger.proto',
|
||||
'-plaintext',
|
||||
server_address,
|
||||
'mes_grpc.MessengerServer/ReadMessages']
|
||||
self._process = subprocess.Popen(
|
||||
grpcurl_cmd,
|
||||
stdout=subprocess.PIPE,
|
||||
stderr=subprocess.PIPE,
|
||||
text=True,
|
||||
)
|
||||
self._messages = queue.Queue()
|
||||
self._reader_error = None
|
||||
self._reader_finished = threading.Event()
|
||||
self._closed = False
|
||||
self._reader = threading.Thread(target=self._read_messages, daemon=True)
|
||||
self._reader.start()
|
||||
|
||||
def _read_messages(self):
|
||||
try:
|
||||
message_lines = []
|
||||
for line in self._process.stdout:
|
||||
message_lines.append(line)
|
||||
if line.rstrip() == '}':
|
||||
self._messages.put(json.loads(''.join(message_lines)))
|
||||
message_lines = []
|
||||
except Exception as exc:
|
||||
self._reader_error = exc
|
||||
finally:
|
||||
self._reader_finished.set()
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_value, traceback):
|
||||
self.close()
|
||||
|
||||
def _failure_detail(self):
|
||||
if self._reader_error is not None:
|
||||
return f'reader failed: {self._reader_error}'
|
||||
returncode = self._process.poll()
|
||||
if returncode is not None:
|
||||
stderr = self._process.stderr.read().strip()
|
||||
return f'grpcurl exited with status {returncode}: {stderr}'
|
||||
return 'grpcurl is still running but produced no matching message'
|
||||
|
||||
def _get_message(self, deadline, timeout_message):
|
||||
while True:
|
||||
if self._reader_error is not None:
|
||||
raise AssertionError(self._failure_detail())
|
||||
if self._reader_finished.is_set() and self._messages.empty():
|
||||
raise AssertionError(self._failure_detail())
|
||||
|
||||
remaining = deadline - time.monotonic()
|
||||
if remaining <= 0:
|
||||
raise AssertionError(f'{timeout_message}: {self._failure_detail()}')
|
||||
try:
|
||||
return self._messages.get(timeout=min(remaining, 0.1))
|
||||
except queue.Empty:
|
||||
pass
|
||||
|
||||
def wait_for_message(self, expected_message, timeout, preceding_messages=None):
|
||||
deadline = time.monotonic() + timeout
|
||||
while True:
|
||||
message = self._get_message(
|
||||
deadline,
|
||||
f'timed out waiting for message {expected_message}',
|
||||
)
|
||||
if message == expected_message:
|
||||
return
|
||||
if preceding_messages is not None:
|
||||
preceding_messages.append(message)
|
||||
|
||||
def read_messages(self, count, timeout):
|
||||
deadline = time.monotonic() + timeout
|
||||
messages = []
|
||||
while len(messages) < count:
|
||||
messages.append(self._get_message(
|
||||
deadline,
|
||||
f'timed out after receiving {len(messages)} of {count} messages',
|
||||
))
|
||||
return messages
|
||||
|
||||
def close(self):
|
||||
if self._closed:
|
||||
return
|
||||
self._closed = True
|
||||
if self._process.poll() is None:
|
||||
self._process.terminate()
|
||||
try:
|
||||
self._process.wait(timeout=PROCESS_STOP_TIMEOUT_S)
|
||||
except subprocess.TimeoutExpired:
|
||||
self._process.kill()
|
||||
try:
|
||||
self._process.wait(timeout=PROCESS_STOP_TIMEOUT_S)
|
||||
except subprocess.TimeoutExpired:
|
||||
pytest.fail('grpcurl did not exit after SIGKILL', pytrace=False)
|
||||
self._reader.join(timeout=PROCESS_STOP_TIMEOUT_S)
|
||||
assert not self._reader.is_alive()
|
||||
|
||||
|
||||
def wait_for_streams(server_address, streams):
|
||||
preceding_messages = [[] for _ in streams]
|
||||
probes = []
|
||||
for attempt in range(10):
|
||||
probe = send_message(
|
||||
server_address,
|
||||
{'author': 'StreamProbe', 'text': f'probe #{attempt}'},
|
||||
)
|
||||
probes.append(probe)
|
||||
streams_ready = True
|
||||
for index, stream in enumerate(streams):
|
||||
try:
|
||||
stream.wait_for_message(
|
||||
probe,
|
||||
timeout=1,
|
||||
preceding_messages=preceding_messages[index],
|
||||
)
|
||||
except AssertionError:
|
||||
streams_ready = False
|
||||
if streams_ready:
|
||||
return [
|
||||
[message for message in messages if message not in probes]
|
||||
for messages in preceding_messages
|
||||
]
|
||||
raise AssertionError('ReadMessages streams did not become ready')
|
||||
|
||||
|
||||
def timestamp_key(timestamp):
|
||||
match = re.fullmatch(r'(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(?:\.(\d{1,9}))?Z', timestamp)
|
||||
assert match is not None, f'invalid protobuf timestamp: {timestamp}'
|
||||
seconds = int(datetime.strptime(match.group(1), '%Y-%m-%dT%H:%M:%S')
|
||||
.replace(tzinfo=timezone.utc).timestamp())
|
||||
nanos = int((match.group(2) or '').ljust(9, '0'))
|
||||
return seconds, nanos
|
||||
|
||||
|
||||
def test_send_smoke(server_addr):
|
||||
send_message(server_addr, test_message)
|
||||
|
||||
|
||||
def test_send_returns_ascending_time(server_addr):
|
||||
outputs = []
|
||||
for _ in range(10):
|
||||
outputs.append(send_message(server_addr, test_message))
|
||||
|
||||
for output1, output2 in zip(outputs, outputs[1:]):
|
||||
assert timestamp_key(output1['sendTime']) < timestamp_key(output2['sendTime'])
|
||||
|
||||
|
||||
def test_get_messages_smoke(server_addr):
|
||||
with MessageStream(server_addr) as stream:
|
||||
wait_for_streams(server_addr, [stream])
|
||||
test_message_with_timestamp = send_message(server_addr, test_message)
|
||||
messages = stream.read_messages(1, timeout=MESSAGE_TIMEOUT_S)
|
||||
|
||||
assert len(messages) == 1
|
||||
assert messages[0] == test_message_with_timestamp
|
||||
|
||||
|
||||
def test_get_only_sends_new(server_addr):
|
||||
messages1 = []
|
||||
messages3 = []
|
||||
n1, n2, n3 = 2, 3, 4
|
||||
|
||||
with MessageStream(server_addr) as stream:
|
||||
messages_before_ready = wait_for_streams(server_addr, [stream])
|
||||
assert messages_before_ready == [[]]
|
||||
for _ in range(n1):
|
||||
message = send_message(server_addr, test_message)
|
||||
messages1.append(message)
|
||||
messages = stream.read_messages(n1, timeout=MESSAGE_TIMEOUT_S)
|
||||
|
||||
assert len(messages1) == len(messages)
|
||||
for m1, m2 in zip(messages1, messages):
|
||||
assert m1 == m2
|
||||
|
||||
for _ in range(n2):
|
||||
send_message(server_addr, test_message)
|
||||
|
||||
with MessageStream(server_addr) as stream:
|
||||
messages_before_ready = wait_for_streams(server_addr, [stream])
|
||||
assert messages_before_ready == [[]]
|
||||
for _ in range(n3):
|
||||
message = send_message(server_addr, test_message)
|
||||
messages3.append(message)
|
||||
messages = stream.read_messages(n3, timeout=MESSAGE_TIMEOUT_S)
|
||||
|
||||
assert len(messages3) == len(messages)
|
||||
for m1, m2 in zip(messages3, messages):
|
||||
assert m1 == m2
|
||||
Binary file not shown.
Binary file not shown.
|
After Width: | Height: | Size: 50 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 59 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 24 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 45 KiB |
Binary file not shown.
|
After Width: | Height: | Size: 69 KiB |
@@ -0,0 +1,347 @@
|
||||
# Семинар 2. TCP/UDP, Protobuf и gRPC
|
||||
|
||||
## Задача: положить число в Storage
|
||||
|
||||
Допустим, у нас есть простой сервис **Storage**. Он хранит одно число, умеет обновлять его через `PutValue` и возвращать через `GetValue`. Клиент хочет положить туда `42`. Казалось бы, очень простая задача. Что может пойти не так?
|
||||
|
||||
|
||||
|
||||
Первый вариант — сервер вообще недоступен: процесс упал, машина выключена или ещё не начала принимать соединения. Второй — сервер жив, но запрос потерялся по дороге.
|
||||
|
||||
|
||||
|
||||
Но есть более неприятные случаи. Сервер получил запрос и начал его выполнять, а потом сломался. Успел он записать число или нет? Клиент этого не знает. Или сервер всё выполнил и отправил ответ, но уже ответ потерялся в сети. Число записано, а клиент об этом не узнает.
|
||||
|
||||
|
||||
|
||||
**Клиент не может отличить «сервер ничего не сделал» от «сервер всё сделал, но не ответил».** Таймаут сообщает только, что мы не дождались ответа – он не дает понять состояние сервера.
|
||||
|
||||
Можно попробовать ещё раз положить `42`: с точки зрения самого числа повторное присваивание даст тот же результат, если между попытками его никто не менял. Это идея идемпотентности. А вот с инкрементом такой приём уже не сработает: повтор увеличит число ещё раз. Можно спросить сервер, что в нём сейчас лежит, но и на этот запрос мы можем не получить ответа.
|
||||
|
||||
Как с этим жить? Для начала, хотелось бы вообще не думать о том, что пакеты могут теряться, дублироваться, фрагментироваться и изменяться. Посмотрим, какие абстракции для этого дают UDP и TCP, и какие проблемы всё равно останутся.
|
||||
|
||||
## Что дают UDP и TCP
|
||||
|
||||
### Откуда берутся проблемы с пакетами
|
||||
|
||||
Сеть ненадёжна по вполне конкретным причинам:
|
||||
|
||||
- **Переполнение буферов.** В маршрутизатор приходит 10 Гбит/с, а передать дальше он может только 1 Гбит/с. Очередь маршрутизатора конечна: когда она заполнится, новые пакеты придётся отбрасывать. Очередь может переполниться и на принимающем компьютере, если он не успевает обрабатывать данные.
|
||||
- **Отказ линии или устройства.** Кабель повредили при ремонте, оборудование выключилось, а маршрутизатор ещё отправляет пакеты по старому маршруту. Отправитель при этом не обязательно сразу узнает, что случилось.
|
||||
- **Повреждение данных.** Помеха изменила биты. Если проверка контрольной суммы обнаружит повреждение, пакет будет отброшен.
|
||||
- **Повторы и разные задержки.** Например, при потере подтверждения на одном из нижних уровней возможна повторная передача уже доставленных данных. Очереди и смена маршрута могут привести к тому, что более поздний пакет обгонит ранний. Сам IP не обещает ни порядка, ни защиты от дублей.
|
||||
- **Фрагментация.** Если IP-пакет больше допустимого размера на пути — MTU, — может потребоваться разбиение на фрагменты. Если потерялся один фрагмент, исходную датаграмму собрать уже нельзя.
|
||||
|
||||
Приложению неудобно разбираться с каждой такой ситуацией. Поэтому поверх IP используются транспортные протоколы. UDP сохраняет целостность отдельных сообщений, а TCP гарантирует порядок и надежность передачи.
|
||||
|
||||
### UDP
|
||||
|
||||
**UDP — User Datagram Protocol.** Мы отправляем датаграмму — отдельный массив байтов. На принимающей стороне сохраняется граница сообщений, даже вопреки фрагментации IP пакетов.
|
||||
|
||||
А вот **доставку, порядок и отсутствие дублей UDP не гарантирует**. Сам протокол не ждёт подтверждений и не повторяет потерянные датаграммы. Поэтому для нашей операции `PutValue` пришлось бы отдельно решать, как понять, дошёл ли запрос и что делать при потере.
|
||||
|
||||
Для других задач такие свойства подходят. Например, в конференции Zoom не обязательно, чтобы дошёл каждый кусочек голоса: небольшой пропуск наш мозг часто может восстановить по смыслу.
|
||||
|
||||
### TCP
|
||||
|
||||
**TCP — Transmission Control Protocol** — даёт абстракцию надёжного упорядоченного потока байтов. Можно представить канал: мы пишем в него байты, а другая сторона читает их в том же порядке, без повторов, дублей и потерь. Ниже этой абстракции все еще остается фрагментация пакетов, потери и повторы – но TCP все это скрывает от пользователя.
|
||||
|
||||
Сначала стороны устанавливают TCP-соединение. Каждая сторона должна сообщить ISN и другие свои параметры и получить подтверждение. Логически это четыре действия, но ответ сервера и его собственное начало обмена объединяются в одно сообщение. Получается три пакета: **SYN → SYN + ACK → ACK**.
|
||||
|
||||
|
||||
|
||||
После этого обе стороны хранят состояние: какие байты уже отправлены, какие подтверждены, какие получены и чего ещё не хватает. Соединение двунаправленное: клиент и сервер могут одновременно передавать данные, у каждого направления свой независимый поток байтов.
|
||||
|
||||
- **Sequence number (`seq`)** — номер первого байта в сегменте.
|
||||
- **Acknowledgment number (`ack`)** — номер следующего ожидаемого байта. Он подтверждает весь непрерывный префикс перед ним.
|
||||
- **Окно приёма** показывает, сколько данных получатель готов принять. Это позволяет отправлять несколько сегментов, не ожидая подтверждения каждого отдельно, но не переполнять буфер получателя. Дополнительно отправитель ограничивает объём данных в пути с учётом перегрузки сети.
|
||||
|
||||
Возьмём условную нумерацию с единицы. Пришли байты 4–6, а 1–3 потерялись. Получатель может сохранить 4–6 в буфере, но продолжает подтверждать, что ждёт байт 1. Отправитель обнаружит потерю по таймеру или повторным подтверждениям и повторит передачу. Когда появятся 1–3, образуется непрерывная последовательность 1–6: её можно отдать приложению, а в `ack` указать 7.
|
||||
|
||||
Если потерялось само подтверждение, отправитель тоже может повторить данные. Получатель узнает дубликат по номерам байтов и не отдаст его приложению второй раз. На схеме также упомянут **SACK**: он позволяет дополнительно сообщить, какие участки уже получены после пропуска.
|
||||
|
||||
|
||||
|
||||
За порядок приходится платить ожиданием: байты 4–6 уже у нас, но приложение не увидит их, пока не придут 1–3. Это **head-of-line blocking**.
|
||||
|
||||
## RPC
|
||||
|
||||
Теперь хочется избавиться от ручной работы: не собирать сообщения `PutValue`, не разбирать ответы и не писать всю сериализацию/десериализацию сообщений самостоятельно.
|
||||
|
||||
Посмотрим на Storage как на обычный объект. У него есть методы `PutValue` и `GetValue`: мы передаём запрос и получаем результат. Можно сделать так, чтобы клиентский код выглядел именно как вызов метода, а отправка сообщения на сервер происходила внутри, практически прозрачно для пользователя.
|
||||
|
||||
Такую астракцию дает **RPC — Remote Procedure Call, удалённый вызов процедуры**. Она находится на уровень выше TCP и UDP. Пока всё работает, нам почти не приходится думать о сети. Но при ошибке нужно вспомнить, что вызов шел через сеть, и сервер мог не ответить по разным причинам.
|
||||
|
||||
Операцию и её параметры можно записать в JSON. Но каждый раз передавать названия полей довольно неэффективно. В нашем примере используются **Protobuf** для описания и сериализации сообщений и **gRPC** — широко используемый фреймворк удалённых вызовов. Показанный gRPC-сервис работает поверх HTTP/2 и TCP. Protobuf определяет, как представить наши данные, а gRPC связывает вызов клиентского метода с обработчиком на сервере.
|
||||
|
||||
## Описываем Storage в `.proto`
|
||||
|
||||
Откроем [storage.proto](grpc-practice/go-server/proto/storage.proto). В начале указаны версия `proto3`, пакет `storage` и настройка `go_package` для генерации Go-кода. Дальше описан сам сервис:
|
||||
|
||||
```protobuf
|
||||
syntax = "proto3";
|
||||
|
||||
package storage;
|
||||
|
||||
option go_package = "hsegrpc/;storagepb";
|
||||
|
||||
import "google/protobuf/timestamp.proto";
|
||||
|
||||
service Storage {
|
||||
rpc PutValue(PutRequest) returns (PutResponse);
|
||||
rpc GetValue(GetRequest) returns (GetResponse);
|
||||
}
|
||||
|
||||
message Value {
|
||||
uint64 payload = 1;
|
||||
optional google.protobuf.Timestamp updated_at = 2;
|
||||
}
|
||||
|
||||
message PutRequest {
|
||||
Value value = 1;
|
||||
}
|
||||
|
||||
message PutResponse {
|
||||
uint64 value = 1;
|
||||
}
|
||||
|
||||
message GetRequest {
|
||||
}
|
||||
|
||||
message GetResponse {
|
||||
Value value = 1;
|
||||
}
|
||||
```
|
||||
|
||||
Здесь `rpc` объявляет метод. `PutValue` принимает `PutRequest` и возвращает `PutResponse`; с `GetValue` всё аналогично.
|
||||
|
||||
`PutRequest` содержит `Value` — значение, которое хотим положить. У `Value` есть само число `payload` и время обновления `updated_at`. `GetRequest` пустой, а `GetResponse` содержит `Value`. В `PutResponse` возвращается записанное число.
|
||||
|
||||
**Что означают единички и двойки после полей?** Это номера, по которым поля узнаются при сериализации. Вместо строки `payload` можно передавать номер `1`, а получатель по своей схеме знает, что это за поле. Номер относится к конкретному типу сообщения: `value = 1` в `PutRequest` и `payload = 1` в `Value` друг другу не мешают. Это не значение поля и не его позиция в исходном файле.
|
||||
|
||||
Эти номера важны для совместимости. Допустим, мы решили убрать `updated_at` и добавить другое поле. Нельзя просто отдать новому полю номер `2`: у пользователей старой версии под этим номером всё ещё описан `Timestamp`. Новому полю нужно дать новый номер, например `3`. Тогда это будет именно новое поле, а не другое значение на месте старого.
|
||||
|
||||
## Из `.proto` получаем код
|
||||
|
||||
Из одного `.proto` можно сгенерировать код для Go, Python, C++ и других языков. На семинаре мы смотрели Go; инструкции для разных языков есть на [grpc.io](https://grpc.io/docs/languages/).
|
||||
|
||||
В первой строке [main.go](grpc-practice/go-server/main.go) приведена команда `protoc` для генерации. Из каталога `grpc-practice/go-server` она выглядит так:
|
||||
|
||||
```bash
|
||||
protoc --go_out=. --go_opt=paths=source_relative \
|
||||
--go-grpc_out=. --go-grpc_opt=paths=source_relative \
|
||||
proto/storage.proto
|
||||
```
|
||||
|
||||
Для неё нужны `protoc` и Go-плагины `protoc-gen-go`, `protoc-gen-go-grpc`. Команда создаёт два файла:
|
||||
|
||||
- [`storage.pb.go`](grpc-practice/go-server/proto/storage.pb.go) — структуры `Value`, `PutRequest`, `PutResponse`, `GetRequest`, `GetResponse` и вспомогательные методы;
|
||||
- [`storage_grpc.pb.go`](grpc-practice/go-server/proto/storage_grpc.pb.go) — клиентский и серверный интерфейсы, а также код для работы с gRPC.
|
||||
|
||||
### Сообщения становятся структурами
|
||||
|
||||
В `storage.pb.go` каждому `message` соответствует Go-структура. Вот все типы из нашего примера; для наглядности здесь опущены служебные поля генератора, теги и вспомогательные методы:
|
||||
|
||||
```go
|
||||
type Value struct {
|
||||
Payload uint64
|
||||
UpdatedAt *timestamppb.Timestamp
|
||||
}
|
||||
|
||||
type PutRequest struct {
|
||||
Value *Value
|
||||
}
|
||||
|
||||
type PutResponse struct {
|
||||
Value uint64
|
||||
}
|
||||
|
||||
type GetRequest struct{}
|
||||
|
||||
type GetResponse struct {
|
||||
Value *Value
|
||||
}
|
||||
```
|
||||
|
||||
Можно прямо сопоставить их с `.proto`: `payload` превратился в `Payload`, `updated_at` — в `UpdatedAt`, а вложенное сообщение `Value` — в поле типа `*Value`. Поэтому запрос «положить 42» мы собираем как обычную структуру:
|
||||
|
||||
```go
|
||||
request := &storagepb.PutRequest{
|
||||
Value: &storagepb.Value{
|
||||
Payload: 42,
|
||||
},
|
||||
}
|
||||
```
|
||||
|
||||
### Клиент
|
||||
|
||||
Наверное, самый сложный момент: **`StorageClient`**. Имея его экземпляр, пользователь может вызвать методы `PutValue` и `GetValue` сервиса `Storage`. Его интерфейс сгенерирован в `storage_grpc.pb.go` (упрощенно):
|
||||
|
||||
```go
|
||||
type StorageClient interface {
|
||||
PutValue(in *PutRequest) (*PutResponse, error)
|
||||
GetValue(in *GetRequest) (*GetResponse, error)
|
||||
}
|
||||
```
|
||||
|
||||
Получить такой клиент позволяет сгенерированный `NewStorageClient`. Если `conn` — уже созданное gRPC-подключение, клиентский код выглядит так:
|
||||
|
||||
```go
|
||||
client := storagepb.NewStorageClient(conn)
|
||||
|
||||
response, err := client.PutValue(request)
|
||||
if err != nil {
|
||||
log.Fatal(err)
|
||||
}
|
||||
fmt.Println(response.GetValue()) // 42
|
||||
```
|
||||
|
||||
### Сервер
|
||||
|
||||
**`StorageServer`** задаёт интерфейс, который серверсу предстоит реализовать. Генератор не знает, как именно наш сервис `Storage` хранит число. Есть заготовка `UnimplementedStorageServer`: её методы отвечают, что операция не реализована. Дальше мы пишем собственные `PutValue` и `GetValue`. Здесь проходит важная граница: описание сервиса и сетевой обмен можно сгенерировать, а решение, что делать с пришедшим значением, остаётся за нами.
|
||||
|
||||
Вот интерфейс и один из методов заготовки из `storage_grpc.pb.go` (упрощенно):
|
||||
|
||||
```go
|
||||
type StorageServer interface {
|
||||
PutValue(*PutRequest) (*PutResponse, error)
|
||||
GetValue(*GetRequest) (*GetResponse, error)
|
||||
mustEmbedUnimplementedStorageServer()
|
||||
}
|
||||
|
||||
type UnimplementedStorageServer struct{}
|
||||
|
||||
func (UnimplementedStorageServer) PutValue(
|
||||
_ *PutRequest,
|
||||
) (*PutResponse, error) {
|
||||
return nil, status.Errorf(
|
||||
codes.Unimplemented, "method PutValue not implemented",
|
||||
)
|
||||
}
|
||||
```
|
||||
|
||||
Свою реализацию мы пишем в [storage/server.go](grpc-practice/go-server/storage/server.go). Наш `Server` включает `UnimplementedStorageServer`, хранит значение и mutex. Клиенты могут обращаться одновременно, поэтому доступ к значению защищён блокировкой:
|
||||
|
||||
```go
|
||||
type Value struct {
|
||||
payload uint64
|
||||
updatedAt time.Time
|
||||
}
|
||||
|
||||
type Server struct {
|
||||
storagepb.UnimplementedStorageServer
|
||||
valueLocker sync.RWMutex
|
||||
value Value
|
||||
}
|
||||
|
||||
func NewServer() *Server {
|
||||
return &Server{}
|
||||
}
|
||||
```
|
||||
|
||||
Здесь `Value` — наша внутренняя структура для хранения числа; `storagepb.Value` из предыдущих листингов — сообщение для передачи по сети. Встраивание `UnimplementedStorageServer` даёт заготовки методов и служебный метод из интерфейса, а нужные `PutValue` и `GetValue` мы определяем у своего `Server`.
|
||||
|
||||
В `PutValue` приходит сгенерированный `PutRequest`. Сначала проверяем, передал ли клиент `value`: он может его не указать, и тогда `request.GetValue()` вернёт `nil`. Если поле нужно для операции, эту проверку делаем сами. Иначе записываем число, выставляем время обновления и возвращаем ответ:
|
||||
|
||||
```go
|
||||
func (s *Server) PutValue(
|
||||
request *storagepb.PutRequest,
|
||||
) (*storagepb.PutResponse, error) {
|
||||
s.valueLocker.Lock()
|
||||
defer s.valueLocker.Unlock()
|
||||
|
||||
if request.GetValue() == nil {
|
||||
return nil, errors.New("missed value")
|
||||
}
|
||||
|
||||
s.value = Value{
|
||||
payload: request.GetValue().GetPayload(),
|
||||
updatedAt: time.Now(),
|
||||
}
|
||||
|
||||
return &storagepb.PutResponse{
|
||||
Value: s.value.payload,
|
||||
}, nil
|
||||
}
|
||||
```
|
||||
|
||||
`GetValue` берёт блокировку на чтение и возвращает текущее значение. Наш вспомогательный метод `toProto` перекладывает его в сгенерированное Protobuf-сообщение:
|
||||
|
||||
```go
|
||||
func (s *Server) GetValue(
|
||||
_ *storagepb.GetRequest,
|
||||
) (*storagepb.GetResponse, error) {
|
||||
s.valueLocker.RLock()
|
||||
defer s.valueLocker.RUnlock()
|
||||
|
||||
return &storagepb.GetResponse{
|
||||
Value: s.value.toProto(),
|
||||
}, nil
|
||||
}
|
||||
|
||||
func (v *Value) toProto() *storagepb.Value {
|
||||
return &storagepb.Value{
|
||||
Payload: v.payload,
|
||||
UpdatedAt: timestamppb.New(v.updatedAt),
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Соединяем реализацию с gRPC-сервером
|
||||
|
||||
Остаётся запустить сервер, используя `main.go`. Ниже основной фрагмент запуска; `addr` — адрес, который в нашем примере по умолчанию равен `0.0.0.0:51000`:
|
||||
|
||||
```go
|
||||
lis, err := net.Listen("tcp", addr)
|
||||
if err != nil {
|
||||
log.Fatalf("failed to listen: %v", err)
|
||||
}
|
||||
|
||||
grpcServer := grpc.NewServer()
|
||||
reflection.Register(grpcServer)
|
||||
|
||||
storageService := storage.NewServer()
|
||||
storagepb.RegisterStorageServer(grpcServer, storageService)
|
||||
|
||||
err = grpcServer.Serve(lis)
|
||||
if err != nil {
|
||||
log.Fatalf("server failed")
|
||||
}
|
||||
```
|
||||
|
||||
`net.Listen("tcp", addr)` открывает TCP-listener. `grpc.NewServer()` создаёт сервер gRPC, а `storage.NewServer()` — нашу реализацию Storage. Вызов `RegisterStorageServer(grpcServer, storageService)` связывает их: теперь gRPC знает, какому объекту передавать вызовы `PutValue` и `GetValue`. `reflection.Register` позволит инструментам вроде `grpcurl` узнать описание сервиса. Наконец, `grpcServer.Serve(lis)` начинает принимать соединения.
|
||||
|
||||
Получается полный путь одного запроса: клиентский метод получает `PutRequest` → сообщение сериализуется и передаётся по сети → сервер gRPC разбирает его → вызывает наш `Server.PutValue` → тот меняет значение и возвращает `PutResponse` → ответ уходит клиенту. Сериализацию и передачу обеспечивает готовый код, а изменение значения написано нами в `server.go`.
|
||||
|
||||
## Проверяем через grpcurl и Postman
|
||||
|
||||
Для этого примера нужны Go и установленный [`grpcurl`](https://github.com/fullstorydev/grpcurl). В каталоге `grpc-practice/go-server` запускаем сервер:
|
||||
|
||||
```bash
|
||||
go run main.go
|
||||
```
|
||||
|
||||
В другом терминале смотрим список сервисов и описание Storage:
|
||||
|
||||
```bash
|
||||
grpcurl -plaintext localhost:51000 list
|
||||
grpcurl -plaintext localhost:51000 describe storage.Storage
|
||||
```
|
||||
|
||||
`-plaintext` отключает TLS: в этом примере работаем без шифрования. `list` показывает доступные сервисы, а `describe` — методы Storage и типы их запросов и ответов. Узнать это у работающего сервера позволяет включённый в `main.go` сервис reflection: `grpcurl` получает описание интерфейса, хотя мы не передавали ему `.proto` отдельным файлом.
|
||||
|
||||
Теперь положим `100500`, как в демонстрации, и прочитаем значение обратно:
|
||||
|
||||
```bash
|
||||
grpcurl -plaintext -d '{"value":{"payload":100500}}' \
|
||||
localhost:51000 storage.Storage/PutValue
|
||||
grpcurl -plaintext -d '{}' localhost:51000 storage.Storage/GetValue
|
||||
```
|
||||
|
||||
В запросе повторяется структура из `.proto`: внутри `value` находится `payload`. `PutValue` возвращает записанное число, `GetValue` — текущее значение и время обновления. В терминале мы вводим JSON, потому что это удобно человеку; `grpcurl` по описанию сервиса преобразует его в Protobuf-сообщение.
|
||||
|
||||
У `PutValue` ожидается ответ `{"value":"100500"}`, а у `GetValue` — вложенный объект `value` с полями `payload` и `updatedAt`. Время выставляет наш сервер. Если к нему одновременно обращается другой пишущий клиент, чтение может уже показать его значение.
|
||||
|
||||
Если не хочется писать команды в консоли, можно использовать **Postman**, как в конце семинара: импортировать `storage.proto`, указать адрес `localhost:51000`, выбрать `PutValue`, заполнить сообщение и нажать **Invoke**. Затем вызвать `GetValue` и увидеть записанное число.
|
||||
|
||||
Получилось, что мы описали сервис в `.proto`, сгенерировали код и дописали небольшую реализацию — и у нас уже есть работающий сервис. Все файлы и команды находятся в [материалах практики](grpc-practice/go-server/README.md).
|
||||
@@ -7,6 +7,8 @@
|
||||
|
||||
### Семинар
|
||||
|
||||
- [Презентация](02-communication-seminar.pptx)
|
||||
- [Конспект: TCP/UDP, gRPC и Protobuf](notes.md)
|
||||
- [Практика с gRPC](grpc-practice)
|
||||
- [Пример gRPC Streaming](grpc-streaming)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user