Жизненный цикл

KinesisBroker::new записывает конфигурацию, connect загружает конфигурацию AWS, каждая подписка открывает свои сегменты при старте, а завершение работы закрывает соединение.

Схема текстом

1. Брокер: KinesisBroker::new(); ConnectedKinesis Broker; Closed = () сообщать нечего
2. В AWS: Только конфигурация, без ввода-вывода; Конфигурация AWS, клиент, хранилище аренд, id владельца; Каждая подписка: DescribeStreamSummary или CreateStream, ListShards, аренда и итератор на сегмент; Сегменты перечисляются каждые 10 с, аренды продлеваются каждые 3 с; Соединение помечено закрытым
3. KinesisBroker::new() → ConnectedKinesis Broker: connect(self)
4. ConnectedKinesis Broker → Closed = () сообщать нечего: shutdown(self)
5. Только конфигурация, без ввода-вывода → Конфигурация AWS, клиент, хранилище аренд, id владельца
6. Конфигурация AWS, клиент, хранилище аренд, id владельца → Каждая подписка: DescribeStreamSummary или CreateStream, ListShards, аренда и итератор на сегмент
7. Каждая подписка: DescribeStreamSummary или CreateStream, ListShards, аренда и итератор на сегмент → Сегменты перечисляются каждые 10 с, аренды продлеваются каждые 3 с
8. Сегменты перечисляются каждые 10 с, аренды продлеваются каждые 3 с → Соединение помечено закрытым
9. Издатель или дескриптор перемотки, переживший соединение, отвечает KinesisError::NotConnected, а не молчаливым успехом

new, from_config, endpoint, region, test_credentials, lease_store и owner_id не выполняют ввода-вывода, поэтому билдер сервиса остаётся синхронным. Копия брокера разделяет с ним соединение. Издатель, полученный от брокера до подключения, работает, как только рантайм подключит брокер.

Подключение загружает конфигурацию AWS через стандартную цепочку SDK, с конечной точкой и регионом, которые задал брокер. Брокер, созданный через from_config, берёт свой SdkConfig как есть. Затем подключение создаёт клиент Kinesis. Без lease_store брокер хранит аренды в MemoryLeaseStore, а без owner_id получает собственный уникальный идентификатор владельца аренды. SDK получает учётные данные лениво, при первом запросе. В сервисе первый запрос - это DescribeStreamSummary первой подписки, поэтому неверные учётные данные прерывают старт на подписке.

Сначала проверяется KinesisStream каждой подписки. Пустое имя или create_if_missing без сегментов дают ошибку KinesisError::Invalid до любого обращения к AWS. Поток, который брокер уже читает через другую подписку, тоже отвергается. Затем подписка запрашивает поток вызовом DescribeStreamSummary. Отсутствующий поток - это ошибка KinesisError::Stream, если только create_if_missing не создаёт его и не ждёт его готовности до минуты.

Подписка возвращает управление, когда уже читает. Первый проход координатора перечисляет сегменты, берёт аренду каждого сегмента, который может взять, и запускает по читателю на сегмент. Каждый из этих читателей получает свой первый итератор до того, как subscribe вернёт управление. Поэтому конец сегмента, с которого он открывается, - это момент открытия подписки, и запись, опубликованная после него, будет прочитана. Ошибка первого прохода приходит в поток доставок подписки, когда тот начнёт выполняться.

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

Завершение работы помечает соединение закрытым и возвращает (): клиенту SDK нечего закрывать и нечего отправлять. Издатель или дескриптор перемотки, созданный до завершения работы, после него отвечает ошибкой KinesisError::NotConnected.