Устройство

Где стоит ruststream-nats: между рантаймом RustStream и клиентом async-nats, все подписки и издатели на одном соединении.

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

1. Ваш сервис: #[subscriber] обработчики-подписчики; RustStream::new(..) .with_broker(..)
2. Ядро RustStream: Рантайм: диспетчер, партиции воркеров, завершение, переобработка
3. ruststream-nats: NatsSubscriber выдаёт NatsMessage; NatsPublisher, JetStreamPublisher; Ключи контекста; NATS внутри процесса; NatsBroker одно соединение; CoreSubject, CoreWildcard, JetStreamSubject
4. async-nats: jetstream::Context на клиенте; Client: одно соединение
5. #[subscriber] обработчики-подписчики → Рантайм: диспетчер, партиции воркеров, завершение, переобработка: монтируются
6. Рантайм: диспетчер, партиции воркеров, завершение, переобработка → NatsSubscriber выдаёт NatsMessage: опрашивает поток
7. NatsSubscriber выдаёт NatsMessage → Client: одно соединение
8. NatsPublisher, JetStreamPublisher → Client: одно соединение
9. Client: одно соединение → Сервер NATS

Крейт реализует трейты брокера из ядра поверх async-nats и ничего не добавляет к протоколу. Рантайм над ним один для всех брокеров. Всё, что относится к NATS, лежит в этом крейте: дескрипторы, смысл ack, адрес копии для переобработки и метаданные доставки.

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

Контекст JetStream строится на рантайме, на котором подключился брокер. Задачи, которые клиент запускает для него (цикл pull-запросов консьюмера и задача подтверждений), выполняются там, из какого бы потока ни пришёл вызов.

Настройка, для которой в крейте нет типа, - это параметр подключения async-nats. Его задаёт with_options до того, как начнётся подключение. client и jetstream выдают клиент и контекст JetStream для того, что крейт не оборачивает, например для создания стрима. from_client принимает клиент, созданный в другом месте.

Как части запускаются и останавливаются, показывает страница Жизненный цикл, а путь одного сообщения - страница Путь.

Транспорт внутри процесса

С фичей testing TestApp::start подключает тот же брокер к транспорту внутри процесса, а не к серверу. Дескрипторы, политики и доставки остаются теми же, что в рабочей сборке. Транспорт сопоставляет субъекты с * и >, распределяет сообщения очереди, отвечает на запрос «no responders» и хранит стримы с их порядковыми номерами, окном дубликатов и ожиданиями, которые объявляет публикация. Его консьюмеры доставляют с позиции своей политики доставки, сообщают метаданные, завершают и откладывают доставки, а после ack_wait выполняют повторную доставку, как это делает сервер.

То, что принадлежит самому серверу, проверяет только живой прогон: список субъектов и пределы стрима, max_ack_pending, пределы ожидающих сообщений подписки, аутентификацию и TLS. Внутри процесса подписка находит любой стрим, который называет. С сервером несуществующий стрим прерывает старт.

Без фичи транспорт не компилируется. Подключённый брокер рабочей сборки - это один общий указатель на соединение, и компилятор проверяет его размер.