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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
//! Encoding and creation for Create Topics requests.
//!
//! ### Example
//! ```rust
//! let mut create_topics_request = protocol::CreateTopicsRequest::new(
//!     correlation_id,
//!     client_id,
//!     timeout_ms,
//!     validate_only,
//! );
//! coordinator_conn.send_request(&create_topics_request).await?;
//! ```
//!
//! ### Protocol Def
//! ```text
//! CreateTopics Request (Version: 3) => [topics] timeout_ms validate_only
//!  topics => name num_partitions replication_factor [assignments] [configs]
//!    name => STRING
//!    num_partitions => INT32
//!    replication_factor => INT16
//!    assignments => partition_index [broker_ids]
//!      partition_index => INT32
//!      broker_ids => INT32
//!    configs => name value
//!      name => STRING
//!      value => NULLABLE_STRING
//!  timeout_ms => INT32
//!  validate_only => BOOLEAN
//! ```
//!
//! Note that we are using version 3 of this API

use crate::{encode::ToByte, error::Result, protocol::HeaderRequest};

const API_KEY_METADATA: i16 = 19;
const API_VERSION: i16 = 3;

/// The base Create Topics request object.
///
/// ### Example
/// ```rust
/// let mut create_topics_request = protocol::CreateTopicsRequest::new(
///     correlation_id,
///     client_id,
///     timeout_ms,
///     validate_only,
/// );
/// coordinator_conn.send_request(&create_topics_request).await?;
/// ```
#[derive(Debug)]
pub struct CreateTopicsRequest<'a> {
    pub header: HeaderRequest<'a>,
    /// The topics to commit offsets for.
    pub topics: Vec<Topic<'a>>,
    /// How long to wait in milliseconds before timing out the request.
    pub timeout_ms: i32,
    /// If true, check that the topics can be created as specified, but don't create anything.
    pub validate_only: bool,
}

/// The topics to create.
#[derive(Debug)]
pub struct Topic<'a> {
    /// The topic name.
    pub name: &'a str,
    /// The number of partitions to create in the topic, or -1 if we are either specifying a manual partition assignment or using the default partitions.
    pub num_partitions: i32,
    /// The number of replicas to create for each partition in the topic, or -1 if we are either specifying a manual partition assignment or using the default replication factor.
    pub replication_factor: i16,
    /// The manual partition assignment, or the empty array if we are using automatic assignment.
    pub assignments: Vec<Assignment>,
    /// The custom topic configurations to set.
    pub configs: Vec<Config>,
}

/// The manual partition assignment, or the empty array if we are using automatic assignment.
#[derive(Debug)]
pub struct Assignment {
    /// The partition index.
    partition_index: i32,
    /// The brokers to place the partition on.
    broker_ids: Vec<i32>,
}

/// The custom topic configurations to set.
#[derive(Debug)]
pub struct Config {
    /// The configuration name.
    name: String,
    /// The configuration value.
    value: Option<String>,
}

impl<'a> CreateTopicsRequest<'a> {
    /// Create a new Create Topics Request
    ///
    /// This request needs to be given topics and partitions to be created
    /// before being sent to the broker. You can do this by using the `add` method.
    pub fn new(
        correlation_id: i32,
        client_id: &'a str,
        timeout_ms: i32,
        validate_only: bool,
    ) -> Result<Self> {
        let header = HeaderRequest::new(API_KEY_METADATA, API_VERSION, correlation_id, client_id);
        Ok(Self {
            header,
            timeout_ms,
            validate_only,
            topics: vec![],
        })
    }

    /// Add a topic to be create
    ///
    /// If the same topic is used twice, it will do nothing the second time
    ///
    /// ### Example
    /// ```rust
    /// create_topics_request.add(
    ///     topic_name,
    ///     num_partitions,
    ///     replication_factor,
    /// );
    /// ```
    pub fn add(&mut self, topic_name: &'a str, num_partitions: i32, replication_factor: i16) {
        match self
            .topics
            .iter_mut()
            .find(|topic| topic.name == topic_name)
        {
            None => self.topics.push(Topic {
                name: topic_name,
                num_partitions,
                replication_factor,
                assignments: vec![],
                configs: vec![],
            }),
            Some(_) => {
                // do nothing
            }
        }
    }
}

impl<'a> ToByte for CreateTopicsRequest<'a> {
    fn encode<T: bytes::BufMut>(&self, buffer: &mut T) -> crate::error::Result<()> {
        tracing::trace!("Encoding CreateTopicsRequest {:?}", self);
        self.header.encode(buffer)?;
        self.topics.encode(buffer)?;
        self.timeout_ms.encode(buffer)?;
        self.validate_only.encode(buffer)?;
        Ok(())
    }
}

impl<'a> ToByte for Topic<'a> {
    fn encode<T: bytes::BufMut>(&self, buffer: &mut T) -> crate::error::Result<()> {
        self.name.encode(buffer)?;
        self.num_partitions.encode(buffer)?;
        self.replication_factor.encode(buffer)?;
        self.assignments.encode(buffer)?;
        self.configs.encode(buffer)?;
        Ok(())
    }
}

impl ToByte for Assignment {
    fn encode<T: bytes::BufMut>(&self, buffer: &mut T) -> crate::error::Result<()> {
        self.partition_index.encode(buffer)?;
        self.broker_ids.encode(buffer)?;
        Ok(())
    }
}

impl ToByte for Config {
    fn encode<T: bytes::BufMut>(&self, buffer: &mut T) -> crate::error::Result<()> {
        self.name.encode(buffer)?;
        self.value.encode(buffer)?;
        Ok(())
    }
}