:fix: add auto reconnect to service client

Signed-off-by: riccardomodanese <[email protected]>

broker/artemis/plugin/src/main/java/org/eclipse/kapua/broker/artemis/plugin/security/SecurityPlugin.java

@@ -379,9 +379,14 @@ private AccountInfo getAdminAccountInfoNoCache() throws JsonProcessingException,
                 SecurityAction.getEntity.name(),
                 EntityType.account.name(),
                 SystemSetting.getInstance().getString(SystemSettingKey.SYS_ADMIN_ACCOUNT));
-        EntityResponse accountResponse = serverContext.getAuthServiceClient().getEntity(accountRequest);
-        if (accountResponse != null) {
-            return new AccountInfo(KapuaEid.parseCompactId(accountResponse.getId()), accountResponse.getName());
+        EntityResponse accountResponse;
+        try {
+            accountResponse = serverContext.getAuthServiceClient().getEntity(accountRequest);
+            if (accountResponse != null) {
+                return new AccountInfo(KapuaEid.parseCompactId(accountResponse.getId()), accountResponse.getName());
+            }
+        } catch (JsonProcessingException | JMSException | InterruptedException e) {
+            logger.warn("Error getting scopeId for user admin", e);
         }
         throw new SecurityException("User not authorized! Cannot get Admin Account info!");
     }
@@ -403,9 +408,13 @@ private KapuaId getScopeIdNoCache(String username) throws JsonProcessingExceptio
                 SecurityAction.getEntity.name(),
                 EntityType.user.name(),
                 username);
-        EntityResponse userResponse = serverContext.getAuthServiceClient().getEntity(userRequest);
-        if (userResponse != null && userResponse.getScopeId() != null) {
-            return KapuaEid.parseCompactId(userResponse.getScopeId());
+        try {
+            EntityResponse userResponse = serverContext.getAuthServiceClient().getEntity(userRequest);
+            if (userResponse != null && userResponse.getScopeId() != null) {
+                return KapuaEid.parseCompactId(userResponse.getScopeId());
+            }
+        } catch (JsonProcessingException | JMSException | InterruptedException e) {
+            logger.warn("Error getting scopeId for username {}", username, e);
         }
         throw new SecurityException("User not authorized! Cannot get scopeId for username:" + username);
     }

client/security/src/main/java/org/eclipse/kapua/client/security/ServiceClientMessagingImpl.java

@@ -13,6 +13,7 @@
 package org.eclipse.kapua.client.security;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
+
 import org.eclipse.kapua.client.security.amqpclient.Client;
 import org.eclipse.kapua.client.security.bean.AuthRequest;
 import org.eclipse.kapua.client.security.bean.AuthResponse;
@@ -44,7 +45,6 @@ public ServiceClientMessagingImpl(MessageListener messageListener, Client client
 
     @Override
     public AuthResponse brokerConnect(AuthRequest authRequest) throws InterruptedException, JMSException, JsonProcessingException {//TODO review exception when Kapua code will be linked (throw KapuaException)
-        client.checkAuthServiceConnection();
         String requestId = MessageHelper.getNewRequestId();
         authRequest.setRequestId(requestId);
         authRequest.setAction(SecurityAction.brokerConnect.name());
@@ -60,7 +60,6 @@ public AuthResponse brokerConnect(AuthRequest authRequest) throws InterruptedExc
 
     @Override
     public AuthResponse brokerDisconnect(AuthRequest authRequest) throws JMSException, InterruptedException, JsonProcessingException {
-        client.checkAuthServiceConnection();
         String requestId = MessageHelper.getNewRequestId();
         authRequest.setRequestId(requestId);
         authRequest.setAction(SecurityAction.brokerDisconnect.name());
@@ -76,7 +75,6 @@ public AuthResponse brokerDisconnect(AuthRequest authRequest) throws JMSExceptio
 
     @Override
     public EntityResponse getEntity(EntityRequest entityRequest) throws JMSException, InterruptedException, JsonProcessingException {
-        client.checkAuthServiceConnection();
         String requestId = MessageHelper.getNewRequestId();
         entityRequest.setRequestId(requestId);
         ResponseContainer<EntityResponse> responseContainer = ResponseContainer.createAnRegisterNewMessageContainer(messageListener, entityRequest);

client/security/src/main/java/org/eclipse/kapua/client/security/amqpclient/Client.java

@@ -14,6 +14,7 @@
 
 import javax.jms.Connection;
 import javax.jms.ConnectionFactory;
+import javax.jms.ExceptionListener;
 import javax.jms.JMSException;
 import javax.jms.MessageConsumer;
 import javax.jms.MessageProducer;
@@ -32,36 +33,39 @@ public class Client {
 
     private static Logger logger = LoggerFactory.getLogger(Client.class);
 
+    private static final long WAIT_BETWEEN_RECONNECTION_ATTEMPT = 2000;
+
     private ConnectionFactory connectionFactory;//is this reference needed?
     private Connection connection;//keep to implement cleanup (and object lifecycle)
     private Session session;
     private MessageConsumer consumer;
     private MessageProducer producer;
+    private String clientId;
+    private String requestAddress;
+    private String replyAddress;
+    private ClientMessageListener clientMessageListener;
+    private ExceptionListener exceptionListener;
 
-    private ConnectionStatus connectionStatus;
+    private boolean active;
+    private boolean connectionStatus;
 
     public Client(String username, String password, String host, int port, String clientId,
             String requestAddress, String replyAddress, ClientMessageListener clientMessageListener) throws JMSException {
-        connectionStatus = new ConnectionStatus();
+        this.clientId = clientId;
+        this.requestAddress = requestAddress;
+        this.replyAddress = replyAddress;
+        this.clientMessageListener = clientMessageListener;
         connectionFactory = new JmsConnectionFactory(username, password, "amqp://" + host + ":" + port);
-        connection = connectionFactory.createConnection();
-        connection.setExceptionListener(new JMSExceptionListner(connectionStatus, clientId));
-        connection.setClientID(clientId);
-        connection.start();
-        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
-        logger.info("AMQP client binding request sender to: {}", requestAddress);
-        logger.info("AMQP client binding message listener to: {}", replyAddress);
-        consumer = session.createConsumer(session.createQueue(replyAddress));
-        consumer.setMessageListener(clientMessageListener);
-        producer = session.createProducer(session.createQueue(requestAddress));
-        clientMessageListener.init(session, producer);
-        connectionStatus.setConnectionAlive();
-    }
+        exceptionListener = new ExceptionListener() {
 
-    public void checkAuthServiceConnection() {
-        if (!isConnected()) {
-            //TODO throw exception and deny operations
-        }
+            @Override
+            public void onException(JMSException exception) {
+                connectionStatus = false;
+                connect();
+            }
+        };
+        active = true;
+        connect();
     }
 
     public void sendMessage(TextMessage message) throws JMSException {
@@ -73,12 +77,61 @@ public TextMessage createTextMessage() throws JMSException {
         return session.createTextMessage();
     }
 
-    public boolean isConnected() {
+    public void stop() throws JMSException {
+        active = false;
+        disconnect();
+    }
+
+    private void disconnect() throws JMSException {
+        if (connection != null) {
+            connection.close();
+        }
+    }
+
+    private void connect() {
+        int connectAttempt = 0;
+        while (active && !isConnected()) {
+            logger.info("Connect attempt {}...", connectAttempt);
+            try {
+                logger.info("Service client {} - restarting attempt... {}", this, connectAttempt);
+                disconnect();
+                connection = connectionFactory.createConnection();
+                connection.setExceptionListener(exceptionListener);
+                connection.setClientID(clientId);
+                connection.start();
+                session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+                logger.info("AMQP client binding request sender to: {}", requestAddress);
+                logger.info("AMQP client binding message listener to: {}", replyAddress);
+                consumer = session.createConsumer(session.createQueue(replyAddress));
+                consumer.setMessageListener(clientMessageListener);
+                producer = session.createProducer(session.createQueue(requestAddress));
+                clientMessageListener.init(session, producer);
+                connectionStatus = true;
+                logger.info("Service client {} - restarting attempt... {} DONE (Connection restored)", this, connectAttempt);
+            } catch (JMSException e) {
+                logger.info("Connect attempt {}... FAIL", connectAttempt, e);
+                //wait a bit
+                waitBeforeRetry();
+            }
+            connectAttempt++;
+        }
+    }
+
+    private boolean isConnected() {
         try {
-            return connection!=null && connection.getClientID()!=null && connectionStatus.isAlive();
+            return connection!=null && connection.getClientID()!=null && connectionStatus;
         } catch (JMSException e) {
             logger.warn("Error while validating connection: {}", e.getMessage(), e);
             return false;
         }
     }
+
+    private void waitBeforeRetry() {
+        try {
+            Thread.sleep(WAIT_BETWEEN_RECONNECTION_ATTEMPT);
+        } catch (InterruptedException e) {
+            logger.error("Wait for connect interrupted!", e);
+        }
+    }
+
 }

commons/src/main/java/org/eclipse/kapua/commons/event/jms/JMSServiceEventBus.java

@@ -63,6 +63,8 @@ public class JMSServiceEventBus implements ServiceEventBus, ServiceEventBusDrive
 
     private static final Logger LOGGER = LoggerFactory.getLogger(JMSServiceEventBus.class);
 
+    private static final long WAIT_BETWEEN_RECONNECTION_ATTEMPT = 2000;
+
     private final int producerPoolMinSize;
     private final int producerPoolMaxSize;
     private final int producerPoolBorrowWait;
@@ -156,9 +158,8 @@ public ServiceEventBus getEventBus() {
     }
 
     private void waitBeforeRetry() {
-        // wait a bit
         try {
-            Thread.sleep(2000);// TODO move to configuration
+            Thread.sleep(WAIT_BETWEEN_RECONNECTION_ATTEMPT);
         } catch (InterruptedException e) {
             LOGGER.error("Wait for connect interrupted!", e);
         }
@@ -318,7 +319,7 @@ synchronized void subscribe(Subscription subscription)
                 final Session jmsSession = jmsConnection.createSession(false, Session.AUTO_ACKNOWLEDGE);
                 Topic jmsTopic = jmsSession.createTopic(subscriptionStr);
                 for (int i = 0; i < consumerPoolSize; i++) {
-                    MessageConsumer jmsConsumer = jmsSession.createSharedDurableConsumer(jmsTopic, subscription.getName());
+                    MessageConsumer jmsConsumer = jmsSession.createSharedConsumer(jmsTopic, subscription.getName());
                     jmsConsumer.setMessageListener(message -> {
                         try {
                             if (message instanceof TextMessage) {
@@ -461,6 +462,7 @@ public void onException(JMSException e) {
                         break;
                     } catch (ServiceEventBusException | JMSException e1) {
                         LOGGER.error("EventBus Listener {} - Cannot start new event bus connection... try again...", this, e1);
+                        // wait a bit
                         waitBeforeRetry();
                     }
                     i++;