QPID-5930 : [Java Broker] Minimize memory footprint for persistent messages

git-svn-id: https://svn.apache.org/repos/asf/qpid/trunk@1613950 13f79535-47bb-0310-9956-ffa450edef68

6 files changed, 100 insertions, 29 deletions

diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
index 4b37823898..ae6d607102 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQChannel.java
@@ -1249,8 +1249,6 @@ public class AMQChannel<T extends AMQProtocolSession<T>>
         final BasicContentHeaderProperties properties =
                   incomingMessage.getContentHeader().getProperties();
 
-        long expiration = properties.getExpiration();
-        message.setExpiration(expiration);
         return message;
     }
 
diff --git a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQMessage.java b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQMessage.java
index 0ed63daf7c..869de2f3a5 100644
--- a/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQMessage.java
+++ b/qpid/java/broker-plugins/amqp-0-8-protocol/src/main/java/org/apache/qpid/server/protocol/v0_8/AMQMessage.java
@@ -29,8 +29,6 @@ import org.apache.qpid.server.message.AMQMessageHeader;
 import org.apache.qpid.server.message.AbstractServerMessageImpl;
 import org.apache.qpid.server.store.StoredMessage;
 
-import java.nio.ByteBuffer;
-
 /**
  * A deliverable message.
  */
@@ -39,10 +37,6 @@ public class AMQMessage extends AbstractServerMessageImpl<AMQMessage, MessageMet
     /** Used for debugging purposes. */
     private static final Logger _log = Logger.getLogger(AMQMessage.class);
 
-    /** Flag to indicate that this message requires 'immediate' delivery. */
-
-    private long _expiration;
-
     private final long _size;
 
     public AMQMessage(StoredMessage<MessageMetaData> handle)
@@ -56,11 +50,6 @@ public class AMQMessage extends AbstractServerMessageImpl<AMQMessage, MessageMet
         _size = handle.getMetaData().getContentSize();
     }
 
-    public void setExpiration(final long expiration)
-    {
-        _expiration = expiration;
-    }
-
     public MessageMetaData getMessageMetaData()
     {
         return getStoredMessage().getMetaData();
@@ -110,16 +99,14 @@ public class AMQMessage extends AbstractServerMessageImpl<AMQMessage, MessageMet
         return getMessagePublishInfo().isImmediate();
     }
 
-
     public boolean isMandatory()
     {
         return getMessagePublishInfo().isMandatory();
     }
 
-
     public long getExpiration()
     {
-        return _expiration;
+        return getMessageHeader().getExpiration();
     }
 
 
diff --git a/qpid/java/broker-plugins/derby-store/src/test/java/org/apache/qpid/server/store/derby/DerbyMessageStoreQuotaEventsTest.java b/qpid/java/broker-plugins/derby-store/src/test/java/org/apache/qpid/server/store/derby/DerbyMessageStoreQuotaEventsTest.java
index 5c8eb94d91..e39a35f9a4 100644
--- a/qpid/java/broker-plugins/derby-store/src/test/java/org/apache/qpid/server/store/derby/DerbyMessageStoreQuotaEventsTest.java
+++ b/qpid/java/broker-plugins/derby-store/src/test/java/org/apache/qpid/server/store/derby/DerbyMessageStoreQuotaEventsTest.java
@@ -20,6 +20,9 @@
  */
 package org.apache.qpid.server.store.derby;
 
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
 import java.util.Collections;
 import java.util.Map;
 
@@ -28,9 +31,6 @@ import org.apache.qpid.server.store.MessageStore;
 import org.apache.qpid.server.store.MessageStoreQuotaEventsTestBase;
 import org.apache.qpid.server.virtualhost.derby.DerbyVirtualHost;
 
-import static org.mockito.Mockito.mock;
-import static org.mockito.Mockito.when;
-
 public class DerbyMessageStoreQuotaEventsTest extends MessageStoreQuotaEventsTestBase
 {
     private static final int NUMBER_OF_MESSAGES_TO_OVERFILL_STORE = 10;
@@ -54,6 +54,7 @@ public class DerbyMessageStoreQuotaEventsTest extends MessageStoreQuotaEventsTes
     {
         final DerbyVirtualHost parent = mock(DerbyVirtualHost.class);
         when(parent.getContext()).thenReturn(createContextSettings());
+        when(parent.getContextKeys()).thenReturn(Collections.emptySet());
         when(parent.getStorePath()).thenReturn(storeLocation);
         when(parent.getStoreOverfullSize()).thenReturn(OVERFULL_SIZE);
         when(parent.getStoreUnderfullSize()).thenReturn(UNDERFULL_SIZE);
diff --git a/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/GenericJDBCConfigurationStore.java b/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/GenericJDBCConfigurationStore.java
index 4f88e011fb..9453c135e9 100644
--- a/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/GenericJDBCConfigurationStore.java
+++ b/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/GenericJDBCConfigurationStore.java
@@ -27,6 +27,7 @@ import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import javax.security.auth.Subject;
@@ -84,7 +85,7 @@ public class GenericJDBCConfigurationStore extends AbstractJDBCConfigurationStor
             JDBCSettings settings = (JDBCSettings)parent;
             _connectionURL = settings.getConnectionUrl();
 
-            JDBCDetails details = JDBCDetails.getDetailsForJdbcUrl(_connectionURL, parent.getContext());
+            JDBCDetails details = JDBCDetails.getDetailsForJdbcUrl(_connectionURL, parent);
 
             if (!details.isKnownVendor() && getLogger().isInfoEnabled())
             {
@@ -111,8 +112,13 @@ public class GenericJDBCConfigurationStore extends AbstractJDBCConfigurationStor
 
             try
             {
-                Map<String, String> providerAttributes = new HashMap(_parent.getContext());
-                providerAttributes.keySet().retainAll(connectionProviderFactory.getProviderAttributeNames());
+                Map<String, String> providerAttributes = new HashMap<>();
+                Set<String> providerAttributeNames = connectionProviderFactory.getProviderAttributeNames();
+                providerAttributeNames.retainAll(parent.getContextKeys());
+                for(String attr : providerAttributeNames)
+                {
+                    providerAttributes.put(attr, parent.getContextValue(String.class, attr));
+                }
 
                 _connectionProvider = connectionProviderFactory.getConnectionProvider(_connectionURL,
                                                                                       settings.getUsername(),
diff --git a/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/GenericJDBCMessageStore.java b/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/GenericJDBCMessageStore.java
index 4fde0a44c7..3304d01d86 100644
--- a/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/GenericJDBCMessageStore.java
+++ b/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/GenericJDBCMessageStore.java
@@ -28,6 +28,9 @@ import java.sql.ResultSet;
 import java.sql.SQLException;
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Set;
+
+import javax.security.auth.Subject;
 
 import org.apache.log4j.Logger;
 
@@ -36,8 +39,6 @@ import org.apache.qpid.server.plugin.JDBCConnectionProviderFactory;
 import org.apache.qpid.server.security.SecurityManager;
 import org.apache.qpid.server.store.StoreException;
 
-import javax.security.auth.Subject;
-
 /**
  * Implementation of a MessageStore backed by a Generic JDBC Database.
  */
@@ -60,7 +61,7 @@ public class GenericJDBCMessageStore extends GenericAbstractJDBCMessageStore
         JDBCSettings settings = (JDBCSettings)parent;
         _connectionURL = settings.getConnectionUrl();
 
-        JDBCDetails details = JDBCDetails.getDetailsForJdbcUrl(_connectionURL, parent.getContext());
+        JDBCDetails details = JDBCDetails.getDetailsForJdbcUrl(_connectionURL, parent);
 
         if (!details.isKnownVendor() && getLogger().isInfoEnabled())
         {
@@ -90,9 +91,13 @@ public class GenericJDBCMessageStore extends GenericAbstractJDBCMessageStore
 
         try
         {
-            Map<String, String> providerAttributes = new HashMap(parent.getContext());
-            providerAttributes.keySet().retainAll(connectionProviderFactory.getProviderAttributeNames());
-
+            Map<String, String> providerAttributes = new HashMap<>();
+            Set<String> providerAttributeNames = connectionProviderFactory.getProviderAttributeNames();
+            providerAttributeNames.retainAll(parent.getContextKeys());
+            for(String attr : providerAttributeNames)
+            {
+                providerAttributes.put(attr, parent.getContextValue(String.class, attr));
+            }
 
             _connectionProvider = connectionProviderFactory.getConnectionProvider(_connectionURL,
                                                                                   settings.getUsername(),
diff --git a/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/JDBCDetails.java b/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/JDBCDetails.java
index a74f852dfa..8cd4996033 100644
--- a/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/JDBCDetails.java
+++ b/qpid/java/broker-plugins/jdbc-store/src/main/java/org/apache/qpid/server/store/jdbc/JDBCDetails.java
@@ -19,9 +19,15 @@
 
 package org.apache.qpid.server.store.jdbc;
 
+import java.util.AbstractMap;
+import java.util.AbstractSet;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.Iterator;
 import java.util.Map;
+import java.util.Set;
+
+import org.apache.qpid.server.model.ConfiguredObject;
 
 public abstract class JDBCDetails
 {
@@ -216,7 +222,75 @@ public abstract class JDBCDetails
         return result;
     }
 
-
+    public static JDBCDetails getDetailsForJdbcUrl(String jdbcUrl, final ConfiguredObject<?> object)
+    {
+        final Set<String> contextKeys = object.getContextKeys();
+        Map<String,String> mapConversion = new AbstractMap<String, String>()
+        {
+            @Override
+            public Set<Entry<String, String>> entrySet()
+            {
+                return new AbstractSet<Entry<String, String>>()
+                {
+                    @Override
+                    public Iterator<Entry<String, String>> iterator()
+                    {
+                        final Iterator<String> underlying = contextKeys.iterator();
+                        return new Iterator<Entry<String, String>>()
+                        {
+                            @Override
+                            public boolean hasNext()
+                            {
+                                return underlying.hasNext();
+                            }
+
+                            @Override
+                            public Entry<String, String> next()
+                            {
+                                final String key = underlying.next();
+                                final String value = object.getContextValue(String.class, key);
+                                return new Entry<String,String>()
+                                {
+
+                                    @Override
+                                    public String getKey()
+                                    {
+                                        return key;
+                                    }
+
+                                    @Override
+                                    public String getValue()
+                                    {
+                                        return value;
+                                    }
+
+                                    @Override
+                                    public String setValue(final String value)
+                                    {
+                                        throw new UnsupportedOperationException();
+                                    }
+                                };
+
+                            }
+
+                            @Override
+                            public void remove()
+                            {
+                                throw new UnsupportedOperationException();
+                            }
+                        };
+                    }
+
+                    @Override
+                    public int size()
+                    {
+                        return contextKeys.size();
+                    }
+                };
+            }
+        };
+        return getDetailsForJdbcUrl(jdbcUrl, mapConversion);
+    }
     public static JDBCDetails getDetailsForJdbcUrl(String jdbcUrl, final Map<String, String> contextMap)
     {
         String[] components = jdbcUrl.split(":", 3);