Create a single ActiveMQ session per connection, to prevent OOM errors

grouper-misc/grouper-messaging-activemq/src/main/java/edu/internet2/middleware/grouperMessagingActiveMQ/GrouperMessagingActiveMQSystem.java

@@ -65,7 +65,7 @@ public GrouperMessageSendResult send(GrouperMessageSendParam grouperMessageSendP
       Connection connection = ActiveMQClientConnectionFactory.INSTANCE.getActiveMQConnection(systemParam.getMessageSystemName());
 
       // Create a non-transactional session with automatic acknowledgement
-      Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+      Session session =  ActiveMQClientConnectionFactory.INSTANCE.getActiveMQSendSession(systemParam.getMessageSystemName());
 
       Destination destination = null;
       if (queueParam.getQueueType() == GrouperMessageQueueType.queue) {
@@ -236,6 +236,29 @@ private enum ActiveMQClientConnectionFactory {
     INSTANCE;
 
     private Map<String, Connection> messagingSystemNameConnection = new HashMap<String, Connection>();
+
+    private Map<String,Session> messagingSystemNameSession = new HashMap<>();
+
+    private Session getActiveMQSendSession(String messagingSystemName) throws JMSException {
+      if (StringUtils.isBlank(messagingSystemName)) {
+        throw new IllegalArgumentException("messagingSystemName is required.");
+      }
+
+      Connection connection =  messagingSystemNameConnection.get(messagingSystemName);
+
+      if (connection == null) {
+        throw new JMSException("Connection does not exist. Create a connection first");
+      }
+
+      Session session =  messagingSystemNameSession.get(messagingSystemName);
+
+      if (session == null) {
+        session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
+        messagingSystemNameSession.put(messagingSystemName,session);
+      }
+
+      return session;
+    }
 
     private Connection getActiveMQConnection(String messagingSystemName) throws JMSException {
 
@@ -255,7 +278,7 @@ private Connection getActiveMQConnection(String messagingSystemName) throws JMSE
           String uri =  grouperMessagingConfig.propertyValueString(GrouperClientConfig.retrieveConfig(), "uri");
           String username = grouperMessagingConfig.propertyValueString(GrouperClientConfig.retrieveConfig(), "username");
           String password = grouperMessagingConfig.propertyValueString(GrouperClientConfig.retrieveConfig(), "password");
-          
+
           if (StringUtils.isNotBlank(password)) {
             password = GrouperClientUtils.decryptFromFileIfFileExists(password, null);
           }
@@ -279,7 +302,6 @@ private Connection getActiveMQConnection(String messagingSystemName) throws JMSE
           connection = factory.createConnection();
           connection.start();
           messagingSystemNameConnection.put(messagingSystemName, connection);
-
         }
       }
       return connection;
@@ -298,6 +320,7 @@ private void closeConnection(String messagingSystemName) throws JMSException {
             throw new RuntimeException("Error occurred while closing ActiveMQ connection for "+messagingSystemName, e);
           } finally {
             messagingSystemNameConnection.remove(messagingSystemName);
+            messagingSystemNameSession.remove(messagingSystemName);
           }
         }
       }

grouper-misc/grouper-messaging-activemq/src/main/resources/grouper.client.activeMq.example.properties

@@ -23,7 +23,7 @@ grouper.messaging.system.activeMqSystem.username =
 
 # password of activemq queue
 grouper.messaging.system.activeMqSystem.password =
- 
+
 # number of seconds to sleep while waiting
 grouper.messaging.system.activeMqSystem.polling.sleep.seconds = 5
 

grouper-misc/grouperClient/conf/grouper.client.base.properties

@@ -989,7 +989,7 @@ grouper.messaging.system.grouperBuiltinMessaging.class = edu.internet2.middlewar
 
 # password of activemq queue
 # {valueType: "password", required: true}
-# grouper.messaging.system.activeMqMessaging.password = 
+# grouper.messaging.system.activeMqMessaging.password =
 
 # if this activemq connector is enabled
 # {valueType: "boolean", defaultValue: "true"}

grouper-misc/grouperClient/src/java/edu/internet2/middleware/grouperClient/messaging/GrouperMessagingConfig.java

@@ -137,6 +137,6 @@ public String propertyValueString(GrouperClientConfig grouperClientConfig, Strin
 
     return null;
   }
-  
+
 }