From ca79258e4f64591f52e482bc39899831010b33d4 Mon Sep 17 00:00:00 2001 From: Devin Zhang Date: Mon, 4 Dec 2023 09:28:26 +0800 Subject: [PATCH] support rabbitmq ssl (#4965) --- .../rabbitmq/producer/CanalRabbitMQProducer.java | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/producer/CanalRabbitMQProducer.java b/connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/producer/CanalRabbitMQProducer.java index 6327fe34..9e4c8b19 100644 --- a/connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/producer/CanalRabbitMQProducer.java +++ b/connector/rabbitmq-connector/src/main/java/com/alibaba/otter/canal/connector/rabbitmq/producer/CanalRabbitMQProducer.java @@ -1,6 +1,9 @@ package com.alibaba.otter.canal.connector.rabbitmq.producer; import java.io.IOException; +import java.net.URISyntaxException; +import java.security.KeyManagementException; +import java.security.NoSuchAlgorithmException; import java.util.List; import java.util.Map; import java.util.Properties; @@ -51,7 +54,13 @@ public class CanalRabbitMQProducer extends AbstractMQProducer implements CanalMQ ConnectionFactory factory = new ConnectionFactory(); String servers = rabbitMQProperties.getHost(); - if (servers.contains(":")) { + if (servers.startsWith("amqp")) { + try { + factory.setUri(servers); + } catch (URISyntaxException | NoSuchAlgorithmException | KeyManagementException ex) { + throw new CanalException("failed to parse host", ex); + } + } else if (servers.contains(":")) { String[] serverHostAndPort = servers.split(":"); factory.setHost(serverHostAndPort[0]); factory.setPort(Integer.parseInt(serverHostAndPort[1]));