samsa 0.1.8

Rust-native Kafka/Redpanda protocol and client implementation.
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
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95

mod testsupport;

use samsa::prelude::{
    self, protocol, BrokerConnection, ClusterMetadata, Error, KafkaCode, TcpConnection,
};
use std::collections::HashMap;

const CLIENT_ID: &str = "create delete topic integration test";
const CORRELATION_ID: i32 = 1;

#[tokio::test]
async fn it_can_create_and_delete_topics() -> Result<(), Box<Error>> {
    let (skip, brokers) = testsupport::get_brokers()?;
    if skip {
        return Ok(());
    }
    let mut metadata =
        ClusterMetadata::<TcpConnection>::new(brokers, 1, "rust".to_string(), vec![]).await?;

    let conn: &mut TcpConnection = metadata
        .broker_connections
        .get_mut(&metadata.controller_id)
        .unwrap();
    //
    // Create topic
    //
    let mut create_req =
        protocol::CreateTopicsRequest::new(CORRELATION_ID, CLIENT_ID, 4000, false)?;

    create_req.add("tester-creation", 2, 1);

    conn.send_request(&create_req).await?;
    let res = conn.receive_response().await?.freeze();

    let create_res = protocol::CreateTopicsResponse::try_from(res)?;

    assert_eq!(create_res.topics[0].error_code, KafkaCode::None);

    //
    // Delete topic
    //
    let mut delete_req = protocol::DeleteTopicsRequest::new(CORRELATION_ID, CLIENT_ID, 4000)?;

    delete_req.add("tester-creation");

    conn.send_request(&delete_req).await?;
    let res = conn.receive_response().await?.freeze();

    let delete_res = protocol::DeleteTopicsResponse::try_from(res)?;

    assert_eq!(delete_res.topics[0].error_code, KafkaCode::None);

    Ok(())
}

#[tokio::test]
async fn it_can_create_and_delete_topics_with_functions() -> Result<(), Box<Error>> {
    let (skip, brokers) = testsupport::get_brokers()?;
    if skip {
        return Ok(());
    }

    let mut metadata =
        ClusterMetadata::<TcpConnection>::new(brokers, 1, "rust".to_string(), vec![]).await?;

    let conn: &mut TcpConnection = metadata
        .broker_connections
        .get_mut(&metadata.controller_id)
        .unwrap();
    //
    // Create topic
    //
    let create_res = prelude::create_topics(
        conn.clone(),
        CORRELATION_ID,
        CLIENT_ID,
        HashMap::from([("function-topic", 2)]),
    )
    .await?;
    assert_eq!(create_res.topics[0].error_code, KafkaCode::None);

    //
    // Delete topic
    //
    let delete_res = prelude::delete_topics(
        conn.clone(),
        CORRELATION_ID,
        CLIENT_ID,
        vec!["function-topic"],
    )
    .await?;
    assert_eq!(delete_res.topics[0].error_code, KafkaCode::None);

    Ok(())
}