Чтение потока

Это руководство читает поток по одной записи и пакетами, укладывается в лимит чтений каждого сегмента и показывает, что содержит доставка.

Пример - сервис аналитики. Он проводит платежи по мере поступления и считает просмотры страниц пакетами.

Как читается поток

Подписка запускает по читателю на каждый сегмент, который она держит. Читатель вызывает GetRecords с итератором сегмента, передаёт записи по порядку и сразу запрашивает снова. После чтения без записей он ждёт poll_interval, по умолчанию одну секунду, и только потом запрашивает снова. Между читателями и обработчиком ждут не больше 64 доставок. Читатель, который застал очередь полной, ждёт, поэтому медленный обработчик замедляет чтение.

Kinesis разрешает каждому сегменту пять вызовов GetRecords в секунду и 2 МБ в секунду. Этот лимит делят все потребители потока. Если сервис читает слишком часто, Kinesis ограничивает частоту его вызовов и отвечает ProvisionedThroughputExceededException. Сначала AWS SDK сам повторяет ограниченный вызов. Когда повторы исчерпаны, читатель ждёт один poll_interval и запрашивает снова с тем же итератором. Обработчик видит паузу, без ошибки и без повторной доставки.

По одной записи

Макросы

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "payments")]
struct Payment {
    id: u64,
    currency: String,
}

/// Books the payments in euros and drops the rest, which another service books.
#[subscriber(
    KinesisStream::new("payments")
        .poll_interval(Duration::from_millis(250))
        .create_if_missing(1)
)]
async fn book(payment: &Payment, Ctx(at): Ctx<Position>) -> HandlerOutcome {
    if payment.currency != "EUR" {
        return HandlerOutcome::drop();
    }
    tracing::info!(payment.id, position = ?at, "payment booked");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
struct Payment {
    id: u64,
    currency: String,
}

impl OutgoingDestination for Payment {
    type Form = FixedName;

    const DESTINATION: &'static str = "payments";
}

impl MessageHeaders for Payment {
    type Contract = NoHeaders;
}

/// Books the payments in euros and drops the rest, which another service books.
struct Book;

impl Handle<Payment, (), (), KinesisContext> for Book {
    fn handle(
        &self,
        payment: &Payment,
        _outs: &(),
        ctx: &mut Context<'_, KinesisContext>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if payment.currency != "EUR" {
            return ready(Err(HandlerOutcome::drop()));
        }
        let at = ctx.context(Position);
        tracing::info!(payment.id, position = ?at, "payment booked");
        ready(Ok(()))
    }
}

book читает поток payments и завершает каждый платёж отдельно. Подписка на одиночные записи запрашивает до 1000 записей за одно чтение. poll_interval(Duration::from_millis(250)) сокращает паузу после пустого чтения, поэтому новый платёж ждёт не больше четверти секунды. Четыре пустых чтения в секунду оставляют другим потребителям одно чтение в секунду из лимита сегмента. Эту настройку задаёт и шаг точки монтирования .poll_interval(..) после b.include(..).

Через Ctx<Position> обработчик получает место записи в потоке: её сегмент и порядковый номер в виде KinesisPosition. Перемотка на эту позицию читает сегмент заново с этой записи (Повторное чтение и перемотка). Каждая доставка содержит ещё три заголовка:

Заголовок	Значение
kinesis-shard-id (SHARD_HEADER)	сегмент, из которого прочитана запись
kinesis-sequence-number (SEQUENCE_HEADER)	её порядковый номер в этом сегменте
partition-key (PARTITION_KEY_HEADER)	ключ сегмента, с которым её опубликовали

Обработчик читает их через ctx.headers(). Остальные заголовки записи - те, что записал её производитель.

Пакеты

Макросы

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
#[outgoing(name = "page-views")]
struct PageView {
    page: String,
}

/// Counts page views a batch at a time and settles the whole batch with one outcome.
#[subscriber(KinesisStream::new("page-views").create_if_missing(1))]
async fn count(views: &[PageView]) -> HandlerOutcome {
    tracing::info!(views = views.len(), "page views counted");
    HandlerOutcome::ack()
}

Вручную

src/main.rs

#[derive(Debug, PartialEq, Deserialize, Serialize, JsonSchema)]
struct PageView {
    page: String,
}

impl OutgoingDestination for PageView {
    type Form = FixedName;

    const DESTINATION: &'static str = "page-views";
}

impl MessageHeaders for PageView {
    type Contract = NoHeaders;
}

/// Counts page views a batch at a time and settles the whole batch with one outcome.
struct Count;

impl Handle<[PageView]> for Count {
    fn handle(
        &self,
        views: &[PageView],
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), Vec<HandlerOutcome>>> {
        tracing::info!(views = views.len(), "page views counted");
        ready(Ok(()))
    }
}

Обработчик, который принимает срез, - пакетный обработчик. Приложение монтирует count с .batch(nonzero!(500)), и 500 становится пределом каждого вызова GetRecords его читателей. Поэтому одно чтение никогда не выбирает больше одного пакета, а Kinesis ограничивает чтение 10 000 записей. Неполный пакет уходит через 50 мс после своей первой записи, поэтому даже малоактивный поток быстро доставляет записи. Один пакет может смешивать записи всех сегментов, которые держит экземпляр.

Размер пакета - выбор между задержкой и лимитом чтений. Маленький пакет - это маленькое чтение, а каждое чтение расходует один из пяти вызовов сегмента в секунду. count завершает весь пакет одним исходом. Обработчик, который возвращает Vec<HandlerOutcome>, завершает каждую запись отдельно, и тогда каждая запись двигает контрольную точку своего сегмента (Контрольные точки и гарантии доставки).

Приложение

Макросы

src/main.rs

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("analytics", "0.1.0")).with_broker(broker, |b| {
        b.include(book);
        b.include(count.batch(nonzero!(500)));
    })
}

#[ruststream::app]
fn app() -> impl App {
    service(
        KinesisBroker::new()
            .endpoint("http://localhost:4566")
            .region("us-east-1")
            .test_credentials(),
    )
}

Вручную

src/main.rs

fn service(broker: KinesisBroker) -> RustStream {
    RustStream::new(AppInfo::new("analytics", "0.1.0")).with_broker(broker, |b| {
        b.include(
            subscriber(
                KinesisStream::new("payments")
                    .poll_interval(Duration::from_millis(250))
                    .create_if_missing(1),
                Book,
            )
            .build(),
        );
        b.include(
            subscriber(KinesisStream::new("page-views").create_if_missing(1), Count)
                .batch(nonzero!(500))
                .build(),
        );
    })
}

fn main() -> ExitCode {
    run_main(|| {
        service(
            KinesisBroker::new()
                .endpoint("http://localhost:4566")
                .region("us-east-1")
                .test_credentials(),
        )
    })
}

По умолчанию обработчик выполняет одну доставку за раз. workers(n) в атрибуте или шаг workers ручной формы выполняет до n доставок одновременно. Руководство Сегменты, ключи сегмента и порядок показывает, как при этом сохранить порядок каждого ключа.

Крейт читает с общей пропускной способностью потока, через GetRecords. Запись, которую агрегировала Kinesis Producer Library, не передаётся обработчику: подписка сообщает о ней ошибкой KinesisError::AggregatedRecord и читает дальше.

Тесты в файле примера запускают приложение внутри процесса. Тест пакетов начинается с остановленными часами, поэтому 50 мс, которые ждёт неполный пакет, проходят мгновенно.