Путь

Путь одного сообщения через сервис RustStream, на файле потока и в канале: чтение, очередь, обработчик, ответ и запись.

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

1. Файл потока: Консьюмер клиента читает файл; Управляющая задача метит поколением; Очередь на 64; FileSubscriber отсев старых поколений; Сборка пакета под batch(n); FileContext, затем декодирование; Обработчик-подписчик #[subscriber] fn; FilePublisher конверт при заголовках; Дозапись и сброс в файл
2. Канал: Поток stdin разбирает строку; Задача чтения подписки; Очередь на 64; StdioSubscriber по сообщению на опрос; Сборка пакета под batch(n); Декодирование; Обработчик-подписчик #[subscriber] fn; StdioPublisher конверт, если нужно; Поток stdout пишет строку
3. Консьюмер клиента читает файл → Управляющая задача метит поколением
4. Управляющая задача метит поколением → Очередь на 64
5. Очередь на 64 → FileSubscriber отсев старых поколений
6. FileSubscriber отсев старых поколений → Сборка пакета под batch(n)
7. Сборка пакета под batch(n) → FileContext, затем декодирование
8. FileContext, затем декодирование → Обработчик-подписчик #[subscriber] fn
9. Обработчик-подписчик #[subscriber] fn → FilePublisher конверт при заголовках
10. FilePublisher конверт при заголовках → Дозапись и сброс в файл
11. Поток stdin разбирает строку → Задача чтения подписки
12. Задача чтения подписки → Очередь на 64
13. Очередь на 64 → StdioSubscriber по сообщению на опрос
14. StdioSubscriber по сообщению на опрос → Сборка пакета под batch(n)
15. Сборка пакета под batch(n) → Декодирование
16. Декодирование → Обработчик-подписчик #[subscriber] fn
17. Обработчик-подписчик #[subscriber] fn → StdioPublisher конверт, если нужно
18. StdioPublisher конверт, если нужно → Поток stdout пишет строку

Задача подписки читает с опережением обработчика. Она забирает у клиента каждое сообщение, открывает конверт и копирует полезную нагрузку в собственный буфер. Номер сообщения она добавляет в заголовок stream-sequence и ставит доставку в очередь, в которой помещается до 64 доставок. Обработчик читает из этой очереди, поэтому следующее сообщение уже прочитано из файла, пока обрабатывается текущее.

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

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

Доставка завершается после ответа. На обоих транспортах ack и nack возвращают ошибку: операция не поддерживается. Это показывает страница Завершение.