QPID-5978 : [Java Client] deal with protocol initialisation failures such as unsupported versions, or attempting to connect to an HTTP port

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

3 files changed, 23 insertions, 11 deletions

diff --git a/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ClientMethodDispatcherImpl.java b/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ClientMethodDispatcherImpl.java
index e1a0e18262..d76fdf25e6 100644
--- a/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ClientMethodDispatcherImpl.java
+++ b/qpid/java/client/src/main/java/org/apache/qpid/client/handler/ClientMethodDispatcherImpl.java
@@ -20,6 +20,9 @@
  */
 package org.apache.qpid.client.handler;
 
+import java.util.HashMap;
+import java.util.Map;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -29,9 +32,6 @@ import org.apache.qpid.client.state.AMQMethodNotImplementedException;
 import org.apache.qpid.client.state.AMQStateManager;
 import org.apache.qpid.framing.*;
 
-import java.util.HashMap;
-import java.util.Map;
-
 public class ClientMethodDispatcherImpl implements MethodDispatcher
 {
 
@@ -101,6 +101,10 @@ public class ClientMethodDispatcherImpl implements MethodDispatcher
         }
         
         DispatcherFactory factory = _dispatcherFactories.get(version);
+        if(factory == null)
+        {
+            throw new UnsupportedOperationException("The protocol version " + version + " is not supported");
+        }
         return factory.createMethodDispatcher(session);
     }
 
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/codec/AMQDecoder.java b/qpid/java/common/src/main/java/org/apache/qpid/codec/AMQDecoder.java
index 3311401ad5..ebecb7b483 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/codec/AMQDecoder.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/codec/AMQDecoder.java
@@ -66,6 +66,8 @@ public class AMQDecoder
 
     private AMQMethodBodyFactory _bodyFactory;
 
+    private boolean _firstRead = true;
+
     private List<ByteArrayInputStream> _remainingBufs = new ArrayList<ByteArrayInputStream>();
 
     /**
@@ -239,6 +241,17 @@ public class AMQDecoder
             msg = new ByteArrayDataInput(buf.array(),buf.arrayOffset()+buf.position(), buf.remaining());
         }
 
+        // If this is the first read then we may be getting a protocol initiation back if we tried to negotiate
+        // an unsupported version
+        if(_firstRead && buf.hasRemaining())
+        {
+            _firstRead = false;
+            if(!_expectProtocolInitiation && buf.get(buf.position()) > 8)
+            {
+                _expectProtocolInitiation = true;
+            }
+        }
+
         boolean enoughData = true;
         while (enoughData)
         {
diff --git a/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQProtocolHeaderException.java b/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQProtocolHeaderException.java
index b0c92d9aab..b55a48067d 100644
--- a/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQProtocolHeaderException.java
+++ b/qpid/java/common/src/main/java/org/apache/qpid/framing/AMQProtocolHeaderException.java
@@ -20,17 +20,12 @@
  */
 package org.apache.qpid.framing;
 
-import org.apache.qpid.AMQException;
+import org.apache.qpid.transport.TransportException;
 
-/**
- * AMQProtocolHeaderException indicates a format error in an AMQP frame header.
- * <p>
- * TODO Not an AMQP exception as no status code.
- */
-public class AMQProtocolHeaderException extends AMQException
+public class AMQProtocolHeaderException extends TransportException
 {
     public AMQProtocolHeaderException(String message, Throwable cause)
     {
-        super(null, message, cause);
+        super(message, cause);
     }
 }