イベントの購読

コンシューマは、1 つの Iggy トピックを読むイベントハンドラの一覧です。イベント型ごとにハンドラを 1 つ書いて登録し、start を呼びます。wakuwaku は割り当てたパーティションごとに 1 つのタスクを起動し、概要 で説明したパイプラインを実行します。

イベントハンドラリンクをコピー

ハンドラは、イベント型 E に対して EventHandler<E> を実装します。

pub trait EventHandler<E: Event>: Send + Sync + 'static {
    const MAX_BATCH: NonZeroUsize;

    fn handle(&self, run: &[E::Target]) -> impl Future<Output = Result<(), HandleError>> + Send;
}

• run には1 つのキーのイベントが、適用すべき順に、E の代数で縮約済みの状態で入っています。要素の型はデコード済みのボディである E::Target です。イベント自身がボディである通常の場合、これは単に E です。
• MAX_BATCH は run の長さの上限です。これより長いランは、最大 MAX_BATCH 件ずつの連続したチャンクに分けて順に渡されます。1 回の呼び出しで 1 件ずつ受け取るには NonZeroUsize::MIN（1）を使います。
• 戻り値 は、HandleError の ErrorClass を通じて、失敗時に何が起きるかを決めます。後述 を参照してください。

handle は async fn として書けます。異なるキーのハンドラは Tokio 上で並行に動くため、その future は Send でなければなりません。std::sync::MutexGuard を .await をまたいで保持しないでください。

イベントの定義 のイベントに対する 2 つのハンドラは次のとおりです。どちらも、データベースのコネクションプールの代わりである OrderStore から組み立てます。

/// 新しい注文を書き込む。
pub struct RecordOrder {
    store: OrderStore,
}

impl ServiceCreation for RecordOrder {
    type Dep = OrderStore;
    fn create_service(store: OrderStore) -> Self {
        Self { store }
    }
}

impl EventHandler<OrderPlaced> for RecordOrder {
    const MAX_BATCH: NonZeroUsize = NonZeroUsize::new(100).unwrap();

    async fn handle(&self, run: &[OrderPlaced]) -> Result<(), HandleError> {
        for order in run {
            self.store.insert(order.order_id, order.total_cents).await;
        }
        Ok(())
    }
}

/// 支払いを適用する。ランは 1 つの支払いに畳み込まれた状態で届く。
pub struct ApplyPayment {
    store: OrderStore,
}

impl ServiceCreation for ApplyPayment {
    type Dep = OrderStore;
    fn create_service(store: OrderStore) -> Self {
        Self { store }
    }
}

impl EventHandler<PaymentReceived> for ApplyPayment {
    const MAX_BATCH: NonZeroUsize = NonZeroUsize::MIN;

    async fn handle(&self, run: &[PaymentReceived]) -> Result<(), HandleError> {
        for payment in run {
            self.store
                .pay(payment.order_id, payment.amount_cents)
                .await
                .map_err(HandleError::retryable)?;
        }
        Ok(())
    }
}

PaymentReceived は Associative なので、同じランに入った 1 つの注文の支払いは、合算された 1 つのイベントとして ApplyPayment に届きます。OrderPlaced は Acquire なので、注文の後に発行された支払いが注文より先に適用されることはありません。

ハンドラは普通の値なので、単体テストでは Iggy なしで handle を直接呼び出せます。

#[cfg(test)]
mod tests {
    use super::*;
    use wakuwaku::iggy::error::ErrorClass;

    #[tokio::test]
    async fn payment_for_unknown_order_is_retryable() {
        let handler = ApplyPayment::create_service(OrderStore::default());
        let payment = PaymentReceived { order_id: 7, amount_cents: 500 };
        let error = handler.handle(&[payment]).await.unwrap_err();
        assert_eq!(error.class(), ErrorClass::Retryable);
    }
}

サービスビルダーで登録するリンクをコピー

ハンドラはたいてい、データベースのプール、HTTP クライアント、Publisher などの共有依存を必要とします。上の例のようにハンドラに ServiceCreation を実装し、ServiceBuilder::iggy_consumer に、サービスビルダー に登録済みの依存から組み立てさせます（複数のサービスを必要とするハンドラは、代わりに 組み立てて登録 します）。

use std::sync::Arc;
use wakuwaku::ServiceBuilder;
use wakuwaku::deps::iggy::prelude::{Client, IggyClient};
use wakuwaku::iggy::ConsumerConfig;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let client = Arc::new(IggyClient::from_connection_string(
        "iggy://iggy:iggy@localhost:8090",
    )?);
    client.connect().await?;

    // `orders` トピックの 2 つのイベント型を 1 つのコンシューマで購読する。
    // トピックのパーティションは 1 つ。Iggy のパーティション ID は 0 始まり。
    let (_services, consumers) = ServiceBuilder::new(OrderStore::default())
        .iggy_consumer::<RecordOrder, _, _>()
        .iggy_consumer::<ApplyPayment, _, _>()
        .start(client, ConsumerConfig::new("shop", "order-projection", vec![0]))
        .await?;

    tokio::signal::ctrl_c().await?;
    consumers.shutdown().await;
    Ok(())
}

3 つの型パラメータは <Handler, Event, Position> です。

• Handler は組み立てるハンドラの型です。
• Event は、ハンドラがちょうど 1 つのイベントに対して EventHandler を実装しているなら _ にできます。1 つのハンドラ型が複数のイベントを扱う場合は名前を指定します: .iggy_consumer::<MyHandler, MyEvent, _>()。
• Position は、ハンドラの Dep がビルダー内のどこにあるかを表します。その型が 1 回だけ登録されていれば _ で推論できます。そうでなければ、他の provide 呼び出しと同様に Here / There で指定します。登録されていない依存はコンパイルエラーになります。

iggy_consumer を呼ぶたびに、その場でハンドラが組み立てられます。ビルダーから Dep をクローンして create_service を呼び、ハンドラを Arc で包みます。iggy_consumer と compose_iggy_consumer は任意の順序でチェーンでき、チェーンの最後は次のどちらかです。

• start(client, config): 購読を開始し、サービスビルダーと実行中の ConsumerRuntime を返します。続けてサービスの残りを組み立てられます。
• into_parts(): サービスビルダーと未起動の IggyConsumerRegisterCenter を返します。後で起動できます。

let (services, orders) = ServiceBuilder::new(OrderStore::default())
    .iggy_consumer::<RecordOrder, _, _>()
    .iggy_consumer::<ApplyPayment, _, _>()
    .into_parts();

// ... `services` からサービスの残りを組み立ててから:
let consumers = orders
    .start(client, ConsumerConfig::new("shop", "order-projection", vec![0]))
    .await?;

複数のサービスを必要とするハンドラリンクをコピー

iggy_consumer は、登録済みの依存 1 つ からハンドラを組み立てます。複数の依存を必要とするハンドラや、手で組み立てるハンドラは、代わりに compose_iggy_consumer で登録します。そのクロージャはサービスビルダーを借用し、必要なものを provide で取り出して、ハンドラを Arc で包んで返します。ハンドラに ServiceCreation の実装は不要です。また、サービスとしては登録されないので、他のコンポーネントの依存にはなれません。

たとえば、注文を記録するときに確認メールも送るとします。ハンドラには OrderStore に加えて Mailer（これも代役で、安価にクローンできるハンドル）が必要になります。

/// 新しい注文を書き込み、確認メールを送る。
pub struct RecordAndConfirmOrder {
    pub store: OrderStore,
    pub mailer: Mailer,
}

impl EventHandler<OrderPlaced> for RecordAndConfirmOrder {
    const MAX_BATCH: NonZeroUsize = NonZeroUsize::new(100).unwrap();

    async fn handle(&self, run: &[OrderPlaced]) -> Result<(), HandleError> {
        for order in run {
            self.store.insert(order.order_id, order.total_cents).await;
            self.mailer
                .confirm(&order.customer, order.order_id)
                .await
                .map_err(HandleError::retryable)?;
        }
        Ok(())
    }
}

let (_services, consumers) = ServiceBuilder::new(OrderStore::default())
    .push(Mailer::default())
    .compose_iggy_consumer(|s| {
        Arc::new(RecordAndConfirmOrder {
            store: s.provide::<OrderStore, _>().clone(),
            mailer: s.provide::<Mailer, _>().clone(),
        })
    })
    .iggy_consumer::<ApplyPayment, _, _>()
    .start(client, ConsumerConfig::new("shop", "order-projection", vec![0]))
    .await?;

ハンドラとイベントの型は、返された Arc から推論されます。ハンドラが複数のイベントに対して EventHandler を実装している場合は、イベントを指定します: .compose_iggy_consumer::<_, MyEvent>(|s| …)。ビルダーより前に組み立てたハンドラや、他のコードと共有するハンドラは、そのまま move で渡せます: .compose_iggy_consumer(move |_| handler)。サービスビルダーの他の compose_* メソッドも同じ仕組みです。クロージャでコンポーネントを組み立てる を参照してください。

直接登録するリンクをコピー

ServiceBuilder::iggy_consumer は、ハンドラの型レベルリストである IggyConsumerRegisterCenter を便利に使うためのものです。サービスに ServiceBuilder がない場合などは、これを直接使えます。

pub async fn start_by_hand(
    client: Arc<IggyClient>,
    store: OrderStore,
) -> Result<ConsumerRuntime, Error> {
    IggyConsumerRegisterCenter::new(Arc::new(RecordOrder::create_service(store.clone())))
        .push(Arc::new(ApplyPayment::create_service(store)))
        .start(client, ConsumerConfig::new("shop", "order-projection", vec![0]))
        .await
}

new で最初のハンドラからリストを始め、push のたびに 1 つ追加します。ハンドラは Arc で渡すので、1 つのハンドラの値で複数のイベント型を処理できます。ハンドラ型が複数のイベントに対して EventHandler を実装している場合は、イベントを指定します。ここでの Audit は EventHandler<OrderPlaced> と EventHandler<PaymentReceived> の両方を実装しています。

let audit = Arc::new(Audit);
let handlers = IggyConsumerRegisterCenter::<_, OrderPlaced>::new(audit.clone())
    .push::<_, PaymentReceived>(audit);

1 つのコンシューマにメイントピックは 1 つリンクをコピー

コンシューマが読むメイントピックは、イベントのキー型の SUPER_PARTITION_NAME ただ 1 つです。したがって 1 つのリストに入るハンドラはすべて、同じトピック名を持つキー型のイベントを扱う必要があります。start はこれを他のセットアップとあわせて検査し、問題があれば購読を始めずにエラーを返します。

別のキー型のイベントを購読するには、2 つ目のコンシューマを起動します。登録したどのハンドラにも一致しないタグのレコードは、ログに記録されたうえで破棄され、オフセットはその先へ進みます。トピックのイベント型の一部だけを登録したコンシューマは、残りを単に読み飛ばします。意図的なら問題ありませんが、うっかりなら静かなデータ欠落です。トピックにイベント型を追加したら、そのハンドラもトピックのコンシューマに登録してください。

設定リンクをコピー

ConsumerConfig::new(stream, consumer_name, partitions) は必須の 3 つのフィールドを設定し、残りを既定値にします。フィールドはすべて公開されています。

フィールドは構造体更新構文で上書きします。retry_delays は &'static のスライスなので、独自の間隔は const として宣言します。

use std::time::Duration;
use wakuwaku::iggy::{ConsumerConfig, DeliveryMode};

const RETRY_DELAYS: &[Duration] = &[
    Duration::from_secs(1),
    Duration::from_secs(5),
    Duration::from_secs(30),
];

pub fn config(partitions: Vec<u32>) -> ConsumerConfig {
    ConsumerConfig {
        mode: DeliveryMode::NoEventLose,
        retry_delays: RETRY_DELAYS,
        poll_count: 500,
        poll_interval: Duration::from_millis(250),
        ..ConsumerConfig::new("shop", "order-projection", partitions)
    }
}

パーティションとタスクリンクをコピー

start は partitions の要素ごとに Tokio タスクを 1 つ起動します。パーティション同士は並列に処理されます。1 つのパーティションの中では、ポーリングした 1 バッチのキーが並行に処理され、現在のバッチのすべてのキーが完了するか、リトライトピックへ移されるか、破棄されるまで、次のバッチはポーリングされません。

partitions は設計上、明示的な割り当てです。コンシューマは列挙したパーティションだけを読み、それ以外は読みません。「すべてのパーティション」を読むモードは意図的に用意していません。リストをトピックから読み取るのも避けてください。トピックのパーティション数に黙って追従するコンシューマは、レプリカ間で安全に分割できません。また、後からトピックに追加されたパーティションを、誰も割り当てていないのに読み始めてしまいます。

Iggy のパーティション ID は 0 始まりです。wakuwaku が対応する Iggy サーバーのバージョンでは、パーティションが n 個のトピックの ID は 0..n です。パブリッシャーはキーのハッシュでイベントをトピックのすべてのパーティションへ振り分けるので、どのコンシューマも列挙していないパーティションはまったく購読されません。

小規模なデプロイではパーティションを 1 つにリンクをコピー

小規模なデプロイでは、トピックを 1 つ のパーティションで作成し（サーバーに必要なもの を参照）、このページの例のように vec![0] を購読します。1 つのパーティションのタスクでも、各バッチのキーは並行に処理されるので、たいていのサービスにはこれで足ります。

レプリカで分割するリンクをコピー

1 つのパーティションで処理が追いつかなくなったら、より多くのパーティションでトピックを作成するか、トピックのパーティションを増やし（キーが別のパーティションへ移ります。発行 を参照）、パーティションをレプリカへ明示的に割り当てます。各レプリカは自分の割り当てを、環境変数などの自身の設定から読み取り、config に渡します。

use std::env;
use std::error::Error;

/// このレプリカに割り当てられたパーティション。例: `ORDER_PARTITIONS=0,2`。
pub fn assigned_partitions() -> Result<Vec<u32>, Box<dyn Error>> {
    env::var("ORDER_PARTITIONS")?
        .split(',')
        .map(|id| Ok(id.trim().parse::<u32>()?))
        .collect()
}

パーティションが 4 つ、レプリカが 2 つの場合、n 個中 i 番目のレプリカが id % n == i の ID を担当すると、次のようになります。

1 つのパーティションのパイプラインリンクをコピー

各パーティションのタスクは次の処理を繰り返します。

1. リトライパーティションを読み切る。 retry_<topic> のパーティション p を最後まで読み、レコードをキーごとにキューに入れます。
2. 期限の来たリトライを実行する。 隔離中のキーのうち、次のリトライ間隔が過ぎたものを計画し直して再実行します。
3. ポーリングする。 メイントピックのパーティション p から、保存済みオフセットの次から最大 poll_count 件を取得します。1 件もなければ poll_interval だけスリープして最初に戻ります。
4. 計画する。 ヘッダーを解析し、レコードをキーハッシュでグループ化し、各キーを順序フェンスでセグメントに分け、セグメントを型ごとのランにまとめます。順序 を参照してください。
5. 隔離中のキーを迂回させる。 リトライトピックにまだレコードが残っているキーは実行しません。そのキーの新しいレコードは、古いレコードの後ろとしてリトライトピックに追記されるので、順序が保たれます。
6. 実行する。 それ以外のキーを並行に実行します。各ランについて、ボディをデコードし、代数 で縮約し、チャンクごとにハンドラを呼び出します。
7. 失敗をリトライへ書き込む。 リトライ可能なエラーで失敗したキーは、失敗した位置からリトライトピックに書き込まれ、隔離されます。
8. コミットする。 配信モードの定めに従ってメイントピックのオフセットを保存します。

Iggy 自体の失敗（ポーリング、オフセットの保存、リトライトピックへの書き込み）でタスクが止まることはありません。各呼び出しは成功するまで retry_delays の間隔で再試行され、そのたびに tracing で警告がログに記録されます。

配信モードリンクをコピー

DeliveryMode はオフセットを保存するタイミング、つまりクラッシュ時に何を失うかを決めます。

NoEventLose では、ハンドラを冪等に書いてください。同じイベントを 2 回適用しても害がないようにする必要があります。また、あるランのいくつかのチャンクが成功した後にリトライ試行が失敗すると、次の試行でそれらのチャンクがもう一度適用されます。NoDoubleConsume は、ベストエフォートの通知のように、イベントを失うほうが繰り返すより安くつく場合にだけ選んでください。

ハンドラが失敗したときリンクをコピー

3 つのコンストラクタのいずれかで HandleError を作って返します。そのクラスによって何が起きるかが決まります。

anyhow::Error に変換できるエラーなら何でも包めます。これには std::error::Error + Send + Sync + 'static を満たすすべての型が含まれます。リトライのスケジュール、隔離、リトライトピックについては エラー処理 で説明します。

シャットダウンリンクをコピー

ConsumerRuntime の止め方は 3 通りあります。

止めたときに何を失うかは 配信モード によって決まります。

shutdown は処理中の作業にかかるだけ待ちます。Transient を返し続けるハンドラは成功するまで再試行され、Iggy の呼び出し（オフセットの保存、リトライトピックへの書き込み）もサーバーに到達できない間は再試行されます。待ち時間は、プラットフォームの猶予期間に合ったタイムアウトで制限してください（たとえば Kubernetes は、既定では停止を求めてから 30 秒後に Pod を強制終了します）。

use std::time::Duration;

tokio::signal::ctrl_c().await?;
if tokio::time::timeout(Duration::from_secs(20), consumers.shutdown())
    .await
    .is_err()
{
    eprintln!("consumer shutdown timed out; the remaining work was aborted");
}

タイムアウトすると、終わっていない shutdown の future が、それが所有するランタイムごとドロップされ、まだ動いているタスクは中断されます。終わらなかった作業は、中断 (abort) した場合と同じ扱いになります。abort を直接使うのは、処理中の作業をまったく待てない場合だけにしてください。

次のステップリンクをコピー