QPID-5088 : [Java Broker] Isolate use of AMQShortString and FieldTable to 0-8/9/9-1 protocol specific plugins

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

9 files changed, 53 insertions, 149 deletions

diff --git a/java/common/src/main/java/org/apache/qpid/AMQChannelException.java b/java/common/src/main/java/org/apache/qpid/AMQChannelException.java
index 44cd603a8d..df71ece787 100644
--- a/java/common/src/main/java/org/apache/qpid/AMQChannelException.java
+++ b/java/common/src/main/java/org/apache/qpid/AMQChannelException.java
@@ -7,9 +7,9 @@
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
- * 
+ *
  *   http://www.apache.org/licenses/LICENSE-2.0
- * 
+ *
  * Unless required by applicable law or agreed to in writing,
  * software distributed under the License is distributed on an
  * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@@ -21,6 +21,7 @@
 package org.apache.qpid;
 
 import org.apache.qpid.framing.AMQFrame;
+import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.MethodRegistry;
 import org.apache.qpid.framing.ProtocolVersion;
 import org.apache.qpid.protocol.AMQConstant;
@@ -56,7 +57,8 @@ public class AMQChannelException extends AMQException
     public AMQFrame getCloseFrame(int channel)
     {
         MethodRegistry reg = MethodRegistry.getMethodRegistry(new ProtocolVersion(major,minor));
-        return new AMQFrame(channel, reg.createChannelCloseBody(getErrorCode() == null ? AMQConstant.INTERNAL_ERROR.getCode() : getErrorCode().getCode(), getMessageAsShortString(),_classId,_methodId));
+        return new AMQFrame(channel, reg.createChannelCloseBody(getErrorCode() == null ? AMQConstant.INTERNAL_ERROR.getCode() : getErrorCode().getCode(),
+                AMQShortString.validValueOf(getMessage()),_classId,_methodId));
     }
 
 }
diff --git a/java/common/src/main/java/org/apache/qpid/AMQConnectionException.java b/java/common/src/main/java/org/apache/qpid/AMQConnectionException.java
index c7a0816f91..ef108eeb67 100644
--- a/java/common/src/main/java/org/apache/qpid/AMQConnectionException.java
+++ b/java/common/src/main/java/org/apache/qpid/AMQConnectionException.java
@@ -22,6 +22,7 @@
 package org.apache.qpid;
 
 import org.apache.qpid.framing.AMQFrame;
+import org.apache.qpid.framing.AMQShortString;
 import org.apache.qpid.framing.MethodRegistry;
 import org.apache.qpid.framing.ProtocolVersion;
 import org.apache.qpid.protocol.AMQConstant;
@@ -64,7 +65,7 @@ public class AMQConnectionException extends AMQException
         MethodRegistry reg = MethodRegistry.getMethodRegistry(new ProtocolVersion(major,minor));
         return new AMQFrame(0,
                             reg.createConnectionCloseBody(getErrorCode().getCode(),
-                                                          getMessageAsShortString(),
+                                    AMQShortString.validValueOf(getMessage()),
                                                           _classId,
                                                           _methodId));
 
diff --git a/java/common/src/main/java/org/apache/qpid/AMQException.java b/java/common/src/main/java/org/apache/qpid/AMQException.java
index 40ecc3a946..591202d934 100644
--- a/java/common/src/main/java/org/apache/qpid/AMQException.java
+++ b/java/common/src/main/java/org/apache/qpid/AMQException.java
@@ -39,7 +39,7 @@ public class AMQException extends Exception
 {
     /** Holds the AMQ error code constant associated with this exception. */
     private AMQConstant _errorCode;
-   
+
     private boolean _isHardError;
 
     /**
@@ -75,13 +75,13 @@ public class AMQException extends Exception
      * Deprecated constructors brought from M2.1
      */
     @Deprecated
-    public AMQException(String msg) 
+    public AMQException(String msg)
     {
         this(null, (msg == null) ? "" : msg);
     }
-    
-    @Deprecated 
-    public AMQException(AMQConstant errorCode, String msg) 
+
+    @Deprecated
+    public AMQException(AMQConstant errorCode, String msg)
     {
         this(errorCode, (msg == null) ? "" : msg, null);
     }
@@ -97,7 +97,7 @@ public class AMQException extends Exception
     {
         return getClass().getName() + ": " + getMessage() + (_errorCode == null ? "" : " [error code " + _errorCode + "]");
     }
-    
+
     /**
      * Gets the AMQ protocol exception code associated with this exception.
      *
@@ -141,14 +141,4 @@ public class AMQException extends Exception
         return newAMQE;
     }
 
-    /**
-     * Truncates the exception message to 255 characters if its length exceeds 255.
-     *
-     * @return exception message
-     */
-    public AMQShortString getMessageAsShortString()
-    {
-        String message = getMessage();
-        return AMQShortString.valueOf(message, true, true);
-    }
 }
diff --git a/java/common/src/main/java/org/apache/qpid/exchange/ExchangeDefaults.java b/java/common/src/main/java/org/apache/qpid/exchange/ExchangeDefaults.java
index c9bf67c11d..5e59628fb6 100644
--- a/java/common/src/main/java/org/apache/qpid/exchange/ExchangeDefaults.java
+++ b/java/common/src/main/java/org/apache/qpid/exchange/ExchangeDefaults.java
@@ -41,31 +41,30 @@ public class ExchangeDefaults
     }
 
     /** The default direct exchange, which is a special internal exchange that cannot be explicitly bound to. */
-    public static final AMQShortString DEFAULT_EXCHANGE_NAME = AMQShortString.EMPTY_STRING;
+    public static final String DEFAULT_EXCHANGE_NAME = "";
 
     /** The pre-defined topic exchange, the broker SHOULD provide this. */
-    public static final AMQShortString TOPIC_EXCHANGE_NAME = new AMQShortString("amq.topic");
+    public static final String TOPIC_EXCHANGE_NAME = "amq.topic";
 
     /** Defines the identifying type name of topic exchanges. */
-    public static final AMQShortString TOPIC_EXCHANGE_CLASS = new AMQShortString("topic");
+    public static final String TOPIC_EXCHANGE_CLASS = "topic";
 
     /** The pre-defined direct exchange, the broker MUST provide this. */
-    public static final AMQShortString DIRECT_EXCHANGE_NAME = new AMQShortString("amq.direct");
+    public static final String DIRECT_EXCHANGE_NAME = "amq.direct";
 
     /** Defines the identifying type name of direct exchanges. */
-    public static final AMQShortString DIRECT_EXCHANGE_CLASS = new AMQShortString("direct");
+    public static final String DIRECT_EXCHANGE_CLASS = "direct";
 
     /** The pre-defined headers exchange, the specification does not say this needs to be provided. */
-    public static final AMQShortString HEADERS_EXCHANGE_NAME = new AMQShortString("amq.match");
+    public static final String HEADERS_EXCHANGE_NAME = "amq.match";
 
     /** Defines the identifying type name of headers exchanges. */
-    public static final AMQShortString HEADERS_EXCHANGE_CLASS = new AMQShortString("headers");
+    public static final String HEADERS_EXCHANGE_CLASS = "headers";
 
     /** The pre-defined fanout exchange, the boker MUST provide this. */
-    public static final AMQShortString FANOUT_EXCHANGE_NAME = new AMQShortString("amq.fanout");
+    public static final String FANOUT_EXCHANGE_NAME = "amq.fanout";
 
     /** Defines the identifying type name of fanout exchanges. */
-    public static final AMQShortString FANOUT_EXCHANGE_CLASS = new AMQShortString("fanout");
-    
-    public static final AMQShortString WILDCARD_ANY = new AMQShortString("*");
+    public static final String FANOUT_EXCHANGE_CLASS = "fanout";
+
 }
diff --git a/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java b/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java
index 4adc59b158..a06fa2d15b 100644
--- a/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java
+++ b/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java
@@ -21,18 +21,15 @@
 
 package org.apache.qpid.framing;
 
+import java.util.concurrent.ConcurrentHashMap;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.DataInput;
 import java.io.DataOutput;
 import java.io.IOException;
-import java.lang.ref.WeakReference;
 import java.util.Arrays;
 import java.util.Collection;
-import java.util.LinkedHashMap;
-import java.util.Map;
-import java.util.WeakHashMap;
 
 /**
  * A short string is a representation of an AMQ Short String
@@ -64,7 +61,7 @@ public final class AMQShortString implements CharSequence, Comparable<AMQShortSt
         {
             if(_count == -1)
             {
-                _count = 1 + AMQShortString.this.occurences(_delim);
+                _count = 1 + AMQShortString.this.occurrences(_delim);
             }
             return _count;
         }
@@ -100,40 +97,8 @@ public final class AMQShortString implements CharSequence, Comparable<AMQShortSt
         return new AMQShortString(_data, from+_offset, to-from);
     }
 
-
-    private static final int LOCAL_INTERN_CACHE_SIZE = 2048;
-
-    private static final ThreadLocal<Map<AMQShortString, AMQShortString>> _localInternMap =
-            new ThreadLocal<Map<AMQShortString, AMQShortString>>()
-            {
-                protected Map<AMQShortString, AMQShortString> initialValue()
-                {
-                    return new LinkedHashMap<AMQShortString, AMQShortString>()
-                    {
-                        @Override
-                        protected boolean removeEldestEntry(Map.Entry<AMQShortString, AMQShortString> eldest)
-                        {
-                            return size() > LOCAL_INTERN_CACHE_SIZE;
-                        }
-                    };
-                };
-            };
-
-    private static final Map<AMQShortString, WeakReference<AMQShortString>> _globalInternMap =
-            new WeakHashMap<AMQShortString, WeakReference<AMQShortString>>();
-
-
-    private static final ThreadLocal<Map<String, WeakReference<AMQShortString>>> _localStringMap =
-            new ThreadLocal<Map<String, WeakReference<AMQShortString>>>()
-            {
-                protected Map<String, WeakReference<AMQShortString>> initialValue()
-                {
-                    return new WeakHashMap<String, WeakReference<AMQShortString>>();
-                };
-            };
-
-    private static final Map<String, WeakReference<AMQShortString>> _globalStringMap =
-            new WeakHashMap<String, WeakReference<AMQShortString>>();
+    private static final ConcurrentHashMap<AMQShortString, AMQShortString> _globalInternMap =
+            new ConcurrentHashMap<AMQShortString, AMQShortString>();
 
     private static final Logger _logger = LoggerFactory.getLogger(AMQShortString.class);
 
@@ -443,7 +408,7 @@ public final class AMQShortString implements CharSequence, Comparable<AMQShortSt
     {
         if (_asString == null)
         {
-            AMQShortString intern = intern();
+            AMQShortString intern = intern(false);
 
             if(intern == this)
             {
@@ -641,35 +606,9 @@ public final class AMQShortString implements CharSequence, Comparable<AMQShortSt
     public AMQShortString intern(boolean keep)
     {
 
-        hashCode();
-
-        Map<AMQShortString, AMQShortString> localMap =
-                _localInternMap.get();
-
-        AMQShortString internString = localMap.get(this);
-
+        AMQShortString internString = keep ? _globalInternMap.putIfAbsent(this,this) : _globalInternMap.get(this);
 
-        if(internString != null)
-        {
-            return internString;
-        }
-
-
-        WeakReference<AMQShortString> ref;
-        synchronized(_globalInternMap)
-        {
-
-            ref = _globalInternMap.get(this);
-            if((ref == null) || ((internString = ref.get()) == null))
-            {
-                internString = keep ? shrink() : copy();
-                ref = new WeakReference(internString);
-                _globalInternMap.put(internString, ref);
-            }
-
-        }
-        localMap.put(internString, internString);
-        return internString;
+        return internString == null ? this : internString;
 
     }
 
@@ -680,7 +619,7 @@ public final class AMQShortString implements CharSequence, Comparable<AMQShortSt
         return new AMQShortString(dataBytes,0,_length);
     }
 
-    private int occurences(final byte delim)
+    private int occurrences(final byte delim)
     {
         int count = 0;
         final int end = _offset + _length;
@@ -794,7 +733,12 @@ public final class AMQShortString implements CharSequence, Comparable<AMQShortSt
         return false;  //To change body of created methods use File | Settings | File Templates.
     }
 
-    public static AMQShortString valueOf(Object obj, boolean truncate, boolean nullAsEmptyString)
+    public static AMQShortString validValueOf(Object obj)
+    {
+        return valueOf(obj,true,true);
+    }
+
+    static AMQShortString valueOf(Object obj, boolean truncate, boolean nullAsEmptyString)
     {
         if (obj == null)
         {
@@ -826,37 +770,11 @@ public final class AMQShortString implements CharSequence, Comparable<AMQShortSt
         {
             return null;
         }
-
-        Map<String, WeakReference<AMQShortString>> localMap =
-                _localStringMap.get();
-
-        WeakReference<AMQShortString> ref = localMap.get(obj);
-        AMQShortString internString;
-
-        if(ref != null)
+        else
         {
-            internString = ref.get();
-            if(internString != null)
-            {
-                return internString;
-            }
+            return new AMQShortString(obj);
         }
 
-
-        synchronized(_globalStringMap)
-        {
-
-            ref = _globalStringMap.get(obj);
-            if((ref == null) || ((internString = ref.get()) == null))
-            {
-                internString = (new AMQShortString(obj)).intern();
-                ref = new WeakReference<AMQShortString>(internString);
-                _globalStringMap.put(obj, ref);
-            }
-
-        }
-        localMap.put(obj, ref);
-        return internString;
     }
 
     public static String toString(AMQShortString amqShortString)
@@ -864,10 +782,4 @@ public final class AMQShortString implements CharSequence, Comparable<AMQShortSt
         return amqShortString == null ? null : amqShortString.asString();
     }
 
-    public static void clearLocalCache()
-    {
-        _localInternMap.remove();
-        _localStringMap.remove();
-    }
-
 }
diff --git a/java/common/src/main/java/org/apache/qpid/framing/amqp_8_0/MethodConverter_8_0.java b/java/common/src/main/java/org/apache/qpid/framing/amqp_8_0/MethodConverter_8_0.java
index 4c7772a3a9..575816db4f 100644
--- a/java/common/src/main/java/org/apache/qpid/framing/amqp_8_0/MethodConverter_8_0.java
+++ b/java/common/src/main/java/org/apache/qpid/framing/amqp_8_0/MethodConverter_8_0.java
@@ -79,9 +79,9 @@ public class MethodConverter_8_0 extends AbstractMethodConverter implements Prot
 
         _basicPublishClassId = BasicPublishBodyImpl.CLASS_ID;
         _basicPublishMethodId = BasicPublishBodyImpl.METHOD_ID;
-                
+
     }
-   
+
     public AMQBody convertToBody(byte[] data)
     {
         return new ContentBody(data);
@@ -97,7 +97,7 @@ public class MethodConverter_8_0 extends AbstractMethodConverter implements Prot
         return new MessagePublishInfoImpl(exchange == null ? null : exchange.intern(),
                                           publishBody.getImmediate(),
                                           publishBody.getMandatory(),
-                                          routingKey == null ? null : routingKey.intern());
+                                          routingKey == null ? null : routingKey.intern(false));
 
     }
 
diff --git a/java/common/src/main/java/org/apache/qpid/url/AMQBindingURL.java b/java/common/src/main/java/org/apache/qpid/url/AMQBindingURL.java
index 11a5a3ad62..77902c3531 100644
--- a/java/common/src/main/java/org/apache/qpid/url/AMQBindingURL.java
+++ b/java/common/src/main/java/org/apache/qpid/url/AMQBindingURL.java
@@ -34,9 +34,9 @@ public class AMQBindingURL implements BindingURL
     private static final Logger _logger = LoggerFactory.getLogger(AMQBindingURL.class);
 
     private String _url;
-    private AMQShortString _exchangeClass = ExchangeDefaults.DIRECT_EXCHANGE_CLASS;
+    private AMQShortString _exchangeClass = AMQShortString.valueOf(ExchangeDefaults.DIRECT_EXCHANGE_CLASS);
     private AMQShortString _exchangeName = new AMQShortString("");
-    private AMQShortString _destinationName = new AMQShortString("");;
+    private AMQShortString _destinationName = new AMQShortString("");
     private AMQShortString _queueName = new AMQShortString("");
     private AMQShortString[] _bindingKeys = new AMQShortString[0];
     private HashMap<String, String> _options;
@@ -93,7 +93,7 @@ public class AMQBindingURL implements BindingURL
     {
 
         _exchangeClass = exchangeClass;
-        if (exchangeClass.equals(ExchangeDefaults.TOPIC_EXCHANGE_CLASS))
+        if (exchangeClass.equals(AMQShortString.valueOf(ExchangeDefaults.TOPIC_EXCHANGE_CLASS)))
         {
             setOption(BindingURL.OPTION_EXCLUSIVE, "true");
         }
@@ -147,11 +147,11 @@ public class AMQBindingURL implements BindingURL
 
     public AMQShortString getRoutingKey()
     {
-        if (_exchangeClass.equals(ExchangeDefaults.DIRECT_EXCHANGE_CLASS))
+        if (_exchangeClass.equals(AMQShortString.valueOf(ExchangeDefaults.DIRECT_EXCHANGE_CLASS)))
         {
             if (containsOption(BindingURL.OPTION_ROUTING_KEY))
             {
-                return new AMQShortString((String)getOption(OPTION_ROUTING_KEY));
+                return new AMQShortString(getOption(OPTION_ROUTING_KEY));
             }
             else
             {
@@ -161,7 +161,7 @@ public class AMQBindingURL implements BindingURL
 
         if (containsOption(BindingURL.OPTION_ROUTING_KEY))
         {
-            return new AMQShortString((String)getOption(OPTION_ROUTING_KEY));
+            return new AMQShortString(getOption(OPTION_ROUTING_KEY));
         }
 
         return getDestinationName();
@@ -191,7 +191,7 @@ public class AMQBindingURL implements BindingURL
 
     public String toString()
     {
-        StringBuffer sb = new StringBuffer();
+        StringBuilder sb = new StringBuilder();
 
         sb.append(_exchangeClass);
         sb.append("://");
@@ -207,7 +207,7 @@ public class AMQBindingURL implements BindingURL
         if (getRoutingKey() == null || getRoutingKey().toString().equals(""))
         {
 
-            if (sb.toString().indexOf("?") == -1)
+            if (!sb.toString().contains("?"))
             {
                 sb.append("?");
             }
diff --git a/java/common/src/main/java/org/apache/qpid/url/BindingURLParser.java b/java/common/src/main/java/org/apache/qpid/url/BindingURLParser.java
index 2adac843ef..ee823978b4 100644
--- a/java/common/src/main/java/org/apache/qpid/url/BindingURLParser.java
+++ b/java/common/src/main/java/org/apache/qpid/url/BindingURLParser.java
@@ -204,14 +204,14 @@ public class BindingURLParser
         // "myQueue?durable='true'" use case
         else if (nextChar == QUESTION_MARK_CHAR)
         {
-            _bindingURL.setExchangeClass(ExchangeDefaults.DIRECT_EXCHANGE_CLASS.asString());
+            _bindingURL.setExchangeClass(ExchangeDefaults.DIRECT_EXCHANGE_CLASS);
             _bindingURL.setExchangeName("");
             _bindingURL.setQueueName(builder.toString());
             return BindingURLParserState.QUESTION_MARK_CHAR;
         }
         else
         {
-            _bindingURL.setExchangeClass(ExchangeDefaults.DIRECT_EXCHANGE_CLASS.asString());
+            _bindingURL.setExchangeClass(ExchangeDefaults.DIRECT_EXCHANGE_CLASS);
             _bindingURL.setExchangeName("");
             _bindingURL.setQueueName(builder.toString());
             return BindingURLParserState.BINDING_URL_END;
diff --git a/java/common/src/test/java/org/apache/qpid/AMQExceptionTest.java b/java/common/src/test/java/org/apache/qpid/AMQExceptionTest.java
index 575058b7f8..9fceef337c 100644
--- a/java/common/src/test/java/org/apache/qpid/AMQExceptionTest.java
+++ b/java/common/src/test/java/org/apache/qpid/AMQExceptionTest.java
@@ -101,7 +101,7 @@ public class AMQExceptionTest extends TestCase
             sb.append("message [" + i + "]");
         }
         AMQException e = new AMQException(AMQConstant.INTERNAL_ERROR, sb.toString(), null);
-        AMQShortString message = e.getMessageAsShortString();
+        AMQShortString message = AMQShortString.validValueOf(e.getMessage());
         assertEquals(sb.substring(0, AMQShortString.MAX_LENGTH - 3) + "...", message.toString());
     }