0.3 からの移行
wakuwaku 0.3 から 1.0.0-rc3 へサービスを移行する方法を、置き換えられた各モジュールの移行前後のコードとともに説明します。
wakuwaku 0.3 はユーティリティの集まりでした。RabbitMQ との連携、Redis と sqlx のヘルパー、リソースプール、そしてサービスビルダーです。1.0.0-rc3 はサービスビルダーを引き継ぎつつ、フレームワークになりました。データアクセス、イベント、バリデーションは別クレートとなってフィーチャーフラグの裏で再エクスポートされ、RabbitMQ は Apache Iggy に置き換わっています。
このページでは変更点をモジュールごとに説明します。各節では、0.3.1 向けに書いたコードと、それに相当する 1.0.0-rc3 のコードを示します。どちらも、それぞれのバージョンに対してコンパイルを確認しています。
まとめ
| 0.3 | 1.0.0-rc3 |
|---|---|
services::builder::ServiceBuilder、services::ServiceCreation | 変更なし。wakuwaku::ServiceBuilder、wakuwaku::ServiceCreation としても再エクスポート |
services::grpc_service、ServiceBuilder::grpc_service | 変更なし |
integration::amqp、services::amqp_consumer、ServiceBuilder::amqp_consumer | wakuwaku::iggy、ServiceBuilder::iggy_consumer |
interval_job::IntervalJobExecutionSignal | CronJob、ServiceBuilder::cron_job |
integration::redis | wakuwaku::entity::redis |
integration::sqlx | wakuwaku::entity::sqlx |
pool | 削除 |
integration::surreal | 削除 |
wakuwaku::Error | 削除。各クレートが独自のエラー型を持ちます |
フィーチャー tonic | フィーチャー grpc |
Cargo.toml を更新する
フィーチャーフラグは次のように変わりました。
| 0.3 のフィーチャー | 1.0.0-rc3 |
|---|---|
amqprs (デフォルト) | 削除。イベントには iggy を使います (ブローカーが異なります。後述)。 |
redis (デフォルト) | redis |
uuid (デフォルト) | 削除。uuid::Uuid を RedisKey に変換するためのものでした。代わりに ID の型に Name を実装します。 |
tracing (デフォルト) | 削除。wakuwaku は常に tracing でログを出力します。 |
tracing-otel | 削除。monotonic_counter.* のメトリクスイベントはなくなりました。 |
sqlx | sqlx-pg。新たに MySQL 用の sqlx-mysql があります。 |
surreal | 削除 |
tonic (暗黙のフィーチャー) | grpc。デフォルトで有効になりました |
| なし | 新規: proof、iggy、entity、meilisearch |
0.3 では kanau、sqlx、tonic に自分で依存していました。1.0.0-rc3 では、バージョンを wakuwaku と一致させるために wakuwaku::deps 経由でインポートします。例外は、クレートをパスで参照するコード生成と derive マクロです。インストール を参照してください。
[dependencies]
wakuwaku = { version = "0.3", features = ["sqlx", "tonic"] }
kanau = { version = "0.6", features = ["message"] }
sqlx = { version = "0.9", features = ["postgres", "runtime-tokio"] }
tonic = "0.14"サービスビルダーと gRPC
ServiceBuilder、ServiceCreation、位置を表す Here / There、ServiceBuilder::grpc_service、GrpcServiceRegisterCenter はこれまでどおり動きます。変わるのはインポートだけです。tonic は wakuwaku::deps::tonic から取り込み、中心となる 2 つの型はクレートのルートからも使えます。
use tonic::service::Routes;
use tonic::transport::Server;
use wakuwaku::services::ServiceCreation;
use wakuwaku::services::builder::ServiceBuilder;以前のパス wakuwaku::services::builder::ServiceBuilder と wakuwaku::services::ServiceCreation も引き続き使えます。
RabbitMQ のコンシューマーは Iggy のイベントハンドラーに
RabbitMQ のサポートはなくなりました。イベントは Apache Iggy を通り、型付きのイベント、キーごとの順序保証、バッチ処理が使えます。単純な置き換えではありません。RabbitMQ のキューに残っているメッセージは Iggy に移されないため、切り替える前にキューを空にする (または内容を Iggy に流し直す) 必要があります。
use kanau::processor::Processor;
use kanau::{JsonMessageDe, JsonMessageSer};
use serde::{Deserialize, Serialize};
use wakuwaku::integration::amqp::{
AmqpExchangeType, AmqpMessageProcessor, AmqpMessageSend, AmqpPool, AmqpRouting,
};
use wakuwaku::services::ServiceCreation;
use wakuwaku::services::builder::ServiceBuilder;
#[derive(Serialize, Deserialize, JsonMessageSer, JsonMessageDe)]
struct OrderPlaced {
order_id: u64,
}
impl AmqpRouting for OrderPlaced {
const EXCHANGE: &'static str = "orders";
const EXCHANGE_TYPE: AmqpExchangeType = AmqpExchangeType::Topic;
const ROUTING_KEY: &'static str = "order.placed";
}
impl AmqpMessageSend for OrderPlaced {}
#[derive(Clone)]
struct Mailer;
struct SendReceipt {
mailer: Mailer,
}
impl ServiceCreation for SendReceipt {
type Dep = Mailer;
fn create_service(mailer: Mailer) -> Self {
Self { mailer }
}
}
impl Processor<OrderPlaced> for SendReceipt {
type Output = ();
type Error = wakuwaku::Error;
async fn process(&self, event: OrderPlaced) -> Result<(), wakuwaku::Error> {
println!("receipt for order {}", event.order_id);
Ok(())
}
}
impl AmqpMessageProcessor<OrderPlaced> for SendReceipt {
const QUEUE: &'static str = "send-receipt";
}
async fn run(pool: &AmqpPool) -> Result<(), wakuwaku::Error> {
let (_services, consumers) = ServiceBuilder::new(Mailer)
.amqp_consumer::<SendReceipt, _, _>()
.setup(pool)
.await?;
OrderPlaced { order_id: 7 }.send(pool).await?;
consumers.closed().await;
Ok(())
}0.3 の各要素は次のように対応します。
| 0.3 | 1.0.0-rc3 |
|---|---|
AmqpRouting::EXCHANGE | Publisher::new と ConsumerConfig::new に渡す Iggy のストリーム |
AmqpRouting::ROUTING_KEY | Event::TYPE_TAG と、キー型の SUPER_PARTITION_NAME (トピック) |
AmqpMessageSend::send | Publisher::publish |
AmqpMessageProcessor<E> (Processor<E>) | イベントのスライス (最大 MAX_BATCH 件) を受け取る EventHandler<E> |
AmqpMessageProcessor::QUEUE | ConsumerConfig::new に渡すコンシューマー名 |
RETRY_POLICY、<queue>.retry | ConsumerConfig::retry_delays、retry_<topic> トピック |
FailureAction、<queue>.dlq | 返した HandleError の ErrorClass。デッドレター用のトピックはありません |
ServiceBuilder::amqp_consumer(..).setup(&pool) | ServiceBuilder::iggy_consumer(..).start(client, config) |
AmqpConsumersRuntime の drop | ConsumerRuntime::shutdown().await |
注意点は次のとおりです。
- ストリームとトピックを先に作成します。
startが作成するのはリトライ用トピックretry_<topic>だけです。キーのSUPER_PARTITION_NAMEを名前とするメイントピックと、設定したパーティションは事前に存在している必要があります。 - パーティション ID は 0 始まりです。 例ではパーティション
0を消費しています。 - リトライはキー単位です。
Retryableなエラーは、そのキーの未処理のイベントをリトライ用トピックへ移し、以降のイベントはその後ろで待ちます。最後のリトライも失敗すると、失敗した run は破棄されます。Unrecoverableなエラーはそのチャンクを破棄してログに記録します。0.3 ではこうしたメッセージはデッドレターキューに退避されていました。
イベントの定義、発行、消費、順序、エラー処理は イベント で説明しています。本体に使っている JSON 用の derive については イベントの定義 を参照してください。
インターバルジョブは cron ジョブに
0.3 の IntervalJobExecutionSignal は、シグナルメッセージを出すべきタイミングを表すものでした。time_pool を自分のループから呼び、シグナルを RabbitMQ に発行し、それをコンシューマーで処理する必要がありました。1.0.0-rc3 では、CronJob がプロセス内で、INTERVAL が宣言するスケジュールに従って処理そのものを実行します。
use std::task::Poll;
use kanau::{JsonMessageDe, JsonMessageSer};
use serde::{Deserialize, Serialize};
use time::{Duration, OffsetDateTime};
use wakuwaku::integration::amqp::{AmqpExchangeType, AmqpMessageSend, AmqpRouting};
use wakuwaku::interval_job::IntervalJobExecutionSignal;
#[derive(Serialize, Deserialize, JsonMessageSer, JsonMessageDe)]
struct ExpireSessionsSignal {
unix_time: i64,
}
impl AmqpRouting for ExpireSessionsSignal {
const EXCHANGE: &'static str = "jobs";
const EXCHANGE_TYPE: AmqpExchangeType = AmqpExchangeType::Direct;
const ROUTING_KEY: &'static str = "expire-sessions";
}
impl AmqpMessageSend for ExpireSessionsSignal {}
impl IntervalJobExecutionSignal for ExpireSessionsSignal {
fn tick(now: OffsetDateTime) -> Self {
Self { unix_time: now.unix_timestamp() }
}
fn time_pool(now: OffsetDateTime, last_time: OffsetDateTime) -> Poll<Self> {
if now - last_time >= Duration::minutes(1) {
Poll::Ready(Self::tick(now))
} else {
Poll::Pending
}
}
}シグナルのコンシューマーにあった処理は process に移します。ジョブを開始したレプリカはそれぞれジョブを実行するので、ジョブを冪等にするか、実行の中で Redis のロックを取得してください。cron ジョブ を参照してください。
Redis のキーバリューヘルパーは entity::redis に
KeyValue、KeyValueRead、KeyValueWrite の各トレイトは、Redis のデータソースと型付きのコレクションに置き換わりました。TTL 付きのキャッシュは Cache の定数になり、すべての操作は Db ハンドルを通して実行するクエリになります。
use std::time::Duration;
use kanau::{JsonMessageDe, JsonMessageSer};
use serde::{Deserialize, Serialize};
use wakuwaku::integration::redis::{KeyValue, KeyValueRead, KeyValueWrite, RedisConnection};
#[derive(Clone, Serialize, Deserialize, JsonMessageSer, JsonMessageDe)]
struct Profile {
name: String,
}
struct ProfileEntry {
user: u64,
profile: Profile,
}
impl KeyValue for ProfileEntry {
type Key = String;
type Value = Profile;
fn key(&self) -> String {
format!("profile:{}", self.user)
}
fn value(&self) -> Profile {
self.profile.clone()
}
fn into_value(self) -> Profile {
self.profile
}
fn new(key: String, profile: Profile) -> Self {
let user = key.trim_start_matches("profile:").parse().unwrap_or_default();
Self { user, profile }
}
}
impl KeyValueRead for ProfileEntry {}
impl KeyValueWrite for ProfileEntry {}
async fn run(conn: &mut RedisConnection) -> Result<(), wakuwaku::Error> {
let entry = ProfileEntry { user: 7, profile: Profile { name: "Haruki".into() } };
entry.write_with_ttl(conn, Duration::from_secs(300)).await?;
let profile: Option<Profile> = ProfileEntry::read(conn, "profile:7".into()).await?;
Ok(())
}RedisConnectionはDb<RedisSource, C>になります。ケイパビリティCが、実行できるクエリを決めます。書き込むコードにはReadWriteを、読み取りだけを行うべきコードにはReadOnlyを使います。- キーの構成が変わりました。コレクションは要素を
@@wakuwaku-entity:<kind>:<namespace>:<name>に保存するので、0.3 が書き込んだ値は見つかりません。キャッシュであれば、空の状態から始まるだけです。 - キャッシュでない値 (TTL がないもの、独自のキー構成を持つもの) は、
RedisSource向けの独自のクエリとして書きます。Redis を参照してください。 - エラーは
wakuwaku::Errorではなくredis::RedisErrorになります。
DatabaseProcessor は SqlxSource と Db に
0.3 では PgPool を DatabaseProcessor で包み、db.db() に対してどんな SQL でも実行できました。1.0.0-rc3 では、各クエリが読み書きする対象を宣言する値になり、その対象をカバーするケイパビリティを持つ Db を通して実行します。
use wakuwaku::integration::sqlx::DatabaseProcessor;
async fn count_users(db: &DatabaseProcessor) -> Result<i64, wakuwaku::Error> {
let (count,): (i64,) = sqlx::query_as("SELECT count(*) FROM users")
.fetch_one(db.db())
.await?;
Ok(count)
}- プールは
SqlxSourceの中に隠されているので、ケイパビリティを迂回できません。マイグレーションは、プールをラップする前にそのプールで実行します。 Db<PgSource, ReadOnly>は、書き込みと宣言されたクエリを実行できません。コンパイルエラーになります。- MySQL は
sqlx-mysqlとMySqlSourceでサポートされています。 - トランザクションはフローとして書きます。フローは状態の型で、そのステップが 1 つの開いたトランザクションの中でクエリを実行し、BEGIN・COMMIT・ROLLBACK はライブラリが受け持ちます。トランザクション を参照してください。
wakuwaku::Error の削除
0.3 にはすべてに使う 1 つのエラー列挙型があり、From<wakuwaku::Error> for tonic::Status の変換も用意されていました。1.0.0-rc3 では、それぞれが包んでいる層のエラー型を使います。sqlx のソースは sqlx::Error、Redis は RedisError、イベントハンドラーは HandleError、バリデーションは ValidateError です。クエリは自分の Error を Execute の実装で宣言するので、自ら拒否しうるクエリ (楽観ロックの喪失、ビジネスルール違反) はその拒否をエラーとして返します。サービス独自のエラー型と、その Status への変換を定義してください。
use tonic::Status;
fn find_user(id: u64) -> Result<String, wakuwaku::Error> {
Err(wakuwaku::Error::NotFound)
}
fn get_user(id: u64) -> Result<String, Status> {
// `wakuwaku::Error` converted into `tonic::Status` for you.
Ok(find_user(id)?)
}代替なしで削除されたもの
pool: 汎用のリソースプールです。wakuwaku が扱うクライアントは、それぞれプールや共有の仕組みを持っています。sqlx のプール、Redis のマルチプレックス接続 (クローンが 1 本のソケットを共有します)、そしてArc<IggyClient>です。integration::surreal: SurrealDB のサポートは廃止されました。