wakuwaku
イベント (Iggy)

イベントの発行

Publisher で型付きイベントを Iggy に送り、書き込まれるヘッダーと、必要なストリーム・トピックの準備を理解します。

発行に必要なのは 1 つの型と 2 つのメソッドだけです。Publisher の、1 つのイベントを送る publish と、1 つの型の多数のイベントを送る publish_batch です。トピック、パーティション、ヘッダー、エンコードといったそれ以外のことは、イベントの定義 で定義したイベント型から決まります。

パブリッシャーを作って発行する

パブリッシャーは、接続済みの IggyClient と 1 つのストリーム名を包みます。そのストリームに対して、どのキー型をキーとする、どの型のイベントでも発行できます。

src/main.rs
use std::sync::Arc;
use wakuwaku::deps::iggy::prelude::{Client, IggyClient};
use wakuwaku::iggy::Publisher;

#[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?;

    // 注文 42 のイベントを 2 つ発行する。どちらもトピック `orders` の同じパーティションに入る。
    let publisher = Publisher::new(client.clone(), "shop")?;
    publisher
        .publish(OrderPlaced {
            order_id: 42,
            customer: "alice".into(),
            total_cents: 12_000,
        })
        .await?;
    publisher
        .publish(PaymentReceived {
            order_id: 42,
            amount_cents: 12_000,
        })
        .await?;

    Ok(())
}

Publisher::new はサーバーと通信しません。ストリーム名が Iggy の識別子として有効かどうかを確認するだけです。publish(event) は次の処理を行います。

  1. キー型の PartitionMerge ハッシャーを使い、key_hash(&event.key()) でキーハッシュを計算します。
  2. イベントをボディに変換し（IntoEventBody::into_event_body）、MessageSer::to_bytes でシリアライズします。
  3. ボディをペイロードとし、wakuwaku のヘッダー を付けた Iggy メッセージを 1 つ組み立てます。
  4. そのメッセージを、ストリーム内の、キー型の SUPER_PARTITION_NAME という名前のトピックへ、キーハッシュを Iggy のメッセージキーとしてパーティショニングして送信します。

したがって 1 つのキーのイベントはすべて同じパーティションに、発行した順で入ります。publish の 1 回の呼び出しで送るメッセージはちょうど 1 つです。多数のイベントを少ないリクエストで送るには publish_batch を使います。

多数のイベントをまとめて発行する

publish_batch(events) は、1 つのイベント型のイベントを任意の数だけ、キーが許す限り少ないリクエストで発行します。

src/settlement.rs
use wakuwaku::iggy::Publisher;
use wakuwaku::iggy::error::Error;

/// 決済ファイルの支払いを記録する。
pub async fn import_settlement(publisher: &Publisher) -> Result<(), Error> {
    publisher
        .publish_batch([
            PaymentReceived { order_id: 42, amount_cents: 5_000 },
            PaymentReceived { order_id: 43, amount_cents: 1_000 },
            PaymentReceived { order_id: 42, amount_cents: 7_000 },
        ])
        .await
}

イベントはキーハッシュでグループ化され、各グループが 1 つのリクエストで送られます。この例ではリクエストは 2 つです。まず注文 42 の 2 つの支払い（5 000、次に 7 000）、次に注文 43 の支払いです。規則は次のとおりです。

  • キーごとの順序は保たれます。 グループの中ではイベントは events での順序を保つので、キーのイベントはその順でパーティションに入ります。グループは、そのキーが最初に現れた順に 1 つずつ送られます。異なるキーのイベント同士はもともと順序づけられないので、それ以上の約束はありません。
  • シリアライズは全部か無かです。 何かを送る前に、すべてのイベントをシリアライズしてメッセージに組み立てます。1 つでも失敗すれば呼び出しはそのエラーを返し、何も発行されません。
  • 送信エラーはバッチを途中で止めます。 失敗したグループより前のグループは発行済みで、失敗したグループとそれ以降は発行されていません。エラーからはどこで止まったかはわかりません。
  • 何もリトライされません。 publish と同じです。

1 つの操作が 1 つの型のイベントを多数生む場合に使ってください。インポート、バックフィル、多数のキーに触れるコマンドなどです。イベントごとの往復が省け、キーのイベントは順序どおりに届きます。送信エラーの後で呼び出し全体をリトライすると、前のグループがもう一度発行されます。ハンドラが冪等なら（NoEventLose ではもともと必要です。配信モード を参照）これは無害です。異なる型のイベントは型ごとに呼び出しを分け、1 つのキーの中での順序が重要なら、1 つずつ await してから次を呼んでください。

パブリッシャーを共有する

Publisher は Clone で、クローンは安価です。クローン同士は同じ IggyClient を共有します。起動時に 1 回だけ作って ServiceBuilder に登録すれば、各コンポーネントがそれを依存として受け取れます。

src/checkout.rs
use wakuwaku::ServiceCreation;
use wakuwaku::iggy::Publisher;
use wakuwaku::iggy::error::Error;

/// イベントを発行するコンポーネント。依存は `Publisher`。
pub struct Checkout {
    publisher: Publisher,
}

impl ServiceCreation for Checkout {
    type Dep = Publisher;
    fn create_service(publisher: Publisher) -> Self {
        Self { publisher }
    }
}

impl Checkout {
    pub async fn place_order(&self, order_id: u64, customer: String) -> Result<(), Error> {
        // ... 注文を保存してから、それを通知する。
        self.publisher
            .publish(OrderPlaced {
                order_id,
                customer,
                total_cents: 12_000,
            })
            .await
    }
}
src/main.rs
let publisher = Publisher::new(client.clone(), "shop")?;
let services = ServiceBuilder::new(client).push(publisher);

ワイヤ上のヘッダー

すべてのメッセージには 3 つのユーザーヘッダーが付きます。値はすべて生のバイト列です。コンシューマはこれを読み戻し、ボディをデコードせずにレコードの振り分け、順序づけ、縮約を行います。この表が必要になるのは、別のプログラム（他の言語、ブリッジ、リプレイツールなど）が wakuwaku のコンシューマに読ませるイベントを書き込む場合だけです。定数は wakuwaku::iggy::events::headers にあります。

ヘッダー内容
wakuwaku-tagイベントの TYPE_TAG。u32 リトルエンディアン
wakuwaku-keyキーハッシュ。u64 リトルエンディアン
wakuwaku-props3 バイト。バージョン 1、メッセージの順序、型の結合性 (associativity)
wakuwaku-retryリトライトピックのレコードにのみ付く。バージョン 1 に続いて、失敗時刻（Unix エポックからのミリ秒、u64 リトルエンディアン）

wakuwaku-props の中のワイヤ上の値は次のとおりです。

順序バイト値代数マーカーバイト値
Relaxed1NonAssociative0
Acquire2Associative1
Release4Idempotent2
AcqRel6Commutative3
IdempotentCommutative4

外部のプロデューサーは、パーティショニングも同じ方法で行う必要があります。キーハッシュのリトルエンディアン 8 バイトを Iggy のメッセージキーとして使えば、Iggy クライアントが Publisher とまったく同じパーティションを選びます。

コンシューマは代数を二重に確認します

結合性のバイト値が、ハンドラのイベント型の代数と一致しないレコードは、ログに記録されたうえで破棄されます。誤ったルールで縮約されることはありません。

サーバーに必要なもの

wakuwaku が Iggy サーバー上に作成するものは最小限です。

リソースwakuwaku が作成するか
ストリーム（例: shop）作成しません。自分で作成してください。
メイントピック（SUPER_PARTITION_NAME、例: orders）とそのパーティション作成しません。必要なパーティション数で自分で作成してください。
リトライトピック retry_<topic>作成します。存在しなければ、コンシューマの start がメイントピックと同じパーティション数で作成します。

存在しないトピックへの publish は Iggy のエラーで失敗し、メイントピックがなければコンシューマは起動を拒否します（Error::TopicNotFound）。どちらもデプロイの一環として作成してください。たとえば、wakuwaku::deps::iggy として再エクスポートされている Iggy クライアントを使い、起動時に冪等なセットアップ処理を実行します。

src/setup.rs
use wakuwaku::deps::iggy::prelude::{
    Identifier, IggyClient, IggyError, StreamClient, TopicClient, TopicCreateOptions,
};
use wakuwaku::iggy::PartitionKey;

/// ストリームとメイントピックがなければ作成する。
pub async fn ensure_topology(client: &IggyClient) -> Result<(), IggyError> {
    let stream = Identifier::named("shop")?;
    if client.get_stream(&stream).await?.is_none() {
        client.create_stream("shop").await?;
    }

    let topic = OrderId::SUPER_PARTITION_NAME.0; // "orders"
    if client.get_topic(&stream, &Identifier::named(topic)?).await?.is_none() {
        // パーティションは 1 つで、`vec![0]` として購読する。「パーティション数は最初に決める」を参照。
        let options = TopicCreateOptions {
            partitions_count: Some(1),
            ..Default::default()
        };
        client.create_topic(&stream, topic, &options).await?;
    }
    Ok(())
}

トピック名を SUPER_PARTITION_NAME から読めば、セットアップとイベント型の食い違いを防げます。partitions_count を指定しない場合も Iggy SDK の既定値はパーティション 1 つですが、明示しておけば、コンシューマのパーティションのリストと並べて数を確認できます。

パーティション数は最初に決める

複数のパーティションによる並行性が必要でない限り、パーティションは 1 つにしてください。1 つのパーティションの中でも、コンシューマはバッチのキーを並行に処理します。キーのパーティションは、キーのハッシュとトピックのパーティション数から決まるので、後からパーティションを追加するとキーが別のパーティションへ移り、変更の前後に発行された 1 つのキーのイベントが順不同で処理されることがあります。また、リトライトピックのパーティション数がメイントピックと一致しなくなると、コンシューマは起動を拒否します（Error::PartitionCountMismatch）。追加したパーティションは、レプリカに 割り当てる まで購読されません。

エラー

Publisher::new、publish、publish_batch は wakuwaku::iggy::error::Error を返します。発行で起こりうるバリアントは次のとおりです。

バリアント原因
Error::Iggyストリーム名やトピック名が有効な識別子でない、メッセージを組み立てられない、または送信に失敗した（サーバーに到達できない、トピックがない、権限がない、など）
Error::Serializeボディの MessageSer::to_bytes が失敗した

送信はリトライされません。publish がエラーを返した場合、wakuwaku が知る限りイベントは書き込まれていません。リトライするか、リクエストを失敗させるか、エラーを呼び出し元に返すかはアプリケーションが決めます（publish_batch については 送信エラーが何を残すか を参照してください）。発行が行うのは Iggy への書き込みだけです。データベースのトランザクションのコミットと同時にイベントを発行する必要がある場合、その調整はアプリケーションのコードの責任です。

次のステップ

イベントの購読

発行したイベントを処理します。

順序

ヘッダーの順序バイトがコンシューマにとって何を意味するか。

イベントの定義

wakuwaku の Iggy イベント層で使うパーティションキー、イベント型、型タグ、ボディのシリアライズを宣言します。

イベントの購読

イベントハンドラを書き、ServiceBuilder または IggyConsumerRegisterCenter で登録し、パーティション、配信モード、シャットダウンを設定します。

目次

パブリッシャーを作って発行する多数のイベントをまとめて発行するパブリッシャーを共有するワイヤ上のヘッダーサーバーに必要なものエラー次のステップ