QPID-6424 : Implement Connection.Redirect in 0-8/9/9-1

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

5 files changed, 65 insertions, 9 deletions

diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/ClientDelegate.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/ClientDelegate.java
index 1866e1fd15..f8eabef161 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/ClientDelegate.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/ClientDelegate.java
@@ -183,7 +183,15 @@ public class ClientDelegate extends ConnectionDelegate
     @Override
     public void connectionRedirect(Connection conn, ConnectionRedirect redir)
     {
-        throw new UnsupportedOperationException();
+        conn.setRedirecting(true);
+        conn.getSender().close();
+        for(ConnectionListener listener : conn.getListeners())
+        {
+            if(listener.redirect(redir.getHost(), redir.getKnownHosts()))
+            {
+                break;
+            }
+        }
     }
 
     @Override
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java
index e63949cc69..331f96d6da 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/Connection.java
@@ -81,6 +81,7 @@ public class Connection extends ConnectionInvoker
     private NetworkConnection _networkConnection;
     private FrameSizeObserver _frameSizeObserver;
     private boolean _messageCompressionSupported;
+    private final AtomicBoolean _redirecting = new AtomicBoolean();
 
     public enum State { NEW, CLOSED, OPENING, OPEN, CLOSING, CLOSE_RCVD, RESUMING }
 
@@ -92,6 +93,12 @@ public class Connection extends ConnectionInvoker
             log.error(exception, "connection exception");
         }
         public void closed(Connection conn) {}
+
+        @Override
+        public boolean redirect(final String host, final List<Object> knownHosts)
+        {
+            return false;
+        }
     }
 
     public static interface SessionFactory
@@ -151,6 +158,11 @@ public class Connection extends ConnectionInvoker
         listeners.add(listener);
     }
 
+    public List<ConnectionListener> getListeners()
+    {
+        return Collections.unmodifiableList(listeners);
+    }
+
     public Sender<ProtocolEvent> getSender()
     {
         return sender;
@@ -226,6 +238,7 @@ public class Connection extends ConnectionInvoker
         synchronized (lock)
         {
             conSettings = settings;
+            _redirecting.set(false);
             state = OPENING;
             userID = settings.getUsername();
             connectionLost.set(false);
@@ -259,7 +272,7 @@ public class Connection extends ConnectionInvoker
             send(new ProtocolHeader(1, 0, 10));
 
             Waiter w = new Waiter(lock, timeout);
-            while (w.hasTime() && state == OPENING && error == null)
+            while (w.hasTime() && ((state == OPENING && error == null) || isRedirecting()))
             {
                 w.await();
             }
@@ -863,4 +876,15 @@ public class Connection extends ConnectionInvoker
     {
         return _messageCompressionSupported;
     }
+
+    public boolean isRedirecting()
+    {
+        return _redirecting.get();
+    }
+
+    public void setRedirecting(final boolean redirecting)
+    {
+        _redirecting.set(redirecting);
+    }
+
 }
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/ConnectionListener.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/ConnectionListener.java
index 616e76825a..b055b9f5e1 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/ConnectionListener.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/ConnectionListener.java
@@ -21,6 +21,8 @@
 package org.apache.qpid.transport;
 
 
+import java.util.List;
+
 /**
  * ConnectionListener
  *
@@ -35,4 +37,5 @@ public interface ConnectionListener
 
     void closed(Connection connection);
 
+    boolean redirect(String host, List<Object> knownHosts);
 }
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/sasl/SASLEncryptor.java b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/sasl/SASLEncryptor.java
index 2a70087c10..5ef94b7d13 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/sasl/SASLEncryptor.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/transport/network/security/sasl/SASLEncryptor.java
@@ -21,6 +21,8 @@
 package org.apache.qpid.transport.network.security.sasl;
 
 
+import java.util.List;
+
 import javax.security.sasl.Sasl;
 import javax.security.sasl.SaslClient;
 
@@ -56,7 +58,13 @@ public abstract class SASLEncryptor implements ConnectionListener
             }
         }
     }
-    
+
+    @Override
+    public boolean redirect(final String host, final List<Object> knownHosts)
+    {
+        return false;
+    }
+
     public void exception(Connection conn, ConnectionException exception){}
     public void closed(Connection conn) {}
     
diff --git a/qpid/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java b/qpid/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java
index 46d1887496..afc2b968f4 100644
--- a/qpid/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java
+++ b/qpid/java/common/src/test/java/org/apache/qpid/transport/ConnectionTest.java
@@ -20,12 +20,6 @@
  */
 package org.apache.qpid.transport;
 
-import org.apache.log4j.Logger;
-import org.apache.qpid.test.utils.QpidTestCase;
-import org.apache.qpid.transport.network.ConnectionBinding;
-import org.apache.qpid.transport.network.io.IoAcceptor;
-import org.apache.qpid.transport.util.Waiter;
-
 import static org.apache.qpid.transport.Option.EXPECTED;
 import static org.apache.qpid.transport.Option.NONE;
 import static org.apache.qpid.transport.Option.SYNC;
@@ -37,6 +31,13 @@ import java.util.List;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.TimeUnit;
 
+import org.apache.log4j.Logger;
+
+import org.apache.qpid.test.utils.QpidTestCase;
+import org.apache.qpid.transport.network.ConnectionBinding;
+import org.apache.qpid.transport.network.io.IoAcceptor;
+import org.apache.qpid.transport.util.Waiter;
+
 /**
  * ConnectionTest
  */
@@ -171,6 +172,12 @@ public class ConnectionTest extends QpidTestCase implements SessionListener
                     closed.countDown();
                 }
             }
+
+            @Override
+            public boolean redirect(final String host, final List<Object> knownHosts)
+            {
+                return false;
+            }
         });
         conn.connect("localhost", port, null, "guest", "guest", false, null);
         return conn;
@@ -437,6 +444,12 @@ public class ConnectionTest extends QpidTestCase implements SessionListener
             conn.connect("localhost", port, null, "guest", "guest", false, null);
             conn.resume();
         }
+
+        @Override
+        public boolean redirect(final String host, final List<Object> knownHosts)
+        {
+            return false;
+        }
     }
 
     class TestSessionListener implements SessionListener