Allowing to connect to RabbitMQ brokers that require SSL

Amended doc accordingly.

Author: wk8

config.properties.example

@@ -239,6 +239,7 @@ kafka.acks=1
 #rabbitmq_routing_key_template=%db%.%table%
 #rabbitmq_message_persistent=false
 #rabbitmq_declare_exchange=true
+#rabbitmq_use_ssl=false
 
 #           *** redis ***
 

docs/docs/config.md

@@ -156,6 +156,7 @@ rabbitmq_exchange_autodelete   | BOOLEAN    | If set, the exchange is deleted wh
 rabbitmq_routing_key_template  | STRING     | A string template for the routing key, `%db%` and `%table%` will be substituted. | `%db%.%table%`.
 rabbitmq_message_persistent    | BOOLEAN    | Eanble message persistence. | false
 rabbitmq_declare_exchange      | BOOLEAN    | Should declare the exchange for rabbitmq publisher | true
+rabbitmq_use_ssl               | BOOLEAN    | If true, will connect to the server using SSL. | false
 
 _See also:_ [RabbitMQ Producer Documentation](/producers#rabbitmq)
 

docs/docs/producers.md

@@ -308,6 +308,7 @@ The remaining configurable properties are:
     - This config controls the routing key, where `%db%` and `%table%` are placeholders that will be substituted at runtime
 - `rabbitmq_message_persistent` - defaults to **false**
 - `rabbitmq_declare_exchange` - defaults to **true**
+- `rabbitmq_use_ssl` - defaults to **false**
 
 For more details on these options, you are encouraged to the read official RabbitMQ documentation here: [https://www.rabbitmq.com/documentation.html](https://www.rabbitmq.com/documentation.html)
 

src/main/java/com/zendesk/maxwell/MaxwellConfig.java

@@ -545,6 +545,10 @@ public class MaxwellConfig extends AbstractConfig {
 	 */
 	public boolean rabbitmqDeclareExchange;
 
+	/**
+	 * {@link com.zendesk.maxwell.producer.RabbitmqProducer} use SSL
+	 */
+	public boolean rabbitmqUseSSL;
 
 	/**
 	 * {@link com.zendesk.maxwell.producer.NatsProducer} URL
@@ -955,6 +959,7 @@ protected MaxwellOptionParser buildOptionParser() {
 		parser.accepts( "rabbitmq_routing_key_template", "A string template for the routing key, '%db%' and '%table%' will be substituted. Default is '%db%.%table%'." ).withRequiredArg();
 		parser.accepts( "rabbitmq_message_persistent", "Message persistence. Defaults to false" ).withOptionalArg();
 		parser.accepts( "rabbitmq_declare_exchange", "Should declare the exchange for rabbitmq publisher. Defaults to true" ).withOptionalArg();
+		parser.accepts( "rabbitmq_use_ssl", "If true, will connect to the server using SSL. Defaults to false" ).withOptionalArg();
 
 		parser.section( "redis" );
 
@@ -1107,6 +1112,7 @@ private void setup(OptionSet options, Properties properties) {
 		this.rabbitmqRoutingKeyTemplate   	= fetchStringOption("rabbitmq_routing_key_template", options, properties, "%db%.%table%");
 		this.rabbitmqMessagePersistent    	= fetchBooleanOption("rabbitmq_message_persistent", options, properties, false);
 		this.rabbitmqDeclareExchange		= fetchBooleanOption("rabbitmq_declare_exchange", options, properties, true);
+		this.rabbitmqUseSSL			= fetchBooleanOption("rabbitmq_use_ssl", options, properties, false);
 
 		this.natsUrl			= fetchStringOption("nats_url", options, properties, "nats://localhost:4222");
 		this.natsSubject		= fetchStringOption("nats_subject", options, properties, "%{database}.%{table}");

src/main/java/com/zendesk/maxwell/producer/RabbitmqProducer.java

@@ -10,6 +10,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.net.ssl.SSLContext;
 import java.io.IOException;
 import java.net.URISyntaxException;
 import java.security.KeyManagementException;
@@ -61,6 +62,10 @@ public RabbitmqProducer(MaxwellContext context) {
 				factory.setHandshakeTimeout(config.rabbitmqHandshakeTimeout);
 			}
 
+			if ( config.rabbitmqUseSSL ) {
+				factory.useSslProtocol(SSLContext.getDefault());
+			}
+
 			this.channel = factory.newConnection().createChannel();
 			if(context.getConfig().rabbitmqDeclareExchange) {
 				this.channel.exchangeDeclare(exchangeName, context.getConfig().rabbitmqExchangeType, context.getConfig().rabbitMqExchangeDurable, context.getConfig().rabbitMqExchangeAutoDelete, null);