QPID-4984: WIP - Merge from trunk r.1525056

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

16 files changed, 214 insertions, 345 deletions

diff --git a/qpid/java/common/src/main/java/common.bnd b/qpid/java/common/src/main/java/common.bnd
index 5635012faa..4d6cd3d844 100755
--- a/qpid/java/common/src/main/java/common.bnd
+++ b/qpid/java/common/src/main/java/common.bnd
@@ -17,7 +17,7 @@
 # under the License.

 #

 

-ver: 0.23.0

+ver: 0.25.0

 

 Bundle-SymbolicName: qpid-common

 Bundle-Version: ${ver}

diff --git a/qpid/java/common/src/main/java/org/apache/qpid/AMQChannelException.java b/qpid/java/common/src/main/java/org/apache/qpid/AMQChannelException.java
index 44cd603a8d..df71ece787 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/AMQChannelException.java
+++ b/qpid/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/qpid/java/common/src/main/java/org/apache/qpid/AMQConnectionException.java b/qpid/java/common/src/main/java/org/apache/qpid/AMQConnectionException.java
index c7a0816f91..ef108eeb67 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/AMQConnectionException.java
+++ b/qpid/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/qpid/java/common/src/main/java/org/apache/qpid/AMQException.java b/qpid/java/common/src/main/java/org/apache/qpid/AMQException.java
index 40ecc3a946..591202d934 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/AMQException.java
+++ b/qpid/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/qpid/java/common/src/main/java/org/apache/qpid/exchange/ExchangeDefaults.java b/qpid/java/common/src/main/java/org/apache/qpid/exchange/ExchangeDefaults.java
index c9bf67c11d..5e59628fb6 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/exchange/ExchangeDefaults.java
+++ b/qpid/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/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java b/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java
index 4adc59b158..a06fa2d15b 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQShortString.java
+++ b/qpid/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/qpid/java/common/src/main/java/org/apache/qpid/framing/amqp_8_0/MethodConverter_8_0.java b/qpid/java/common/src/main/java/org/apache/qpid/framing/amqp_8_0/MethodConverter_8_0.java
index 4c7772a3a9..575816db4f 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/framing/amqp_8_0/MethodConverter_8_0.java
+++ b/qpid/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/qpid/java/common/src/main/java/org/apache/qpid/ssl/SSLContextFactory.java b/qpid/java/common/src/main/java/org/apache/qpid/ssl/SSLContextFactory.java
index 158006f072..e9dec362a6 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/ssl/SSLContextFactory.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/ssl/SSLContextFactory.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
@@ -22,7 +22,6 @@ package org.apache.qpid.ssl;
 
 import org.apache.qpid.transport.network.security.ssl.QpidClientX509KeyManager;
 import org.apache.qpid.transport.network.security.ssl.QpidMultipleTrustManager;
-import org.apache.qpid.transport.network.security.ssl.QpidPeersOnlyTrustManager;
 import org.apache.qpid.transport.network.security.ssl.SSLUtil;
 
 import javax.net.ssl.KeyManager;
@@ -42,55 +41,17 @@ import java.util.Collections;
 /**
  * Factory used to create SSLContexts. SSL needs to be configured
  * before this will work.
- * 
+ *
  */
 public class SSLContextFactory
 {
     public static final String TRANSPORT_LAYER_SECURITY_CODE = "TLS";
-    
-    public static class TrustStoreWrapper
-    {
-        private final String trustStorePath;
-        private final String trustStorePassword;
-        private final String trustStoreType;
-        private final Boolean trustStorePeersOnly;
-        private String trustManagerFactoryAlgorithm;
-        
-        public TrustStoreWrapper(final String trustStorePath, final String trustStorePassword,
-                final String trustStoreType, final Boolean trustStorePeersOnly,
-                final String trustManagerFactoryAlgorithm)
-        {
-            this.trustStorePath = trustStorePath;
-            this.trustStorePassword = trustStorePassword;
-            this.trustStoreType = trustStoreType;
-            this.trustStorePeersOnly = trustStorePeersOnly;
-            this.trustManagerFactoryAlgorithm = trustManagerFactoryAlgorithm;
-        }
-    }
 
     private SSLContextFactory()
     {
         //no instances
     }
 
-    public static SSLContext buildServerContext(final String keyStorePath,
-            final String keyStorePassword, final String keyStoreType,
-            final String keyManagerFactoryAlgorithm)
-            throws GeneralSecurityException, IOException
-    {
-        return buildContext(Collections.<TrustStoreWrapper>emptyList(), keyStorePath,
-                keyStorePassword, keyStoreType, keyManagerFactoryAlgorithm, null);
-    }
-    
-    public static SSLContext buildClientContext(Collection<TrustStoreWrapper> trustStores,
-            final String keyStorePath, final String keyStorePassword,
-            final String keyStoreType, final String keyManagerFactoryAlgorithm,
-            final String certAlias) throws GeneralSecurityException, IOException
-    {
-        return buildContext(trustStores, keyStorePath, keyStorePassword, keyStoreType,
-                            keyManagerFactoryAlgorithm, certAlias);
-    }
-
     public static SSLContext buildClientContext(final String trustStorePath,
             final String trustStorePassword, final String trustStoreType,
             final String trustManagerFactoryAlgorithm, final String keyStorePath,
@@ -98,17 +59,25 @@ public class SSLContextFactory
             final String keyManagerFactoryAlgorithm, final String certAlias)
             throws GeneralSecurityException, IOException
     {
-        TrustStoreWrapper trstWrapper = new TrustStoreWrapper(trustStorePath, trustStorePassword,
-                                                              trustStoreType, Boolean.FALSE,
-                                                              trustManagerFactoryAlgorithm);
-        return buildContext(Collections.singletonList(trstWrapper), keyStorePath,
-                keyStorePassword, keyStoreType, keyManagerFactoryAlgorithm, certAlias);
+        return buildContext(trustStorePath,
+                trustStorePassword,
+                trustStoreType,
+                trustManagerFactoryAlgorithm,
+                keyStorePath,
+                keyStorePassword,
+                keyStoreType,
+                keyManagerFactoryAlgorithm,
+                certAlias);
     }
-    
-    private static SSLContext buildContext(final Collection<TrustStoreWrapper> trstWrappers,
-            final String keyStorePath, final String keyStorePassword, 
-            final String keyStoreType, final String keyManagerFactoryAlgorithm,
-            final String certAlias)
+
+    private static SSLContext buildContext(String trustStorePath,
+                                           String trustStorePassword,
+                                           String trustStoreType,
+                                           String trustManagerFactoryAlgorithm,
+                                           String keyStorePath,
+                                           String keyStorePassword,
+                                           String keyStoreType,
+                                           String keyManagerFactoryAlgorithm, String certAlias)
             throws GeneralSecurityException, IOException
     {
         // Initialize the SSLContext to work with our key managers.
@@ -117,53 +86,20 @@ public class SSLContextFactory
 
         final TrustManager[] trustManagers;
         final KeyManager[] keyManagers;
-        
-        final Collection<TrustManager> trustManagersCol = new ArrayList<TrustManager>();
-        final QpidMultipleTrustManager mulTrustManager = new QpidMultipleTrustManager();
-        for (TrustStoreWrapper tsw : trstWrappers)
-        {
-            if (tsw.trustStorePath != null)
-            {
-                final KeyStore ts = SSLUtil.getInitializedKeyStore(tsw.trustStorePath,
-                        tsw.trustStorePassword, tsw.trustStoreType);
-                final TrustManagerFactory tmf = TrustManagerFactory
-                        .getInstance(tsw.trustManagerFactoryAlgorithm);
-                tmf.init(ts);
-                TrustManager[] delegateManagers = tmf.getTrustManagers();
-                for (TrustManager tm : delegateManagers)
-                {
-                    if (tm instanceof X509TrustManager)
-                    {
-                        if (Boolean.TRUE.equals(tsw.trustStorePeersOnly))
-                        {
-                            // truststore is supposed to trust only clients which peers certificates
-                            // are directly in the store. CA signing will not be considered.
-                            mulTrustManager.addTrustManager(new QpidPeersOnlyTrustManager(ts, (X509TrustManager) tm));
-                        }
-                        else
-                        {
-                            mulTrustManager.addTrustManager((X509TrustManager) tm);
-                        }
-                    }
-                    else
-                    {
-                        trustManagersCol.add(tm);
-                    }
-                }
-            }
-        }
-        if (! mulTrustManager.isEmpty())
-        {
-            trustManagersCol.add(mulTrustManager);
-        }
-        
-        if (trustManagersCol.isEmpty())
+
+        if (trustStorePath != null)
         {
-            trustManagers = null;
+            final KeyStore ts = SSLUtil.getInitializedKeyStore(trustStorePath,
+                    trustStorePassword, trustStoreType);
+            final TrustManagerFactory tmf = TrustManagerFactory
+                    .getInstance(trustManagerFactoryAlgorithm);
+            tmf.init(ts);
+
+            trustManagers = tmf.getTrustManagers();
         }
         else
         {
-            trustManagers = trustManagersCol.toArray(new TrustManager[trustManagersCol.size()]);
+            trustManagers = null;
         }
 
         if (keyStorePath != null)
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayerFactory.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayerFactory.java
index 478355edc1..bfd1ae8181 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayerFactory.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/SecurityLayerFactory.java
@@ -124,7 +124,7 @@ public class SecurityLayerFactory
 
         public String getUserID()
         {
-            return SSLUtil.retriveIdentity(_engine);
+            return SSLUtil.retrieveIdentity(_engine);
         }
     }
 
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLUtil.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLUtil.java
index a527c436f8..7553e3dad8 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLUtil.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/ssl/SSLUtil.java
@@ -1,5 +1,5 @@
 /*
- * 
+ *
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
@@ -7,19 +7,22 @@
  * 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
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
- * 
+ *
  */
 package org.apache.qpid.transport.network.security.ssl;
 
+import javax.naming.InvalidNameException;
+import javax.naming.ldap.LdapName;
+import javax.naming.ldap.Rdn;
 import org.apache.qpid.transport.TransportException;
 import org.apache.qpid.transport.util.Logger;
 
@@ -50,21 +53,21 @@ public class SSLUtil
           Certificate cert = engine.getSession().getPeerCertificates()[0];
           Principal p = ((X509Certificate)cert).getSubjectDN();
           String dn = p.getName();
-          String hostname = null; 
-          
+          String hostname = null;
+
           if (dn.contains("CN="))
           {
               hostname = dn.substring(3,
                       dn.indexOf(",") == -1? dn.length(): dn.indexOf(","));
-          }   
-          
+          }
+
           if (log.isDebugEnabled())
           {
               log.debug("Hostname expected : " + hostnameExpected);
               log.debug("Distinguished Name for server certificate : " + dn);
               log.debug("Host Name obtained from DN : " + hostname);
           }
-          
+
           if (hostname != null && !(hostname.equalsIgnoreCase(hostnameExpected) ||
                   hostname.equalsIgnoreCase(hostnameExpected + ".localdomain")))
           {
@@ -72,60 +75,78 @@ public class SSLUtil
                                            " Expected : " + hostnameExpected +
                                            " Found in cert : " + hostname);
           }
-          
+
         }
         catch(SSLPeerUnverifiedException e)
         {
             log.warn("Exception received while trying to verify hostname",e);
             // For some reason the SSL engine sets the handshake status to FINISH twice
-            // in succession. The first time the peer certificate 
+            // in succession. The first time the peer certificate
             // info is not available. The second time it works !
             // Therefore have no choice but to ignore the exception here.
         }
     }
-    
-    public static String retriveIdentity(SSLEngine engine)
+
+    public static String getIdFromSubjectDN(String dn)
     {
-        StringBuffer id = new StringBuffer(); 
+        String cnStr = null;
+        String dcStr = null;
+        if(dn == null)
+        {
+            return "";
+        }
+        else
+        {
+            try
+            {
+                LdapName ln = new LdapName(dn);
+                for(Rdn rdn : ln.getRdns())
+                {
+                    if("CN".equalsIgnoreCase(rdn.getType()))
+                    {
+                        cnStr = rdn.getValue().toString();
+                    }
+                    else if("DC".equalsIgnoreCase(rdn.getType()))
+                    {
+                        if(dcStr == null)
+                        {
+                            dcStr = rdn.getValue().toString();
+                        }
+                        else
+                        {
+                            dcStr = rdn.getValue().toString() + '.' + dcStr;
+                        }
+                    }
+                }
+                return cnStr == null || cnStr.length()==0 ? "" : dcStr == null ? cnStr : cnStr + '@' + dcStr;
+            }
+            catch (InvalidNameException e)
+            {
+                log.warn("Invalid name: '"+dn+"'. ");
+                return "";
+            }
+        }
+    }
+
+
+    public static String retrieveIdentity(SSLEngine engine)
+    {
+        String id = "";
+        Certificate cert = engine.getSession().getLocalCertificates()[0];
+        Principal p = ((X509Certificate)cert).getSubjectDN();
+        String dn = p.getName();
         try
         {
-          Certificate cert = engine.getSession().getLocalCertificates()[0];
-          Principal p = ((X509Certificate)cert).getSubjectDN();
-          String dn = p.getName();
-                    
-          if (dn.contains("CN="))
-          {
-              String str = dn.substring(dn.indexOf("CN=")+3, dn.length());
-              id.append(str.substring(0,
-                      str.indexOf(",") == -1? str.length(): str.indexOf(",")));
-          } 
-          
-          if (dn.contains("DC="))
-          {
-              id.append("@");
-              int c = 0;
-              for (String toks : dn.split(","))
-              {
-                  if (toks.contains("DC"))
-                  {
-                     if (c > 0) {id.append(".");}                         
-                     id.append(toks.substring(
-                             toks.indexOf("=")+1, 
-                             toks.indexOf(",") == -1? toks.length(): toks.indexOf(",")));
-                     c++;
-                  }
-              }
-          }
+            id = SSLUtil.getIdFromSubjectDN(dn);
         }
-        catch(Exception e)
+        catch (Exception e)
         {
-            log.info("Exception received while trying to retrive client identity from SSL cert",e);
+            log.info("Exception received while trying to retrive client identity from SSL cert", e);
         }
-        
         log.debug("Extracted Identity from client certificate : " + id);
-        return id.toString();
+        return id;
     }
-    
+
     public static KeyStore getInitializedKeyStore(String storePath, String storePassword, String keyStoreType) throws GeneralSecurityException, IOException
     {
         KeyStore ks = KeyStore.getInstance(keyStoreType);
@@ -137,7 +158,7 @@ public class SSLUtil
             {
                 in = new FileInputStream(f);
             }
-            else 
+            else
             {
                 in = Thread.currentThread().getContextClassLoader().getResourceAsStream(storePath);
             }
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/url/AMQBindingURL.java b/qpid/java/common/src/main/java/org/apache/qpid/url/AMQBindingURL.java
index 3b9a0baab2..77902c3531 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/url/AMQBindingURL.java
+++ b/qpid/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;
@@ -56,7 +56,6 @@ public class AMQBindingURL implements BindingURL
     {
         BindingURLParser parser = new BindingURLParser();
         parser.parse(_url,this);
-        processOptions();
         _logger.debug("URL Parsed: " + this);
     }
 
@@ -80,10 +79,6 @@ public class AMQBindingURL implements BindingURL
         setExchangeName(new AMQShortString(exchangeName));
     }
 
-    private void processOptions() throws URISyntaxException
-    {
-    }
-
     public String getURL()
     {
         return _url;
@@ -98,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");
         }
@@ -152,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
             {
@@ -166,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();
@@ -196,7 +191,7 @@ public class AMQBindingURL implements BindingURL
 
     public String toString()
     {
-        StringBuffer sb = new StringBuffer();
+        StringBuilder sb = new StringBuilder();
 
         sb.append(_exchangeClass);
         sb.append("://");
@@ -212,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/qpid/java/common/src/main/java/org/apache/qpid/url/BindingURL.java b/qpid/java/common/src/main/java/org/apache/qpid/url/BindingURL.java
index 61585443b1..80a1ae540b 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/url/BindingURL.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/url/BindingURL.java
@@ -32,8 +32,6 @@ public interface BindingURL
     public static final String OPTION_AUTODELETE = "autodelete";
     public static final String OPTION_DURABLE = "durable";
     public static final String OPTION_BROWSE = "browse";
-    public static final String OPTION_CLIENTID = "clientid";
-    public static final String OPTION_SUBSCRIPTION = "subscription";
     public static final String OPTION_ROUTING_KEY = "routingkey";
     public static final String OPTION_BINDING_KEY = "bindingkey";
     public static final String OPTION_EXCHANGE_AUTODELETE = "exchangeautodelete";
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/url/BindingURLParser.java b/qpid/java/common/src/main/java/org/apache/qpid/url/BindingURLParser.java
index 939080e252..ee823978b4 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/url/BindingURLParser.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/url/BindingURLParser.java
@@ -1,5 +1,5 @@
 /*
- * 
+ *
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
@@ -7,16 +7,16 @@
  * 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
  * KIND, either express or implied.  See the License for the
  * specific language governing permissions and limitations
  * under the License.
- * 
+ *
  */
 package org.apache.qpid.url;
 
@@ -61,7 +61,8 @@ public class BindingURLParser
     }
 
     //<exch_class>://<exch_name>/[<destination>]/[<queue>]?<option>='<value>'[,<option>='<value>']*
-    public synchronized void parse(String url,AMQBindingURL bindingURL)  throws URISyntaxException
+    public synchronized void
+    parse(String url,AMQBindingURL bindingURL)  throws URISyntaxException
     {
         _url = (url + END_OF_URL_MARKER_CHAR).toCharArray();
         _bindingURL = bindingURL;
@@ -203,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;
@@ -418,31 +419,11 @@ public class BindingURLParser
         {
             throw new URISyntaxException(String.valueOf(_url),"It is illegal to specify both a routingKey and a bindingKey in the same URL",-1);
         }
-
-        // check for durable subscriptions
-        if (_bindingURL.getExchangeClass().equals(ExchangeDefaults.TOPIC_EXCHANGE_CLASS))
-        {
-            String queueName = null;
-            if (Boolean.parseBoolean(_bindingURL.getOption(BindingURL.OPTION_DURABLE)))
-            {
-                if (_bindingURL.containsOption(BindingURL.OPTION_CLIENTID) && _bindingURL.containsOption(BindingURL.OPTION_SUBSCRIPTION))
-                {
-                    queueName = _bindingURL.getOption(BindingURL.OPTION_CLIENTID) + ":" + _bindingURL.getOption(BindingURL.OPTION_SUBSCRIPTION);
-                }
-                else
-                {
-                    throw new URISyntaxException(String.valueOf(_url),"Durable subscription must have values for " + BindingURL.OPTION_CLIENTID
-                        + " and " + BindingURL.OPTION_SUBSCRIPTION , -1);
-
-                }
-            }
-            _bindingURL.setQueueName(queueName);
-        }
     }
 
     public static void main(String[] args)
     {
-        
+
         String[] urls = new String[]
            {
              "topic://amq.topic//myTopic?routingkey='stocks.#'",
diff --git a/qpid/java/common/src/test/java/org/apache/qpid/AMQExceptionTest.java b/qpid/java/common/src/test/java/org/apache/qpid/AMQExceptionTest.java
index 575058b7f8..9fceef337c 100644
--- a/qpid/java/common/src/test/java/org/apache/qpid/AMQExceptionTest.java
+++ b/qpid/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());
     }
 
diff --git a/qpid/java/common/src/test/java/org/apache/qpid/ssl/SSLContextFactoryTest.java b/qpid/java/common/src/test/java/org/apache/qpid/ssl/SSLContextFactoryTest.java
index 21b8871d9a..c5fa852f95 100644
--- a/qpid/java/common/src/test/java/org/apache/qpid/ssl/SSLContextFactoryTest.java
+++ b/qpid/java/common/src/test/java/org/apache/qpid/ssl/SSLContextFactoryTest.java
@@ -36,25 +36,7 @@ public class SSLContextFactoryTest extends QpidTestCase
     private static final String DEFAULT_TRUST_MANAGER_ALGORITHM = TrustManagerFactory.getDefaultAlgorithm();
     private static final String CERT_ALIAS_APP1 = "app1";
 
-    public void testBuildServerContext() throws Exception
-    {
-        SSLContext context = SSLContextFactory.buildServerContext(BROKER_KEYSTORE_PATH, STORE_PASSWORD, STORE_TYPE, DEFAULT_KEY_MANAGER_ALGORITHM);
-        assertNotNull("SSLContext should not be null", context);
-    }
 
-    public void testBuildServerContextWithIncorrectPassword() throws Exception
-    {
-        try
-        {
-            SSLContextFactory.buildServerContext(BROKER_KEYSTORE_PATH, "sajdklsad", STORE_TYPE, DEFAULT_KEY_MANAGER_ALGORITHM);
-            fail("Exception was not thrown due to incorrect password");
-        }
-        catch (IOException e)
-        {
-            //expected
-        }
-    }
-    
     public void testTrustStoreDoesNotExist() throws Exception
     {
         try
@@ -79,7 +61,7 @@ public class SSLContextFactoryTest extends QpidTestCase
         SSLContext context = SSLContextFactory.buildClientContext(CLIENT_TRUSTSTORE_PATH, STORE_PASSWORD, STORE_TYPE, DEFAULT_TRUST_MANAGER_ALGORITHM, CLIENT_KEYSTORE_PATH, STORE_PASSWORD, STORE_TYPE, DEFAULT_KEY_MANAGER_ALGORITHM, null);
         assertNotNull("SSLContext should not be null", context);
     }
-    
+
     public void testBuildClientContextWithForClientAuthWithCertAlias() throws Exception
     {
         SSLContext context = SSLContextFactory.buildClientContext(CLIENT_TRUSTSTORE_PATH, STORE_PASSWORD, STORE_TYPE, DEFAULT_TRUST_MANAGER_ALGORITHM, CLIENT_KEYSTORE_PATH, STORE_PASSWORD, STORE_TYPE, DEFAULT_KEY_MANAGER_ALGORITHM, CERT_ALIAS_APP1);
diff --git a/qpid/java/common/src/test/java/org/apache/qpid/transport/network/security/ssl/SSLUtilTest.java b/qpid/java/common/src/test/java/org/apache/qpid/transport/network/security/ssl/SSLUtilTest.java
new file mode 100644
index 0000000000..2d17f7a3c7
--- /dev/null
+++ b/qpid/java/common/src/test/java/org/apache/qpid/transport/network/security/ssl/SSLUtilTest.java
@@ -0,0 +1,52 @@
+/*
+ *
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * 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
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ *
+ */
+package org.apache.qpid.transport.network.security.ssl;
+
+import org.apache.qpid.test.utils.QpidTestCase;
+
+public class SSLUtilTest extends QpidTestCase
+{
+    public void testGetIdFromSubjectDN()
+    {
+        // "normal" dn
+        assertEquals("user@somewhere.example.org",SSLUtil.getIdFromSubjectDN("cn=user,dc=somewhere,dc=example,dc=org"));
+        // quoting of values, case of types, spacing all ignored
+        assertEquals("user2@somewhere.example.org",SSLUtil.getIdFromSubjectDN("DC=somewhere, dc=example,cn=\"user2\",dc=org"));
+        // only first cn is used
+        assertEquals("user@somewhere.example.org",SSLUtil.getIdFromSubjectDN("DC=somewhere, dc=example,cn=\"user\",dc=org, cn=user2"));
+        // no cn, no Id
+        assertEquals("",SSLUtil.getIdFromSubjectDN("DC=somewhere, dc=example,dc=org"));
+        // cn in value is ignored
+        assertEquals("",SSLUtil.getIdFromSubjectDN("C=CZ,O=Scholz,OU=\"JAKUB CN=USER1\""));
+        // cn with no dc gives just user
+        assertEquals("someone",SSLUtil.getIdFromSubjectDN("ou=someou, CN=\"someone\""));
+        // null results in empty string
+        assertEquals("",SSLUtil.getIdFromSubjectDN(null));
+        // invalid name results in empty string
+        assertEquals("",SSLUtil.getIdFromSubjectDN("ou=someou, ="));
+        // component containing whitespace
+        assertEquals("me@example.com",SSLUtil.getIdFromSubjectDN("CN=me,DC=example, DC=com, O=My Company Ltd, L=Newbury, ST=Berkshire, C=GB"));
+        // empty CN
+        assertEquals("",SSLUtil.getIdFromSubjectDN("CN=,DC=somewhere, dc=example,dc=org"));
+
+
+    }
+}