Merge bug 24991

Author: Simon MacMullen

src/com/rabbitmq/client/impl/AMQCommand.java

@@ -32,15 +32,15 @@
  */
 public class AMQCommand implements Command {
 
-    /** EMPTY_CONTENT_BODY_FRAME_SIZE = 8 = 1 + 2 + 4 + 1
+    /** EMPTY_FRAME_SIZE = 8 = 1 + 2 + 4 + 1
      * <ul><li>1 byte of frame type</li>
      * <li>2 bytes of channel number</li>
      * <li>4 bytes of frame payload length</li>
      * <li>1 byte of payload trailer FRAME_END byte</li></ul>
-     * See {@link #checkEmptyContentBodyFrameSize}, an assertion
-     * checked at startup.
+     * See {@link #checkEmptyFrameSize}, an assertion checked at
+     * startup.
      */
-    private static final int EMPTY_CONTENT_BODY_FRAME_SIZE = 8;
+    public static final int EMPTY_FRAME_SIZE = 8;
 
     /** The assembler for this command - synchronised on - contains all the state */
     private final CommandAssembler assembler;
@@ -108,7 +108,7 @@ public void transmit(AMQChannel channel) throws IOException {
 
                 int frameMax = connection.getFrameMax();
                 int bodyPayloadMax = (frameMax == 0) ? body.length : frameMax
-                        - EMPTY_CONTENT_BODY_FRAME_SIZE;
+                        - EMPTY_FRAME_SIZE;
 
                 for (int offset = 0; offset < body.length; offset += bodyPayloadMax) {
                     int remaining = body.length - offset;
@@ -157,26 +157,26 @@ private static StringBuilder contentBodyStringBuilder(byte[] body, boolean suppr
 
     /** Called to check internal code assumptions. */
     public static void checkPreconditions() {
-        checkEmptyContentBodyFrameSize();
+        checkEmptyFrameSize();
     }
 
     /**
-     * Since we're using a pre-computed value for
-     * EMPTY_CONTENT_BODY_FRAME_SIZE we check this is
-     * actually correct when run against the framing code in Frame.
+     * Since we're using a pre-computed value for EMPTY_FRAME_SIZE we
+     * check this is actually correct when run against the framing
+     * code in Frame.
      */
-    private static void checkEmptyContentBodyFrameSize() {
+    private static void checkEmptyFrameSize() {
         Frame f = new Frame(AMQP.FRAME_BODY, 0, new byte[0]);
         ByteArrayOutputStream s = new ByteArrayOutputStream();
         try {
             f.writeTo(new DataOutputStream(s));
         } catch (IOException ioe) {
-            throw new AssertionError("IOException while checking EMPTY_CONTENT_BODY_FRAME_SIZE");
+            throw new AssertionError("IOException while checking EMPTY_FRAME_SIZE");
         }
         int actualLength = s.toByteArray().length;
-        if (EMPTY_CONTENT_BODY_FRAME_SIZE != actualLength) {
-            throw new AssertionError("Internal error: expected EMPTY_CONTENT_BODY_FRAME_SIZE("
-                    + EMPTY_CONTENT_BODY_FRAME_SIZE
+        if (EMPTY_FRAME_SIZE != actualLength) {
+            throw new AssertionError("Internal error: expected EMPTY_FRAME_SIZE("
+                    + EMPTY_FRAME_SIZE
                     + ") is not equal to computed value: " + actualLength);
         }
     }

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

@@ -21,39 +21,38 @@
 
 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.AMQCommand;
 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 +68,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 +128,55 @@ 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() {
+            // the RabbitMQ broker permits frames that are oversize by
+            // up to EMPTY_FRAME_SIZE octets
+            return super.getFrameMax() + AMQCommand.EMPTY_FRAME_SIZE + 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");
+        }
+    }
+
 }