Migrating from 0.3

wakuwaku 0.3 was a set of utilities: a RabbitMQ integration, Redis and sqlx helpers, a resource pool, and a service builder. 1.0.0-rc3 keeps the service builder, but it is now a framework. Data access, events, and validation are separate crates, re-exported behind feature flags, and RabbitMQ is replaced by Apache Iggy.

This page goes through the changes one module at a time. Each section shows code written against 0.3.1 and the equivalent for 1.0.0-rc3. Both sides were compiled against their respective versions.

SummaryCopy Anchor Link

Update Cargo.tomlCopy Anchor Link

The feature flags changed:

In 0.3 you depended on kanau, sqlx, and tonic yourself. In 1.0.0-rc3, import them through wakuwaku::deps so their versions match wakuwaku's. The exceptions are code generators and derive macros that name a crate by path; see Installation.

[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"

Service builder and gRPCCopy Anchor Link

ServiceBuilder, ServiceCreation, the Here / There positions, ServiceBuilder::grpc_service, and GrpcServiceRegisterCenter work as before. Only the imports change: tonic comes from wakuwaku::deps::tonic, and the two core types are also available at the crate root.

use tonic::service::Routes;
use tonic::transport::Server;
use wakuwaku::services::ServiceCreation;
use wakuwaku::services::builder::ServiceBuilder;

The old paths wakuwaku::services::builder::ServiceBuilder and wakuwaku::services::ServiceCreation still work.

RabbitMQ consumers become Iggy event handlersCopy Anchor Link

RabbitMQ support is gone. Events now go through Apache Iggy, with typed events, per-key ordering, and batching. This is not a drop-in replacement. Messages still queued in RabbitMQ are not moved to Iggy, so drain the queues (or replay their contents into Iggy) before you switch.

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(())
}

How the 0.3 pieces map:

Things to watch for:

• Create the stream and topics first. start creates only the retry topic retry_<topic>. The main topic, named after the key's SUPER_PARTITION_NAME, and the configured partitions must already exist.
• Partition ids are zero-based. The example consumes partition 0.
• Retries are per key. A Retryable error moves the key's unfinished events to the retry topic, and its later events wait behind them. If the last retry fails, the failed runs are dropped. An Unrecoverable error drops the chunk and logs it. 0.3 parked such messages in a dead-letter queue instead.

Events covers defining events, publishing, consuming, ordering, and error handling. Defining events explains the JSON derives used for the body.

Interval jobs become cron jobsCopy Anchor Link

In 0.3, IntervalJobExecutionSignal described when a signal message was due. You had to call time_pool from a loop of your own, publish the signal to RabbitMQ, and handle it in a consumer. In 1.0.0-rc3, a CronJob runs the work itself, inside the process, on the schedule its INTERVAL declares:

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
        }
    }
}

The work that used to live in the signal's consumer moves into process. Every replica that starts the job runs it, so make the job idempotent or take a Redis lock inside the run. See Cron jobs.

Redis key-value helpers become entity::redisCopy Anchor Link

The KeyValue, KeyValueRead, and KeyValueWrite traits are replaced by a Redis data source and typed collections. A cache with a TTL becomes a Cache constant, and every operation is a query that runs through a Db handle:

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 becomes Db<RedisSource, C>. The capability C decides which queries may run. Use ReadWrite for code that writes and ReadOnly for code that should only read.
• The key layout changed. Collections store their items under @@wakuwaku-entity:<kind>:<namespace>:<name>, so values written by 0.3 are not found. For caches this only means a cold start.
• Values that are not caches (no TTL, or your own key layout) are written as your own queries on RedisSource. See Redis.
• Errors are redis::RedisError instead of wakuwaku::Error.

DatabaseProcessor becomes SqlxSource and DbCopy Anchor Link

0.3 wrapped a PgPool in DatabaseProcessor, and any SQL could run on db.db(). In 1.0.0-rc3 every query is a value that declares what it reads and writes, and it runs through a Db whose capability must cover that:

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)
}

• The pool stays private inside SqlxSource, so the capability cannot be bypassed. Run migrations on the pool before wrapping it.
• A Db<PgSource, ReadOnly> cannot run a query declared as a write. That is a compile error.
• MySQL is supported through sqlx-mysql and MySqlSource.
• A transaction is a flow: a state type whose steps run queries inside one open transaction, while the library owns BEGIN, COMMIT, and ROLLBACK. See Transactions.

Queries and sqlx cover this in detail.

wakuwaku::Error is removedCopy Anchor Link

0.3 had one error enum for everything, including a From<wakuwaku::Error> for tonic::Status conversion. In 1.0.0-rc3, each part uses the error type of the layer it wraps: sqlx::Error for sqlx sources, RedisError for Redis, HandleError for event handlers, and ValidateError for validation. A query declares its own Error on its Execute impl, so a query that can refuse (a lost optimistic lock, a broken business rule) returns that refusal as its error. Define your service's own error type and its mapping to 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)?)
}

Removed without replacementCopy Anchor Link

• pool: the generic resource pool. The clients wakuwaku works with bring their own pooling or sharing: sqlx pools, Redis multiplexed connections (clones share one socket), and an Arc<IggyClient>.
• integration::surreal: SurrealDB support was dropped.

Next stepsCopy Anchor Link