Skip to content

Commit 4eec9be

Browse files
liunaijieliunaijie
authored
[Improve] rabbit mq options (#8740)
1 parent b83076c commit 4eec9be

File tree

9 files changed

+371
-384
lines changed

9 files changed

+371
-384
lines changed

Diff for: ‎seatunnel-ci-tools/src/test/java/org/apache/seatunnel/api/ConnectorOptionCheckTest.java

-2
Original file line numberDiff line numberDiff line change
@@ -171,7 +171,6 @@ private Set<String> buildWhiteList() {
171171
Set<String> whiteList = new HashSet<>();
172172
whiteList.add("JdbcSinkOptions");
173173
whiteList.add("TypesenseSourceOptions");
174-
whiteList.add("RabbitmqSourceOptions");
175174
whiteList.add("TypesenseSinkOptions");
176175
whiteList.add("EmailSinkOptions");
177176
whiteList.add("HudiSinkOptions");
@@ -188,7 +187,6 @@ private Set<String> buildWhiteList() {
188187
whiteList.add("MongodbSinkOptions");
189188
whiteList.add("IoTDBSinkOptions");
190189
whiteList.add("EasysearchSourceOptions");
191-
whiteList.add("RabbitmqSinkOptions");
192190
whiteList.add("IcebergSourceOptions");
193191
whiteList.add("HbaseSourceOptions");
194192
whiteList.add("PaimonSourceOptions");

Diff for: ‎seatunnel-connectors-v2/connector-rabbitmq/src/main/java/org/apache/seatunnel/connectors/seatunnel/rabbitmq/config/RabbitmqBaseOptions.java

+135
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,135 @@
1+
/*
2+
* Licensed to the Apache Software Foundation (ASF) under one or more
3+
* contributor license agreements. See the NOTICE file distributed with
4+
* this work for additional information regarding copyright ownership.
5+
* The ASF licenses this file to You under the Apache License, Version 2.0
6+
* (the "License"); you may not use this file except in compliance with
7+
* the License. You may obtain a copy of the License at
8+
*
9+
* http://www.apache.org/licenses/LICENSE-2.0
10+
*
11+
* Unless required by applicable law or agreed to in writing, software
12+
* distributed under the License is distributed on an "AS IS" BASIS,
13+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+
* See the License for the specific language governing permissions and
15+
* limitations under the License.
16+
*/
17+
18+
package org.apache.seatunnel.connectors.seatunnel.rabbitmq.config;
19+
20+
import org.apache.seatunnel.api.configuration.Option;
21+
import org.apache.seatunnel.api.configuration.Options;
22+
import org.apache.seatunnel.api.options.ConnectorCommonOptions;
23+
24+
public class RabbitmqBaseOptions extends ConnectorCommonOptions {
25+
26+
public static final Option<String> HOST =
27+
Options.key("host")
28+
.stringType()
29+
.noDefaultValue()
30+
.withDescription("the default host to use for connections");
31+
32+
public static final Option<Integer> PORT =
33+
Options.key("port")
34+
.intType()
35+
.noDefaultValue()
36+
.withDescription("the default port to use for connections");
37+
38+
public static final Option<String> VIRTUAL_HOST =
39+
Options.key("virtual_host")
40+
.stringType()
41+
.noDefaultValue()
42+
.withDescription("the virtual host to use when connecting to the broker");
43+
44+
public static final Option<String> QUEUE_NAME =
45+
Options.key("queue_name")
46+
.stringType()
47+
.noDefaultValue()
48+
.withDescription("the queue to write the message to");
49+
50+
public static final Option<String> USERNAME =
51+
Options.key("username")
52+
.stringType()
53+
.noDefaultValue()
54+
.withDescription("the AMQP user name to use when connecting to the broker");
55+
56+
public static final Option<String> PASSWORD =
57+
Options.key("password")
58+
.stringType()
59+
.noDefaultValue()
60+
.withDescription("the password to use when connecting to the broker");
61+
62+
public static final Option<String> URL =
63+
Options.key("url")
64+
.stringType()
65+
.noDefaultValue()
66+
.withDescription(
67+
"convenience method for setting the fields in an AMQP URI: host, port, username, password and virtual host");
68+
69+
public static final Option<String> ROUTING_KEY =
70+
Options.key("routing_key")
71+
.stringType()
72+
.noDefaultValue()
73+
.withDescription("the routing key to publish the message to");
74+
75+
public static final Option<String> EXCHANGE =
76+
Options.key("exchange")
77+
.stringType()
78+
.noDefaultValue()
79+
.withDescription("the exchange to publish the message to");
80+
81+
public static final Option<Integer> NETWORK_RECOVERY_INTERVAL =
82+
Options.key("network_recovery_interval")
83+
.intType()
84+
.noDefaultValue()
85+
.withDescription(
86+
"how long will automatic recovery wait before attempting to reconnect, in ms");
87+
88+
public static final Option<Boolean> TOPOLOGY_RECOVERY_ENABLED =
89+
Options.key("topology_recovery_enabled")
90+
.booleanType()
91+
.noDefaultValue()
92+
.withDescription("if true, enables topology recovery");
93+
94+
public static final Option<Boolean> AUTOMATIC_RECOVERY_ENABLED =
95+
Options.key("AUTOMATIC_RECOVERY_ENABLED")
96+
.booleanType()
97+
.noDefaultValue()
98+
.withDescription("if true, enables connection recovery");
99+
100+
public static final Option<Integer> CONNECTION_TIMEOUT =
101+
Options.key("connection_timeout")
102+
.intType()
103+
.noDefaultValue()
104+
.withDescription("connection TCP establishment timeout in milliseconds");
105+
106+
public static final Option<Boolean> FOR_E2E_TESTING =
107+
Options.key("for_e2e_testing")
108+
.booleanType()
109+
.noDefaultValue()
110+
.withDescription("use to recognize E2E mode");
111+
112+
public static final Option<Boolean> DURABLE =
113+
Options.key("durable")
114+
.booleanType()
115+
.defaultValue(true)
116+
.withDescription(
117+
"true: The queue will survive a server restart."
118+
+ "false: The queue will be deleted on server restart.");
119+
120+
public static final Option<Boolean> EXCLUSIVE =
121+
Options.key("exclusive")
122+
.booleanType()
123+
.defaultValue(false)
124+
.withDescription(
125+
"true: The queue is used only by the current connection and will be deleted when the connection closes."
126+
+ "false: The queue can be used by multiple connections.");
127+
128+
public static final Option<Boolean> AUTO_DELETE =
129+
Options.key("auto_delete")
130+
.booleanType()
131+
.defaultValue(false)
132+
.withDescription(
133+
"true: The queue will be deleted automatically when the last consumer unsubscribes."
134+
+ "false: The queue will not be automatically deleted.");
135+
}

0 commit comments

Comments
 (0)