add tests for exceeding frame size

Author: rade

test/src/com/rabbitmq/client/test/functional/FrameMax.java

@@ -21,39 +21,37 @@
 
 import java.io.IOException;
 import java.net.Socket;
+import java.util.concurrent.ExecutorService;
 
+import com.rabbitmq.client.Address;
+import com.rabbitmq.client.AMQP;
+import com.rabbitmq.client.Connection;
 import com.rabbitmq.client.ConnectionFactory;
 import com.rabbitmq.client.GetResponse;
+import com.rabbitmq.client.impl.AMQConnection;
 import com.rabbitmq.client.impl.Frame;
 import com.rabbitmq.client.impl.FrameHandler;
+import com.rabbitmq.client.impl.LongStringHelper;
 import com.rabbitmq.client.impl.SocketFrameHandler;
 
-/* Publish a message of size FRAME_MAX.  The broker should split this
- * into two frames before sending back. */
 public class FrameMax extends BrokerTestCase {
     /* This value for FrameMax is larger than the minimum and less
      * than what Rabbit suggests. */
     final static int FRAME_MAX = 70000;
     final static int REAL_FRAME_MAX = FRAME_MAX - 8;
 
-    private String queueName;
-
     public FrameMax() {
         connectionFactory = new MyConnectionFactory();
         connectionFactory.setRequestedFrameMax(FRAME_MAX);
     }
 
-    @Override
-    protected void createResources()
-        throws IOException
-    {
-        queueName = channel.queueDeclare().getQueue();
-    }
-
-    /* Frame content should be less or equal to frame-max - 8. */
+    /* Publish a message of size FRAME_MAX.  The broker should split
+     * this into two frames before sending back.  Frame content should
+     * be less or equal to frame-max - 8. */
     public void testFrameSizes()
         throws IOException, InterruptedException
     {
+        String queueName = channel.queueDeclare().getQueue();
         /* This should result in at least 3 frames. */
         int howMuch = 2*FRAME_MAX;
         basicPublishVolatile(new byte[howMuch], queueName);
@@ -69,6 +67,38 @@ public void testFrameSizes()
         }
     }
 
+    /* server should reject frames larger than AMQP.FRAME_MIN_SIZE
+     * during connection negotiation */
+    public void testRejectLargeFramesDuringConnectionNegotiation()
+        throws IOException
+    {
+        ConnectionFactory cf = new ConnectionFactory();
+        cf.getClientProperties().put("too_long", LongStringHelper.asLongString(new byte[AMQP.FRAME_MIN_SIZE]));
+        try {
+            cf.newConnection();
+            fail("Expected exception during connection negotiation");
+        } catch (IOException e) {
+        }
+    }
+
+    /* server should reject frames larger than the negotiated frame
+     * size */
+    public void testRejectExceedingFrameMax()
+        throws IOException
+    {
+        closeChannel();
+        closeConnection();
+        ConnectionFactory cf = new GenerousConnectionFactory();
+        connection = cf.newConnection();
+        openChannel();
+        try {
+            basicPublishVolatile(new byte[connection.getFrameMax()], "void");
+            channel.basicQos(0);
+            fail("Expected exception when publishing");
+        } catch (IOException e) {
+        }
+    }
+
     /* ConnectionFactory that uses MyFrameHandler rather than
      * SocketFrameHandler. */
     private static class MyConnectionFactory extends ConnectionFactory {
@@ -97,4 +127,53 @@ public Frame readFrame() throws IOException {
             return f;
         }
     }
+
+    /*
+      AMQConnection with a frame_max that is one higher than what it
+      tells the server.
+    */
+    private static class GenerousAMQConnection extends AMQConnection {
+
+        public GenerousAMQConnection(ConnectionFactory factory,
+                                     FrameHandler      handler,
+                                     ExecutorService   executor) {
+            super(factory.getUsername(),
+                  factory.getPassword(),
+                  handler,
+                  executor,
+                  factory.getVirtualHost(),
+                  factory.getClientProperties(),
+                  factory.getRequestedFrameMax(),
+                  factory.getRequestedChannelMax(),
+                  factory.getRequestedHeartbeat(),
+                  factory.getSaslConfig());
+        }
+
+        @Override public int getFrameMax() {
+            return super.getFrameMax() + 1;
+        }
+
+    }
+
+    private static class GenerousConnectionFactory extends ConnectionFactory {
+
+        @Override public Connection newConnection(ExecutorService executor, Address[] addrs)
+            throws IOException
+        {
+            IOException lastException = null;
+            for (Address addr : addrs) {
+                try {
+                    FrameHandler frameHandler = createFrameHandler(addr);
+                    AMQConnection conn = new GenerousAMQConnection(this, frameHandler, executor);
+                    conn.start();
+                    return conn;
+                } catch (IOException e) {
+                    lastException = e;
+                }
+            }
+            throw (lastException != null) ? lastException
+                : new IOException("failed to connect");
+        }
+    }
+
 }