taba-client (Rust)

The Taba Rust connector. Documentation: Connectors, Rust.

Other languages: Português (Brasil) · all languages

• Its own HTTP/1.1 over std::net::TcpStream (building the request and reading the response are pure functions in http), TLS 1.2 or later through rustls with the system roots (plus an optional CA of your own) and its own JSON and base64. The only dependencies are rustls and rustls-native-certs (plus serde, only in the optional serde feature).
• Thread-safe: Client is Send + Sync; a single one serves many.
• Call retries of the common contract: 429 and 503 (with Retry-After), 502 and 504 only on FIFO, a network failure before sending; never a 4xx nor a timeout.

use taba_client::{Client, Config, ConsumeOptions, PublishOptions, TopicRequest};

fn main() -> Result<(), taba_client::Error> {
    let mut config = Config::new("http://127.0.0.1:8080");
    config.token = std::env::var("TABA_TOKEN").ok();
    let client = Client::new(config)?;

    client.create_topic(&TopicRequest {
        name: "orders".into(),
        topic_type: "queue".into(),
        ..TopicRequest::default()
    })?;
    client.publish("orders", br#"{"order": 9001}"#, &PublishOptions::default())?;

    let options = ConsumeOptions { wait_time: Some(20), ..ConsumeOptions::default() };
    for message in client.consume("orders", &options)? {
        println!("{}", String::from_utf8_lossy(&message.body));
        client.ack("orders", &message.id, &message.receipt_token)?;
    }
    Ok(())
}

Consuming with a handler and objects (D140)

The simplest use: publish and consume objects, without serializing or deleting anything by hand. With the serde feature (taba-client = { features = ["serde"] }; without it the connector stays with rustls only):

#[derive(serde::Serialize, serde::Deserialize)]
struct Order { number: u32 }

client.publish_object("orders", &Order { number: 9001 })?;

client.consume_typed::<Order, _>("orders", &ConsumeOptions::default(), |d| {
    process_order(&d.body)?;              // d.body: Order already deserialized
    Ok(())
})?;

consume_typed reads with long polling, hands over the body already deserialized, deletes the message if the handler returns Ok and renews the invisibility while it runs. The details and customizations follow.

If the body is not a T, consume_typed returns client.invalid-response before the handler and the message is not deleted. Run the tests with the feature: cargo test --workspace --all-features.

Without serde: consume_with

Besides consume, which returns the list, consume_with reads from the queue with long polling (20 s), one message at a time and 120 s of invisibility (this mode's defaults; pass a ConsumeOptions to change them), hands each message to a Delivery and deletes the message if the handler returns Ok (delivery.delete is true by default). While the handler runs, a second thread renews the invisibility every third of it, so no other consumer takes the message. If the handler returns Err, the message is not deleted and the error propagates as ConsumeError::Handler.

Minimal:

loop {
    client.consume_with("orders", &ConsumeOptions::default(), |d| {
        println!("{}", String::from_utf8_lossy(&d.message.body));
        Ok(())
    })?;
}

Complete, annotated:

let stop = std::sync::atomic::AtomicBool::new(false); // elsewhere (a signal) sets true to end
while !stop.load(std::sync::atomic::Ordering::Relaxed) {
    // Each round is a long polling: with no messages it returns 0 after `wait_time`; with messages, it calls the handler.
    let handled = client.consume_with(
        "orders",
        &ConsumeOptions {
            max_messages: Some(1),        // default 1
            wait_time: Some(20),          // default 20 s (long polling)
            visibility_timeout: Some(120), // default 120 s; the renewal repeats it every 1/3
            ..ConsumeOptions::default()
        },
        |d| {
            // d.message carries everything: id, delivery_count, published_at, group_id, receipt_token, body (bytes)...
            process(&d.message.body)?;     // the `?` returns Err: the message is NOT deleted and goes back to the queue
            // Need more time in one go? This changes the invisibility and the base of the renewals:
            // d.change_visibility(600)?;
            d.delete = true;               // default true; false leaves the message in the queue
            if d.lease_lost() { /* the server refused a renewal: the message already went back to the queue */ }
            Ok(())
        },
    )?;                                    // returns how many messages were handled
}

Client::call and Call are the lower layer (route, query, body): it is what taba-cli uses to print the responses as the server returned them. taba_client::testing is a fake transport for the tests of whoever uses the connector.

Still missing: the async form (async feature, tokio) and a Transport the application can swap.

Declaring a topic

create_or_update_topic(&TopicRequest, Recreate) (PUT /topics/{topic}) creates the topic or brings it to the declared state, and returns a TopicPut with the topic and the outcome (TopicOutcome::Created, Updated or Recreated). What is left out goes back to the server default (a full replacement, unlike the update); another type, fifo or volatile is the error 409 topic.immutable-mismatch unless recreate is if-empty (only when the queue and its dead-letter are empty, 409 topic.not-empty otherwise) or always (drops the messages; new id). After a 503 the call is repeated like the others. delete_topic_if_empty(name) only removes a queue with no message. create_topic does not change (409 when it exists). Details: Create or update a topic.

Batches

Many messages in one request, next to the single calls (optional): publish_batch(topic, &[BatchMessage]), ack_batch(topic, &messages) and change_visibility_batch(topic, &messages, seconds) return one BatchResult per item (is_ok(), status, id, error_code). Each item is answered as its single call, in order, and an item that failed comes back with its status and error code instead of throwing. The bodies go in base64, so any type fits. Details: Messages in batch.

License

Apache-2.0. The Taba server is AGPL-3.0-only; the connectors are Apache-2.0 so any application, open or closed, can embed them.