Use only ByteBuffer for basic.publish

byte[] is wrapped into a ByteBuffer.

Author: acogoluegnes

pom.xml

@@ -798,6 +798,7 @@
               <include>src/test/java/com/rabbitmq/client/test/TestUtils.java</include>
               <include>src/test/java/com/rabbitmq/client/test/RpcTopologyRecordingTest.java</include>
               <include>src/test/java/com/rabbitmq/client/test/PublishWithByteBufferTest.java</include>
+              <include>src/test/java/com/rabbitmq/client/test/ByteBufferPublishTest.java</include>
             </includes>
             <googleJavaFormat>
               <version>${google-java-format.version}</version>

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

@@ -48,47 +48,26 @@ public class AMQCommand implements Command {
     private final CommandAssembler assembler;
     private final Lock assemblerLock = new ReentrantLock();
 
+    /** Construct a command for inbound frame assembly with a max body length. */
     AMQCommand(int maxBodyLength) {
-        this(null, null, null, maxBodyLength);
+        this.assembler = new CommandAssembler(null, null, maxBodyLength);
     }
 
-    /** Construct a command ready to fill in by reading frames */
+    /** Construct a command ready to fill in by reading frames. */
     public AMQCommand() {
-        this(null, null, null, Integer.MAX_VALUE);
+        this(Integer.MAX_VALUE);
     }
 
     /**
      * Construct a command with just a method, and without header or body.
      * @param method the wrapped method
      */
     public AMQCommand(com.rabbitmq.client.Method method) {
-        this(method, null, null, Integer.MAX_VALUE);
+        this.assembler = new CommandAssembler((Method) method, null, Integer.MAX_VALUE);
     }
 
     /**
-     * Construct a command with a specified method, header and body.
-     * @param method the wrapped method
-     * @param contentHeader the wrapped content header
-     * @param body the message body data
-     */
-    public AMQCommand(com.rabbitmq.client.Method method, AMQContentHeader contentHeader, byte[] body) {
-        this.assembler = new CommandAssembler((Method) method, contentHeader, body, Integer.MAX_VALUE);
-    }
-
-    /**
-     * Construct a command with a specified method, header and body.
-     * @param method the wrapped method
-     * @param contentHeader the wrapped content header
-     * @param body the message body data
-     * @param maxBodyLength the maximum size for an inbound message body
-     */
-    public AMQCommand(com.rabbitmq.client.Method method, AMQContentHeader contentHeader, byte[] body,
-                      int maxBodyLength) {
-        this.assembler = new CommandAssembler((Method) method, contentHeader, body, maxBodyLength);
-    }
-
-    /**
-     * Construct a command with a ByteBuffer body for zero-copy transmission.
+     * Construct a command with a ByteBuffer body for transmission.
      * @param method the wrapped method
      * @param contentHeader the wrapped content header
      * @param body the message body as a ByteBuffer
@@ -133,11 +112,29 @@ public void transmit(AMQChannel channel) throws IOException {
         try {
             Method m = this.assembler.getMethod();
             if (m.hasContent()) {
-                ByteBuffer bbBody = this.assembler.getByteBufferBody();
-                if (bbBody != null) {
-                    transmitWithByteBuffer(m, bbBody, channelNumber, connection);
-                } else {
-                    transmitWithByteArray(m, this.assembler.getContentBody(), channelNumber, connection);
+                ByteBuffer body = this.assembler.getByteBufferBody();
+                int bodySize = body == null ? 0 : body.remaining();
+                Frame headerFrame = this.assembler.getContentHeader().toFrame(channelNumber, bodySize);
+
+                int frameMax = connection.getFrameMax();
+                boolean cappedFrameMax = frameMax > 0;
+                int bodyPayloadMax = cappedFrameMax ? frameMax - EMPTY_FRAME_SIZE : bodySize;
+
+                if (cappedFrameMax && headerFrame.size() > frameMax) {
+                    String msg = String.format("Content headers exceeded max frame size: %d > %d", headerFrame.size(), frameMax);
+                    throw new IllegalArgumentException(msg);
+                }
+                connection.writeFrame(m.toFrame(channelNumber));
+                connection.writeFrame(headerFrame);
+
+                if (body != null) {
+                    int bodyPosition = body.position();
+                    for (int offset = 0; offset < bodySize; offset += bodyPayloadMax) {
+                        int remaining = bodySize - offset;
+                        int fragmentLength = (remaining < bodyPayloadMax) ? remaining : bodyPayloadMax;
+                        Frame frame = Frame.fromBodyFragment(channelNumber, body, bodyPosition + offset, fragmentLength);
+                        connection.writeFrame(frame);
+                    }
                 }
             } else {
                 connection.writeFrame(m.toFrame(channelNumber));
@@ -149,52 +146,6 @@ public void transmit(AMQChannel channel) throws IOException {
         connection.flush();
     }
 
-    private void transmitWithByteArray(Method m, byte[] body, int channelNumber, AMQConnection connection) throws IOException {
-        Frame headerFrame = this.assembler.getContentHeader().toFrame(channelNumber, body.length);
-
-        int frameMax = connection.getFrameMax();
-        boolean cappedFrameMax = frameMax > 0;
-        int bodyPayloadMax = cappedFrameMax ? frameMax - EMPTY_FRAME_SIZE : body.length;
-
-        if (cappedFrameMax && headerFrame.size() > frameMax) {
-            String msg = String.format("Content headers exceeded max frame size: %d > %d", headerFrame.size(), frameMax);
-            throw new IllegalArgumentException(msg);
-        }
-        connection.writeFrame(m.toFrame(channelNumber));
-        connection.writeFrame(headerFrame);
-
-        for (int offset = 0; offset < body.length; offset += bodyPayloadMax) {
-            int remaining = body.length - offset;
-            int fragmentLength = (remaining < bodyPayloadMax) ? remaining : bodyPayloadMax;
-            Frame frame = Frame.fromBodyFragment(channelNumber, body, offset, fragmentLength);
-            connection.writeFrame(frame);
-        }
-    }
-
-    private void transmitWithByteBuffer(Method m, ByteBuffer body, int channelNumber, AMQConnection connection) throws IOException {
-        int bodySize = body.remaining();
-        Frame headerFrame = this.assembler.getContentHeader().toFrame(channelNumber, bodySize);
-
-        int frameMax = connection.getFrameMax();
-        boolean cappedFrameMax = frameMax > 0;
-        int bodyPayloadMax = cappedFrameMax ? frameMax - EMPTY_FRAME_SIZE : bodySize;
-
-        if (cappedFrameMax && headerFrame.size() > frameMax) {
-            String msg = String.format("Content headers exceeded max frame size: %d > %d", headerFrame.size(), frameMax);
-            throw new IllegalArgumentException(msg);
-        }
-        connection.writeFrame(m.toFrame(channelNumber));
-        connection.writeFrame(headerFrame);
-
-        int bodyPosition = body.position();
-        for (int offset = 0; offset < bodySize; offset += bodyPayloadMax) {
-            int remaining = bodySize - offset;
-            int fragmentLength = (remaining < bodyPayloadMax) ? remaining : bodyPayloadMax;
-            Frame frame = Frame.fromBodyFragment(channelNumber, body, bodyPosition + offset, fragmentLength);
-            connection.writeFrame(frame);
-        }
-    }
-
     @Override public String toString() {
         return toString(false);
     }

src/main/java/com/rabbitmq/client/impl/ChannelN.java

@@ -716,34 +716,8 @@ public void basicPublish(String exchange, String routingKey,
                              BasicProperties props, byte[] body)
         throws IOException
     {
-        final long deliveryTag;
-        if (nextPublishSeqNo > 0) {
-            deliveryTag = getNextPublishSeqNo();
-            unconfirmedSet.add(deliveryTag);
-            nextPublishSeqNo++;
-        } else {
-            deliveryTag = 0;
-        }
-        if (props == null) {
-            props = MessageProperties.MINIMAL_BASIC;
-        }
-        AMQP.Basic.Publish publish = new Basic.Publish.Builder()
-                .exchange(exchange)
-                .routingKey(routingKey)
-                .mandatory(mandatory)
-                .immediate(immediate)
-                .build();
-        try {
-            ObservationCollector.PublishCall publishCall = properties -> {
-                AMQCommand command = new AMQCommand(publish, properties, body);
-                transmit(command);
-            };
-            observationCollector.publish(publishCall, publish, props, body, this.connectionInfo());
-        } catch (IOException | AlreadyClosedException e) {
-            metricsCollector.basicPublishFailure(this, e);
-            throw e;
-        }
-        metricsCollector.basicPublish(this, deliveryTag);
+        basicPublish(exchange, routingKey, mandatory, immediate, props,
+                body == null ? null : ByteBuffer.wrap(body));
     }
 
     /** Public API - {@inheritDoc} */

src/main/java/com/rabbitmq/client/impl/CommandAssembler.java

@@ -59,7 +59,7 @@ private enum CAState {
 
     private final int maxBodyLength;
 
-    public CommandAssembler(Method method, AMQContentHeader contentHeader, byte[] body,
+    public CommandAssembler(Method method, AMQContentHeader contentHeader,
                             int maxBodyLength) {
         this.method = method;
         this.contentHeader = contentHeader;
@@ -68,7 +68,6 @@ public CommandAssembler(Method method, AMQContentHeader contentHeader, byte[] bo
         this.byteBufferBody = null;
         this.remainingBodyBytes = 0;
         this.maxBodyLength = maxBodyLength;
-        appendBodyFragment(body);
         if (method == null) {
             this.state = CAState.EXPECTING_METHOD;
         } else if (contentHeader == null) {
@@ -183,7 +182,7 @@ public synchronized byte[] getContentBody() {
         return coalesceContentBody();
     }
 
-    public synchronized ByteBuffer getByteBufferBody() {
+    public ByteBuffer getByteBufferBody() {
         return this.byteBufferBody;
     }
 

src/main/java/com/rabbitmq/client/impl/Frame.java

@@ -45,7 +45,7 @@ public class Frame {
     /** Frame payload (for outbound frames) */
     private final ByteArrayOutputStream accumulator;
 
-    /** Frame payload as a ByteBuffer (for zero-copy outbound body frames) */
+    /** Frame payload as a ByteBuffer (for basic.publish body frames) */
     private final ByteBuffer byteBufferPayload;
 
     private static final int NON_BODY_SIZE = 1 /* type */ + 2 /* channel */ + 4 /* payload size */ + 1 /* end character */;
@@ -74,6 +74,12 @@ public Frame(int type, int channel, byte[] payload) {
         this.byteBufferPayload = null;
     }
 
+    /**
+     * For basic.publish frames.
+     * @param type
+     * @param channel
+     * @param byteBufferPayload
+     */
     private Frame(int type, int channel, ByteBuffer byteBufferPayload) {
         this.type = type;
         this.channel = channel;
@@ -82,17 +88,8 @@ private Frame(int type, int channel, ByteBuffer byteBufferPayload) {
         this.byteBufferPayload = byteBufferPayload;
     }
 
-    public static Frame fromBodyFragment(int channelNumber, byte[] body, int offset, int length)
-        throws IOException
-    {
-        Frame frame = new Frame(AMQP.FRAME_BODY, channelNumber);
-        DataOutputStream bodyOut = frame.getOutputStream();
-        bodyOut.write(body, offset, length);
-        return frame;
-    }
-
     /**
-     * Zero-copy body frame from a ByteBuffer slice. The buffer is not copied;
+     * Body frame from a ByteBuffer slice. The buffer is not copied;
      * the caller must not modify it after this call.
      */
     public static Frame fromBodyFragment(int channelNumber, ByteBuffer body, int offset, int length) {

src/main/java/com/rabbitmq/client/observation/NoOpObservationCollector.java

@@ -23,17 +23,6 @@
 
 final class NoOpObservationCollector implements ObservationCollector {
 
-  @Override
-  public void publish(
-      PublishCall call,
-      AMQP.Basic.Publish publish,
-      AMQP.BasicProperties properties,
-      byte[] body,
-      ConnectionInfo connectionInfo)
-      throws IOException {
-    call.publish(properties);
-  }
-
   @Override
   public void publish(
       PublishCall call,
@@ -45,17 +34,6 @@ public void publish(
     call.publish(properties);
   }
 
-  @Override
-  public void publish(
-      PublishCall call,
-      AMQP.Basic.Publish publish,
-      AMQP.BasicProperties properties,
-      int bodySize,
-      ConnectionInfo connectionInfo)
-      throws IOException {
-    call.publish(properties);
-  }
-
   @Override
   public Consumer basicConsume(String queue, String consumerTag, Consumer consumer) {
     return consumer;

src/main/java/com/rabbitmq/client/observation/ObservationCollector.java

@@ -41,9 +41,8 @@ public interface ObservationCollector {
   /**
    * Decorate message publishing.
    *
-   * <p>Implementations are expected to call {@link PublishCall#publish( PublishCall,
-   * AMQP.Basic.Publish, AMQP.BasicProperties, byte[], ConnectionInfo)} to make sure the message is
-   * actually sent.
+   * <p>Implementations are expected to call {@link PublishCall#publish(AMQP.BasicProperties)} to
+   * make sure the message is actually sent.
    *
    * @param call
    * @param publish
@@ -53,55 +52,12 @@ public interface ObservationCollector {
    * @throws IOException
    */
   void publish(
-      PublishCall call,
-      AMQP.Basic.Publish publish,
-      AMQP.BasicProperties properties,
-      byte[] body,
-      ConnectionInfo connectionInfo)
-      throws IOException;
-
-  /**
-   * Decorate message publishing with a ByteBuffer body.
-   *
-   * <p>The default implementation delegates to the byte[]-based overload using the body size.
-   * Implementations that need access to the actual body bytes should override this method.
-   *
-   * @param call
-   * @param publish
-   * @param properties
-   * @param body
-   * @param connectionInfo
-   * @throws IOException
-   */
-  default void publish(
       PublishCall call,
       AMQP.Basic.Publish publish,
       AMQP.BasicProperties properties,
       ByteBuffer body,
       ConnectionInfo connectionInfo)
-      throws IOException {
-    publish(call, publish, properties, body == null ? 0 : body.remaining(), connectionInfo);
-  }
-
-  /**
-   * Decorate message publishing using only the body size.
-   *
-   * @param call
-   * @param publish
-   * @param properties
-   * @param bodySize
-   * @param connectionInfo
-   * @throws IOException
-   */
-  default void publish(
-      PublishCall call,
-      AMQP.Basic.Publish publish,
-      AMQP.BasicProperties properties,
-      int bodySize,
-      ConnectionInfo connectionInfo)
-      throws IOException {
-    publish(call, publish, properties, new byte[0], connectionInfo);
-  }
+      throws IOException;
 
   /**
    * Decorate consumer registration.