HORNETQ-1242 :

Allowing the topology message to be received correctly Changing the client version so it can connect to the older server by allowing a list of compatible server versions. Filling the fields that are not present in HornetQ 2.2.x and thus won't get filled during deserialization. Making them not final to be able to do this.
FlexiGroup · Dec 18, 2013 · 9c4e9d8 · 9c4e9d8
1 parent cee17bf
commit 9c4e9d8
Show file tree

Hide file tree

Showing 6 changed files with 95 additions and 27 deletions.
diff --git a/hornetq-core-client/src/main/java/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java b/hornetq-core-client/src/main/java/org/hornetq/core/client/impl/ClientSessionFactoryImpl.java
@@ -767,14 +767,47 @@ private void failoverOrReconnect(final Object connectionID, final HornetQExcepti
          }
       }
    }
-
+   
    private ClientSession createSessionInternal(final String username,
                                                final String password,
                                                final boolean xa,
                                                final boolean autoCommitSends,
                                                final boolean autoCommitAcks,
                                                final boolean preAcknowledge,
                                                final int ackBatchSize) throws HornetQException
+   {
+       for(Version clientVersion : VersionLoader.getClientVersions())
+       {
+           try
+           {
+               return createSessionInternal(clientVersion, 
+                                            username, 
+                                            password, 
+                                            xa, 
+                                            autoCommitSends, 
+                                            autoCommitAcks, 
+                                            preAcknowledge, 
+                                            ackBatchSize);
+           } catch(HornetQException e)
+           {
+               if (e.getType() != HornetQExceptionType.INCOMPATIBLE_CLIENT_SERVER_VERSIONS)
+               {
+                   throw e;
+               }
+           }
+       }
+       connection.destroy();
+       throw new HornetQException(HornetQExceptionType.INCOMPATIBLE_CLIENT_SERVER_VERSIONS);
+   }
+
+   private ClientSession createSessionInternal(final Version clientVersion, 
+                                               final String username,
+                                               final String password,
+                                               final boolean xa,
+                                               final boolean autoCommitSends,
+                                               final boolean autoCommitAcks,
+                                               final boolean preAcknowledge,
+                                               final int ackBatchSize) throws HornetQException
    {
       synchronized (createSessionLock)
       {
@@ -785,8 +818,6 @@ private ClientSession createSessionInternal(final String username,
          boolean retry = false;
          do
          {
-            Version clientVersion = VersionLoader.getVersion();
-
             Lock lock = null;
 
             try

diff --git a/hornetq-core-client/src/main/java/org/hornetq/core/client/impl/ServerLocatorImpl.java b/hornetq-core-client/src/main/java/org/hornetq/core/client/impl/ServerLocatorImpl.java
@@ -102,8 +102,8 @@ private enum STATE
 
    private final Topology topology;
 
-   //needs to be serializable
-   private final String topologyArrayGuard = new String();
+   //needs to be serializable and not final for retrocompatibility
+   private String topologyArrayGuard = new String();
 
    private volatile Pair<TransportConfiguration, TransportConfiguration>[] topologyArray;
 
@@ -180,11 +180,8 @@ private enum STATE
 
    private int initialMessagePacketSize;
 
-   /**
-    * As the class is Serializable, the guard field must be of a serializable type in order to be
-    * final.
-    */
-   private final String stateGuard = new String();
+   //needs to be serializable and not final for retrocompatibility
+   private String stateGuard = new String();
    private transient STATE state;
    private transient CountDownLatch latch;
 
@@ -1788,9 +1785,18 @@ private void addFactory(ClientSessionFactoryInternal factory)
    private void readObject(ObjectInputStream is) throws ClassNotFoundException, IOException
    {
       is.defaultReadObject();
+      if(stateGuard == null)
+      {
+          stateGuard = new String();
+      }
+      if(topologyArrayGuard == null)
+      {
+          topologyArrayGuard = new String();
+      }
       //is transient so need to create, for compatibility issues
       packetDecoder = ClientPacketDecoder.INSTANCE;
    }
+
    private final class StaticConnector implements Serializable
    {
       private static final long serialVersionUID = 6772279632415242634l;

diff --git a/.../java/org/hornetq/core/protocol/core/impl/wireformat/ClusterTopologyChangeMessage_V2.java b/.../java/org/hornetq/core/protocol/core/impl/wireformat/ClusterTopologyChangeMessage_V2.java
@@ -137,7 +137,9 @@ public void decodeRest(final HornetQBuffer buffer)
          pair = new Pair<TransportConfiguration, TransportConfiguration>(a, b);
          last = buffer.readBoolean();
       }
-      nodeName = buffer.readNullableString();
+      if(buffer.readableBytes() > 0) {
+        nodeName = buffer.readNullableString();
+      }
    }
 
    @Override

diff --git a/hornetq-core-client/src/main/java/org/hornetq/utils/VersionLoader.java b/hornetq-core-client/src/main/java/org/hornetq/utils/VersionLoader.java
@@ -19,6 +19,10 @@
 import java.net.URLClassLoader;
 import java.security.AccessController;
 import java.security.PrivilegedAction;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Comparator;
+import java.util.List;
 import java.util.Properties;
 import java.util.StringTokenizer;
 
@@ -41,7 +45,7 @@ public final class VersionLoader
 
    private static String PROP_FILE_NAME;
 
-   private static Version version;
+   private static Version[] versions;
 
    static
    {
@@ -69,24 +73,34 @@ public String run()
             PROP_FILE_NAME = VersionLoader.DEFAULT_PROP_FILE_NAME;
          }
 
-         VersionLoader.version = VersionLoader.load();
+         VersionLoader.versions = VersionLoader.load();
       }
       catch (Throwable e)
       {
-         VersionLoader.version = null;
+         VersionLoader.versions = null;
          HornetQClientLogger.LOGGER.error(e.getMessage(), e);
       }
 
    }
+
+   public static Version[] getClientVersions()
+   {
+      if (VersionLoader.versions == null)
+      {
+         throw new RuntimeException(VersionLoader.PROP_FILE_NAME + " is not available");
+      }
+
+      return VersionLoader.versions;
+   }
 
    public static Version getVersion()
    {
-      if (VersionLoader.version == null)
+      if (VersionLoader.versions == null)
       {
          throw new RuntimeException(VersionLoader.PROP_FILE_NAME + " is not available");
       }
 
-      return VersionLoader.version;
+      return VersionLoader.versions[0];
    }
 
    public static String getClasspathString() {
@@ -101,7 +115,7 @@ public static String getClasspathString() {
        return classpath.toString();
    }
 
-   private static Version load()
+   private static Version[] load()
    {
       Properties versionProps = new Properties();
       final InputStream in = VersionImpl.class.getClassLoader().getResourceAsStream(VersionLoader.PROP_FILE_NAME);
@@ -119,17 +133,31 @@ private static Version load()
             int majorVersion = Integer.valueOf(versionProps.getProperty("hornetq.version.majorVersion"));
             int minorVersion = Integer.valueOf(versionProps.getProperty("hornetq.version.minorVersion"));
             int microVersion = Integer.valueOf(versionProps.getProperty("hornetq.version.microVersion"));
-            int incrementingVersion = Integer.valueOf(versionProps.getProperty("hornetq.version.incrementingVersion"));
+            int[] incrementingVersions = parseCompatibleVersionList(versionProps.getProperty("hornetq.version.incrementingVersion"));
             String versionSuffix = versionProps.getProperty("hornetq.version.versionSuffix");
             int[] compatibleVersionArray = parseCompatibleVersionList(versionProps.getProperty("hornetq.version.compatibleVersionList"));
-
-            return new VersionImpl(versionName,
+            List<Version> definedVersions = new ArrayList<Version>(incrementingVersions.length);
+            for(int incrementingVersion : incrementingVersions)
+            {
+                definedVersions.add(new VersionImpl(versionName,
                                    majorVersion,
                                    minorVersion,
                                    microVersion,
                                    incrementingVersion,
                                    versionSuffix,
-                                   compatibleVersionArray);
+                                   compatibleVersionArray));
+            }
+            //We want the higher version to be the first
+            Collections.sort(definedVersions, new Comparator<Version>() 
+            {
+                @Override
+                public int compare(Version version1, Version version2) 
+                {
+                    return version2.getIncrementingVersion() - version1.getIncrementingVersion();
+                }
+
+            });
+            return definedVersions.toArray(new Version[incrementingVersions.length]);            
          }
          catch (IOException e)
          {

diff --git a/pom.xml b/pom.xml
@@ -38,7 +38,7 @@
       <hornetq.version.majorVersion>2</hornetq.version.majorVersion>
       <hornetq.version.minorVersion>5</hornetq.version.minorVersion>
       <hornetq.version.microVersion>0</hornetq.version.microVersion>
-      <hornetq.version.incrementingVersion>124</hornetq.version.incrementingVersion>
+      <hornetq.version.incrementingVersion>124,123,122</hornetq.version.incrementingVersion>
       <hornetq.version.versionSuffix>SNAPSHOT</hornetq.version.versionSuffix>
       <hornetq.version.versionTag>SNAPSHOT</hornetq.version.versionTag>
       <HornetQ-Version>

diff --git a/tests/unit-tests/src/test/java/org/hornetq/tests/unit/util/VersionLoaderTest.java b/tests/unit-tests/src/test/java/org/hornetq/tests/unit/util/VersionLoaderTest.java
@@ -16,6 +16,7 @@
 import org.junit.Test;
 
 import java.util.Properties;
+import java.util.StringTokenizer;
 
 import org.junit.Assert;
 
@@ -50,13 +51,13 @@ public void testLoadVersion() throws Exception
       Assert.assertEquals(props.get("hornetq.version.versionName"), version.getVersionName());
       Assert.assertEquals(props.get("hornetq.version.versionSuffix"), version.getVersionSuffix());
 
-      Assert.assertEquals(Integer.parseInt((String)props.get("hornetq.version.majorVersion")),
+      Assert.assertEquals(Integer.parseInt(props.getProperty("hornetq.version.majorVersion")),
                           version.getMajorVersion());
-      Assert.assertEquals(Integer.parseInt((String)props.get("hornetq.version.minorVersion")),
+      Assert.assertEquals(Integer.parseInt(props.getProperty("hornetq.version.minorVersion")),
                           version.getMinorVersion());
-      Assert.assertEquals(Integer.parseInt((String)props.get("hornetq.version.microVersion")),
-                          version.getMicroVersion());
-      Assert.assertEquals(Integer.parseInt((String)props.get("hornetq.version.incrementingVersion")),
+      Assert.assertEquals(Integer.parseInt(props.getProperty("hornetq.version.microVersion")),
+                          version.getMicroVersion());           
+      Assert.assertEquals(Integer.parseInt(new StringTokenizer(props.getProperty("hornetq.version.incrementingVersion"), ",").nextToken()),
                           version.getIncrementingVersion());
    }