ruststream 0.7.1

# Memory

`MemoryBroker` за фичей `memory` - полноценный брокер внутри процесса. Он подходит, когда очередь
нужна в рамках одного приложения, а не в сети. На нём построен шаблон `cargo generate` по умолчанию
(`templates/memory`), поэтому свежему проекту не нужны ни сервер, ни отдельный крейт с
адаптацией библиотеки брокера.

```toml
ruststream = { version = "0.7", features = ["macros", "memory", "json"] }
```

<!-- inline-rust: two-line constructor sketch; the broker in context is exercised by every memory-feature example (e.g. quickstart.rs:app) -->
```rust
use ruststream::memory::MemoryBroker;

let broker = MemoryBroker::new();
```

## Сколько он хранит { #retention }

Брокер из `MemoryBroker::new()` не хранит ничего. Сообщение живёт от публикации до чтения последним
подписчиком, поэтому долгоживущий сервис держит в памяти только ту работу, которую обработчики ещё
не сделали, сколько бы сообщений через него ни прошло.

Сервису с перемоткой нужна история, и он называет её объём:

```rust
--8<-- "examples/seek.rs:retaining"
```

`Retention` ограничивает одну тему: `Messages(n)` хранит `n` свежих сообщений каждой темы,
`Bytes(n)` - свежие тела, которые помещаются в `n` байт, а `MessagesAndBytes { .. }` применяет оба
предела сразу. Брокер, публикующий в тысячу тем, держит столько на каждую из них. Самое свежее
сообщение остаётся всегда, поэтому тело шире предела по байтам хранится в одиночку, а не пропадает в
момент публикации.

Две формы брокера - разные типы. `MemoryBroker::retaining(..)` даёт ту, у которой подписки
реализуют `Seekable`: монтирование со стартовой позицией или с ручкой перемотки компилируется только
против неё. Перемотка на брокере без журнала - ошибка компиляции, а не повтор, который молча ничего
не находит.

## Прелюдия, которую импортирует точка монтирования { #prelude }

Точка монтирования импортирует прелюдию брокера одной строкой:

<!-- inline-rust: the import shape; every memory-feature example under examples/ mounts through it -->
```rust
use ruststream::memory::prelude::*;
```

`ruststream::memory::prelude` устроена так же, как прелюдия любой адаптации библиотеки
брокера. Она реэкспортирует три слоя:

- прелюдию ядра;
- типы брокера;
- политики публикации под общими для всех брокеров именами.

Типы брокера - это `MemoryBroker` и его `Routing`, источники подписки `MemorySource` и
`MemoryPattern`, `MemoryError`, `MemoryPosition`, `Retention` с режимами журнала `Discarding` и
`Retaining`, ключи контекста `MemoryContext`, `MemoryBatchContext`, `Position` и `SeekHandle`.

Политики называются `Publish`, `TransactionalPublish` и `Request`. Это псевдонимы `MemoryPublish` и
`MemoryRequest`.

**Транзакции.** Издатель этого брокера реализует обе разновидности транзакций. Поэтому
`TransactionalPublish` здесь - та же политика, что `Publish`. У брокера с отдельной транзакционной
конфигурацией это имя указывает на другую политику.

**Совместимости.** Трейты-совместимости `TransactionalPublisher`, `OwnedTransactions`,
`Transaction`, `RequestReply`, `Positioned` и `Seeker` прелюдия тоже вводит в область видимости. Их
операции доступны там же, где политики.

**`Partitioned`** импортируйте сами, если читаете ключи партиционирования. У `IncomingMessage` есть
одноимённый метод, и в прелюдии `Partitioned` сделал бы вызов `msg.partition_key()`
неоднозначным.

**Тело обработчика** импортирует `use ruststream::prelude::*;`. Оно работает с совместимостями, а не
с политиками, и не знает, на каком брокере выполняется. Если тело и точка монтирования лежат в
одном файле, хватит одной прелюдии брокера.

## Семантика

- **Тема сравнивается по имени целиком.** Подписка на `orders` получает сообщения, опубликованные в
  `orders`. [Подписка по шаблону](#patterns) читает все подходящие темы.
- **Доставка всем подписчикам.** Каждый подписчик темы получает каждое сообщение, опубликованное
  после подписки. Какие шаблоны тоже его получат, решает `Routing` брокера.
- **Доставку завершают `ack` и `nack(requeue: false)`.** `nack(requeue: true)` и доставка,
  отпущенная без завершения, повторно доставляют ту же полезную нагрузку тому же подписчику.
- **`retry_after` брокер выполняет сам.** По истечении задержки он повторно доставляет сообщение
  тому же подписчику, и заново ничего не публикуется.
- **Доставки считаются.** Каждая доставка сообщает, в который раз брокер отдаёт это сообщение
  подписчику. Этот счётчик расходует предел `max_attempts(n)`, а исчерпавшая его доставка уходит в DLQ,
  заданную через `dead_letter(name)`.
- **Совместное владение.** `MemoryBroker` - дескриптор с подсчётом ссылок: все его копии работают с
  одним состоянием. Копия в тесте видит всё, что публикует приложение.

Рантайм доставляет сообщения одним путём для всех брокеров, поэтому обработчик, middleware и
декодирование работают здесь так же, как с сетевым брокером.

## Совместимости {#capabilities}

Следующие трейты-совместимости `MemoryBroker` реализует на своей внутрипроцессной семантике.

**Запрос и ответ.** `broker.requester()` возвращает `MemoryRequester`. Его `request` публикует
сообщение с уникальной внутрипроцессной темой в заголовке `reply-to` и ждёт первое сообщение,
доставленное в эту тему. Отвечающая сторона читает `reply-to` из запроса и публикует ответ туда.
Если никто не ответил, запрос возвращает ошибку `RequestError::Timeout`.

`MemoryRequester` конструирует политика `MemoryRequest`. К ней вы и привязываете слот с
ограничением `Out<impl RequestReply, ..>`.

**Пакеты.** `MemorySubscriber` реализует `BatchSubscriber`. В пакет попадает первая пришедшая
доставка и всё, что уже ждёт в буфере. Размер ограничивает `batch(n)`, заданный при регистрации
обработчика. Неполный пакет доставляется сразу.

**Транзакции.** `MemoryPublisher` реализует обе разновидности транзакций, а конструирует его
политика `MemoryPublish`. К ней вы и привязываете слот или связывание с ограничением
`TransactionalPublisher` или `OwnedTransactions`.

- Публикации внутри транзакции копятся в буфере. `commit` доставляет их всем подписчикам разом в
  порядке публикации, `abort` отбрасывает.
- У каждой владеющей транзакции свой буфер, копии дескриптора издателя его не разделяют.
- Вызовы не по порядку возвращают `MemoryError`. Второй `begin_transaction` при открытой
  транзакции возвращает `TransactionBusy`, и открытая транзакция остаётся нетронутой. `commit` или
  `abort` без транзакции возвращает `NoTransaction`.

**Ключи партиционирования.** `MemoryMessage` реализует `Partitioned` и берёт ключ из заголовка
`partition-key` (`memory::PARTITION_KEY_HEADER`).

**Перемотка журнала.** У [хранящего брокера](#retention) `MemorySubscriber` реализует `Seekable`
поверх журнала каждой темы. Чтобы перемотать подписку, получите `MemorySeeker` до начала чтения и
вызовите `seek` с позицией `MemoryPosition`.

- Позицию можно снять с доставленного сообщения через `Positioned::position`, и тогда это
  сообщение доставится заново.
- Позицию можно и построить: `MemoryPosition::start()` - самое старое сохранённое сообщение,
  `end()` - конец журнала после всего опубликованного, `sequence(n)` - сообщение с номером `n`.
- Номера абсолютны: номер называет то же сообщение, даже когда предел хранения вытесняет более
  старые. Для вытесненного номера `seek` возвращает ошибку `MemoryError::PositionEvicted` с самой
  старой оставшейся позицией.
- Перемотка вперёд пропускает доставки, которые стояли в очереди до цели.
- Перемотка действует на один экземпляр подписчика, а через дескриптор остановленной шины
  возвращает ошибку `MemoryError::ShutDown`.
- В приложении позицию сообщения и `MemorySeeker` содержит `MemoryContext`, а обработчик читает их
  по ключам `Position` и `SeekHandle` (см.
  [Перемотку](https://docs.rs/ruststream/latest/ruststream/runtime/index.html#seeking)).
- Пакетный обработчик читает `MemoryBatchContext`: в нём есть `SeekHandle`, а `Position` нет,
  потому что пакет охватывает много доставок.

**Остановка.** `MemoryBroker::connect(self)` возвращает `ConnectedMemoryBroker`. Его `shutdown`
поглощает `self` и возвращает `ClosedMemoryBroker` с числом регистраций подписчиков, отброшенных
при остановке. После остановки публикация, фиксация транзакции и запрос через ранее выданные
дескрипторы возвращают ошибку `MemoryError::ShutDown` или `RequestError::ShutDown`.

## Источник подписки

`ConnectedMemoryBroker` реализует `Subscribe`. Поэтому `#[subscriber("orders")]` работает без
дескриптора.

Ту же подписку можно задать дескриптором `MemorySource`, как у любого брокера.

Пример из
[`routed_service`](https://github.com/powersemmi/ruststream/tree/main/examples/routed_service):

=== "Макросы"

    ```rust
    use ruststream::memory::prelude::*;

    --8<-- "examples/routed_service/orders.rs:descriptor"
    ```

=== "Вручную"

    ```rust
    use ruststream::memory::prelude::*;

    --8<-- "examples/manual/routed_service_orders.rs:descriptor"
    ```

## Подписка по шаблону { #patterns }

<!-- inline-rust: the mount-site shape; the compiled twin is the doctest of the `memory` module overview on docs.rs -->
```rust
use ruststream::memory::prelude::*;

#[subscriber("orders.eu")]
async fn europe(order: &Order) -> HandlerOutcome {
    HandlerOutcome::ack()
}

#[subscriber(MemoryPattern::new("orders.*"))]
async fn other_regions(order: &Order) -> HandlerOutcome {
    HandlerOutcome::ack()
}

fn app() -> RustStream {
    let broker = MemoryBroker::new().routing(Routing::MostSpecific);
    RustStream::new(AppInfo::new("orders", "0.1.0")).with_broker(broker, |b| {
        b.include(europe);
        b.include(other_regions);
    })
}
```

`MemoryPattern` подписывается на шаблон. Синтаксис шаблона тот же, что у тем NATS. Имя темы
делится на токены по точке. Токен `*` подходит ровно к одному токену, а `>` в конце шаблона - к
одному токену или больше. Так, `orders.*` читает `orders.eu` и `orders.us`, а `orders.>` читает
ещё и `orders.eu.created`. Брокер проверяет шаблон, когда открывает подписку. С неверным шаблоном
сервис не запускается: он получает ошибку `MemoryError::InvalidPattern`, и в ней указан шаблон.
Подписка по имени читает одну тему, поэтому подстановочный токен в имени тоже останавливает запуск,
с ошибкой `MemoryError::WildcardName`.

Кто получит сообщение, под которое подходят несколько подписок, решает `Routing`:

- **`Routing::EveryMatch`** действует по умолчанию и доставляет сообщение всем подходящим
  подпискам, как NATS. Выбирайте его, когда шаблон читает рядом с обработчиками отдельных тем:
  журнал аудита, сбор метрик.
- **`Routing::MostSpecific`** доставляет сообщение подписчикам точного имени темы. Если таких
  нет, сообщение получает самый точный из подходящих шаблонов. Выбирайте его, когда шаблон
  подбирает темы, у которых нет своего обработчика.

Шаблоны сравниваются по токенам слева направо. На первом токене, где они расходятся, буквальный
токен точнее `*`, а `*` точнее `>`. Для публикации в `orders.eu.created` шаблон `orders.eu.*`
точнее `orders.*.created`, тот точнее `orders.>`, а `orders.>` точнее `*.eu.created`.

Брокер без подписок по шаблону сопоставляет только точное имя темы и за шаблоны ничего не платит.
Подписку по шаблону нельзя перемотать даже на хранящем брокере: журнал ведётся по каждой теме
отдельно.

## Для тестов

Приложение на `MemoryBroker` проверяется обвязкой
[`TestApp`](https://docs.rs/ruststream/latest/ruststream/testing/index.html). Тест передаёт
приложение в `TestApp::start`, публикует сообщения и проверяет, что обработчики получили и что
опубликовали. Полный пример разобран в разделе
[Тестирование](https://docs.rs/ruststream/latest/ruststream/testing/index.html#examples).

**Маршрутизация.** Обвязка учитывает `Routing` брокера: публикация ждёт ровно те подписки, которые
выбирает правило. Так работают и `TestApp::start`, и `TestApp::start_live`.

**Журнал публикаций.** Во время прогона обвязка записывает всё, что публикует сервис. Поэтому
`published::<T>(..)` читает один и тот же список на любой форме брокера.

Вне обвязки `TestableBroker::published` показывает то, что брокер хранит. Хранящий брокер
показывает всё в пределах своего ограничения, брокер по умолчанию - ничего.