Compare commits
1
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
 syakovlev
|
11f14108b4 |
@@ -1,5 +0,0 @@
|
|||||||
target
|
|
||||||
**/target
|
|
||||||
**/__pycache__
|
|
||||||
**/.pytest_cache
|
|
||||||
*.pyc
|
|
||||||
@@ -1,125 +0,0 @@
|
|||||||
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
|
|
||||||
@@ -1,296 +0,0 @@
|
|||||||
# Мессенджер на 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).
|
|
||||||
@@ -1 +0,0 @@
|
|||||||
# TODO: Write Docker file for client
|
|
||||||
@@ -1,5 +0,0 @@
|
|||||||
syntax = "proto3";
|
|
||||||
|
|
||||||
package mes_grpc;
|
|
||||||
|
|
||||||
// TODO: Add messages and service
|
|
||||||
@@ -1 +0,0 @@
|
|||||||
# TODO: Write Docker file for server
|
|
||||||
@@ -1,11 +0,0 @@
|
|||||||
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"]
|
|
||||||
@@ -1,12 +0,0 @@
|
|||||||
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
|
|
||||||
)
|
|
||||||
@@ -1,80 +0,0 @@
|
|||||||
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)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
@@ -1,3 +0,0 @@
|
|||||||
module github.com/distsys-course/grpc-messenger/grpc
|
|
||||||
|
|
||||||
go 1.23
|
|
||||||
@@ -1,7 +0,0 @@
|
|||||||
syntax = "proto3";
|
|
||||||
|
|
||||||
package mes_grpc;
|
|
||||||
|
|
||||||
option go_package = "proto/;mes_grpc";
|
|
||||||
|
|
||||||
// TODO: Add messages and service
|
|
||||||
@@ -1,60 +0,0 @@
|
|||||||
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)
|
|
||||||
@@ -1,3 +0,0 @@
|
|||||||
aiohttp==3.12.15
|
|
||||||
grpcio==1.75.0
|
|
||||||
grpcio-tools==1.75.0
|
|
||||||
@@ -1,11 +0,0 @@
|
|||||||
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"]
|
|
||||||
@@ -1,102 +0,0 @@
|
|||||||
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()
|
|
||||||
@@ -1,2 +0,0 @@
|
|||||||
grpcio==1.75.0
|
|
||||||
grpcio-tools==1.75.0
|
|
||||||
@@ -1,5 +0,0 @@
|
|||||||
syntax = "proto3";
|
|
||||||
|
|
||||||
package mes_grpc;
|
|
||||||
|
|
||||||
// TODO: Add messages and service
|
|
||||||
@@ -1,21 +0,0 @@
|
|||||||
# 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"]
|
|
||||||
@@ -1,109 +0,0 @@
|
|||||||
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())
|
|
||||||
@@ -1,5 +0,0 @@
|
|||||||
grpcio==1.75.0
|
|
||||||
grpcio-tools==1.75.0
|
|
||||||
protobuf==6.32.1
|
|
||||||
pytest==8.4.2
|
|
||||||
requests==2.32.5
|
|
||||||
@@ -1,179 +0,0 @@
|
|||||||
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
|
|
||||||
@@ -1,140 +0,0 @@
|
|||||||
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'),
|
|
||||||
)
|
|
||||||
@@ -1,285 +0,0 @@
|
|||||||
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.
@@ -7,7 +7,6 @@
|
|||||||
|
|
||||||
### Семинар
|
### Семинар
|
||||||
|
|
||||||
- [Презентация](02-communication-seminar.pptx)
|
|
||||||
- [Конспект: TCP/UDP, gRPC и Protobuf](notes.md)
|
- [Конспект: TCP/UDP, gRPC и Protobuf](notes.md)
|
||||||
- [Практика с gRPC](grpc-practice)
|
- [Практика с gRPC](grpc-practice)
|
||||||
- [Пример gRPC Streaming](grpc-streaming)
|
- [Пример gRPC Streaming](grpc-streaming)
|
||||||
|
|||||||
Reference in New Issue
Block a user