reuse rabbitMQ connection and channel (#27)

Author: dcherix

src/main/java/com/github/aznamier/keycloak/event/provider/RabbitMqEventListenerProvider.java

@@ -15,37 +15,18 @@
 import com.rabbitmq.client.AMQP.BasicProperties;
 import com.rabbitmq.client.AMQP.BasicProperties.Builder;
 import com.rabbitmq.client.Channel;
-import com.rabbitmq.client.Connection;
-import com.rabbitmq.client.ConnectionFactory;
 
 public class RabbitMqEventListenerProvider implements EventListenerProvider {
 
 	private static final Logger log = Logger.getLogger(RabbitMqEventListenerProvider.class);
-
+	
 	private final RabbitMqConfig cfg;
-	private final ConnectionFactory factory;
+	private Channel channel;
 
 	private final EventListenerTransaction tx = new EventListenerTransaction(this::publishAdminEvent, this::publishEvent);
 
-	public RabbitMqEventListenerProvider(RabbitMqConfig cfg, KeycloakSession session) {
+	public RabbitMqEventListenerProvider(Channel channel, KeycloakSession session, RabbitMqConfig cfg) {
 		this.cfg = cfg;
-		
-		this.factory = new ConnectionFactory();
-
-		this.factory.setUsername(cfg.getUsername());
-		this.factory.setPassword(cfg.getPassword());
-		this.factory.setVirtualHost(cfg.getVhost());
-		this.factory.setHost(cfg.getHostUrl());
-		this.factory.setPort(cfg.getPort());
-
-		if(cfg.getUseTls()) {
-			try {
-				this.factory.useSslProtocol();
-			} catch (Exception e) {
-				log.error("Could not use SSL protocol", e);
-			}
-		}
-
 		session.getTransactionManager().enlistAfterCompletion(tx);
 	}
 
@@ -97,16 +78,11 @@ private static BasicProperties getMessageProps(String className) {
 
 	private void publishNotification(String messageString, BasicProperties props, String routingKey) {
 		try {
-			Connection conn = factory.newConnection();
-			Channel channel = conn.createChannel();
-			
 			channel.basicPublish(cfg.getExchange(), routingKey, props, messageString.getBytes(StandardCharsets.UTF_8));
-			log.infof("keycloak-to-rabbitmq SUCCESS sending message: %s%n", routingKey);
-			channel.close();
-			conn.close();
-
+			log.tracef("keycloak-to-rabbitmq SUCCESS sending message: %s%n", routingKey);
 		} catch (Exception ex) {
 			log.errorf(ex, "keycloak-to-rabbitmq ERROR sending message: %s%n", routingKey);
 		}
 	}
-}
+
+}

src/main/java/com/github/aznamier/keycloak/event/provider/RabbitMqEventListenerProviderFactory.java

@@ -1,5 +1,11 @@
 package com.github.aznamier.keycloak.event.provider;
 
+import com.rabbitmq.client.Channel;
+import com.rabbitmq.client.Connection;
+import com.rabbitmq.client.ConnectionFactory;
+import java.io.IOException;
+import java.util.concurrent.TimeoutException;
+import org.jboss.logging.Logger;
 import org.keycloak.Config.Scope;
 import org.keycloak.events.EventListenerProvider;
 import org.keycloak.events.EventListenerProviderFactory;
@@ -8,31 +14,73 @@
 
 public class RabbitMqEventListenerProviderFactory implements EventListenerProviderFactory {
 
-	private RabbitMqConfig cfg;
+    private static final Logger log = Logger.getLogger(RabbitMqEventListenerProviderFactory.class);
+    private RabbitMqConfig cfg;
+    private ConnectionFactory connectionFactory;
+    private Connection connection;
+    private Channel channel;
 
-	@Override
-	public EventListenerProvider create(KeycloakSession session) {
-		return new RabbitMqEventListenerProvider(cfg, session);
-	}
+    @Override
+    public EventListenerProvider create(KeycloakSession session) {
+        checkConnectionAndChannel();
+        return new RabbitMqEventListenerProvider(channel, session, cfg);
+    }
 
-	@Override
-	public void init(Scope config) {
-		cfg = RabbitMqConfig.createFromScope(config);
-	}
+    private synchronized void checkConnectionAndChannel() {
+        try {
+            if (connection == null || !connection.isOpen()) {
+                this.connection = connectionFactory.newConnection();
+            }
+            if (channel == null || !channel.isOpen()) {
+                channel = connection.createChannel();
+            }
+        }
+        catch (IOException | TimeoutException e) {
+            log.error("keycloak-to-rabbitmq ERROR on connection to rabbitmq", e);
+        }
+    }
 
-	@Override
-	public void postInit(KeycloakSessionFactory factory) {
+    @Override
+    public void init(Scope config) {
+        cfg = RabbitMqConfig.createFromScope(config);
+        this.connectionFactory = new ConnectionFactory();
 
-	}
+        this.connectionFactory.setUsername(cfg.getUsername());
+        this.connectionFactory.setPassword(cfg.getPassword());
+        this.connectionFactory.setVirtualHost(cfg.getVhost());
+        this.connectionFactory.setHost(cfg.getHostUrl());
+        this.connectionFactory.setPort(cfg.getPort());
+        this.connectionFactory.setAutomaticRecoveryEnabled(true);
 
-	@Override
-	public void close() {
+        if (cfg.getUseTls()) {
+            try {
+                this.connectionFactory.useSslProtocol();
+            }
+            catch (Exception e) {
+                log.error("Could not use SSL protocol", e);
+            }
+        }
+    }
 
-	}
+    @Override
+    public void postInit(KeycloakSessionFactory factory) {
 
-	@Override
-	public String getId() {
-		return "keycloak-to-rabbitmq";
-	}
+    }
+
+    @Override
+    public void close() {
+        try {
+            channel.close();
+            connection.close();
+        }
+        catch (IOException | TimeoutException e) {
+            log.error("keycloak-to-rabbitmq ERROR on close", e);
+        }
+    }
+
+    @Override
+    public String getId() {
+        return "keycloak-to-rabbitmq";
+    }
 
 }