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

// Copyright 2024 tison <wander4096@gmail.com>
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use byteorder::ReadBytesExt;

use crate::codec::*;
use crate::IoResult;

// Version 1 adds validateOnly.
//
// Version 4 makes partitions/replicationFactor optional even when assignments are not present
// (KIP-464)
//
// Version 5 is the first flexible version.
// Version 5 also returns topic configs in the response (KIP-525).
//
// Version 6 is identical to version 5 but may return a THROTTLING_QUOTA_EXCEEDED error
// in the response if the topics creation is throttled (KIP-599).
//
// Version 7 is the same as version 6.

#[derive(Debug, Default, Clone)]
pub struct CreateTopicsRequest {
    /// The topics to create.
    pub topics: Vec<CreatableTopic>,
    /// 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,
    /// Unknown tagged fields.
    pub unknown_tagged_fields: Vec<RawTaggedField>,
}

impl Decodable for CreateTopicsRequest {
    fn read<B: ReadBytesExt>(buf: &mut B, version: i16) -> IoResult<Self> {
        let mut res = CreateTopicsRequest {
            topics: NullableArray(Struct(version), version >= 5)
                .decode(buf)?
                .ok_or_else(|| err_decode_message_null("name"))?,
            timeout_ms: Int32.decode(buf)?,
            ..Default::default()
        };
        if version >= 1 {
            res.validate_only = Bool.decode(buf)?;
        }
        if version >= 5 {
            res.unknown_tagged_fields = RawTaggedFieldList.decode(buf)?;
        }
        Ok(res)
    }
}

#[derive(Debug, Default, Clone)]
pub struct CreatableTopic {
    /// The topic name.
    pub name: String,
    /// 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<CreatableReplicaAssignment>,
    /// The custom topic configurations to set.
    pub configs: Vec<CreatableTopicConfig>,
    /// Unknown tagged fields.
    pub unknown_tagged_fields: Vec<RawTaggedField>,
}

impl Decodable for CreatableTopic {
    fn read<B: ReadBytesExt>(buf: &mut B, version: i16) -> IoResult<Self> {
        let mut res = CreatableTopic {
            name: NullableString(version >= 5)
                .decode(buf)?
                .ok_or_else(|| err_decode_message_null("name"))?,
            num_partitions: Int32.decode(buf)?,
            replication_factor: Int16.decode(buf)?,
            assignments: NullableArray(Struct(version), version >= 5)
                .decode(buf)?
                .ok_or_else(|| err_decode_message_null("assignments"))?,
            configs: NullableArray(Struct(version), version >= 5)
                .decode(buf)?
                .ok_or_else(|| err_decode_message_null("assignments"))?,
            ..Default::default()
        };
        if version >= 5 {
            res.unknown_tagged_fields = RawTaggedFieldList.decode(buf)?;
        }
        Ok(res)
    }
}

#[derive(Debug, Default, Clone)]
pub struct CreatableTopicConfig {
    /// The configuration name.
    pub name: String,
    /// The configuration value.
    pub value: Option<String>,
    /// Unknown tagged fields.
    pub unknown_tagged_fields: Vec<RawTaggedField>,
}

impl Decodable for CreatableTopicConfig {
    fn read<B: ReadBytesExt>(buf: &mut B, version: i16) -> IoResult<Self> {
        let mut res = CreatableTopicConfig {
            name: NullableString(version >= 5)
                .decode(buf)?
                .ok_or_else(|| err_decode_message_null("name"))?,
            value: NullableString(version >= 5).decode(buf)?,
            ..Default::default()
        };
        if version >= 5 {
            res.unknown_tagged_fields = RawTaggedFieldList.decode(buf)?;
        }
        Ok(res)
    }
}

#[derive(Debug, Default, Clone)]
pub struct CreatableReplicaAssignment {
    /// The partition index.
    pub partition_index: i32,
    /// The brokers to place the partition on.
    pub broker_ids: Vec<i32>,
    /// Unknown tagged fields.
    pub unknown_tagged_fields: Vec<RawTaggedField>,
}

impl Decodable for CreatableReplicaAssignment {
    fn read<B: ReadBytesExt>(buf: &mut B, version: i16) -> IoResult<Self> {
        let mut res = CreatableReplicaAssignment {
            partition_index: Int32.decode(buf)?,
            ..Default::default()
        };
        res.broker_ids = NullableArray(Int32, version >= 5)
            .decode(buf)?
            .ok_or_else(|| err_decode_message_null("broker_ids"))?;
        if version >= 5 {
            res.unknown_tagged_fields = RawTaggedFieldList.decode(buf)?;
        }
        Ok(res)
    }
}