kafrust 0.2.6

A pure Rust Kafka client with no librdkafka or C client binding dependency.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

mod common;

use kafrust::{
    AclBinding, AclFilter, AclOperation, AclPatternType, AclPermissionType, AclResourceType,
    AdminClient, ClientConfig, Error,
};

#[tokio::main]
async fn main() -> kafrust::Result<()> {
    let bootstrap_servers = common::bootstrap_servers_from_env();
    let topic =
        std::env::var("KAFRUST_ACL_TOPIC").unwrap_or_else(|_| "kafrust-acl-smoke".to_owned());
    let principal =
        std::env::var("KAFRUST_ACL_PRINCIPAL").unwrap_or_else(|_| "User:ANONYMOUS".to_owned());
    let config = common::apply_security(
        ClientConfig::new(bootstrap_servers).client_id("kafrust-acl-example"),
    )?;
    let admin = AdminClient::new(config);
    let binding = AclBinding::new(
        AclResourceType::Topic,
        &topic,
        AclPatternType::Literal,
        &principal,
        "*",
        AclOperation::Read,
        AclPermissionType::Allow,
    );

    let created = admin.create_acls(std::slice::from_ref(&binding)).await?;
    if !created.is_success() {
        return Err(Error::Broker {
            code: created
                .results()
                .first()
                .map(|result| result.error_code())
                .unwrap_or(-1),
            context: format!("create ACL for topic {topic}"),
        });
    }
    println!("created ACL for {principal} on topic {topic}");

    let filter = AclFilter::any()
        .resource_type(AclResourceType::Topic)
        .resource_name(&topic)
        .pattern_type(AclPatternType::Literal)
        .principal(&principal)
        .host("*")
        .operation(AclOperation::Read)
        .permission_type(AclPermissionType::Allow);
    let described = admin.describe_acls(&filter).await?;
    if !described.is_success()
        || !described
            .bindings()
            .iter()
            .any(|candidate| candidate == &binding)
    {
        return Err(Error::Broker {
            code: described.error_code(),
            context: format!("describe ACL for topic {topic}"),
        });
    }
    println!("described ACL for {principal} on topic {topic}");

    let deleted = admin.delete_acls(&[filter]).await?;
    if !deleted.is_success() {
        return Err(Error::Broker {
            code: deleted
                .filter_results()
                .first()
                .map(|result| result.error_code())
                .unwrap_or(-1),
            context: format!("delete ACL for topic {topic}"),
        });
    }
    println!("deleted ACL for {principal} on topic {topic}");

    Ok(())
}