AMQCommand is now thread-safe

Author: Steve Powell

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

@@ -28,7 +28,7 @@
  * method, header and body from a series of frames, unless these are
  * supplied at construction time.
  * <p/><b>Concurrency</b><br/>
- * This class is <i>not</i> thread-safe.
+ * This class is thread-safe.
  */
 public class AMQCommand implements Command {
 
@@ -38,13 +38,12 @@ public class AMQCommand implements Command {
      * <li>4 bytes of frame payload length</li>
      * <li>1 byte of payload trailer FRAME_END byte</li></ul>
      * See {@link #checkEmptyContentBodyFrameSize}, an assertion
-     * called at startup.
+     * checked at startup.
      */
     private static final int EMPTY_CONTENT_BODY_FRAME_SIZE = 8;
 
-    /** The current assembler for this command */
-    private CommandAssembler assembler;
-    
+    /** The assembler for this command - synchronised on - contains all the state */
+    private final CommandAssembler assembler;
 
     /** Construct a command ready to fill in by reading frames */
     public AMQCommand() {
@@ -71,7 +70,6 @@ public AMQCommand(com.rabbitmq.client.Method method, AMQContentHeader contentHea
 
     /** Public API - {@inheritDoc} */
     public Method getMethod() {
-        if (this.assembler == null) return null;
         return this.assembler.getMethod();
     }
 
@@ -85,9 +83,6 @@ public byte[] getContentBody() {
         return this.assembler.getContentBody();
     }
 
-    public boolean isComplete() {
-        return assembler.isComplete();
-    }
     public boolean handleFrame(Frame f) throws IOException {
         return this.assembler.handleFrame(f);
     }
@@ -102,23 +97,28 @@ public void transmit(AMQChannel channel) throws IOException {
         int channelNumber = channel.getChannelNumber();
         AMQConnection connection = channel.getConnection();
 
-        Method m = getMethod();
-        connection.writeFrame(m.toFrame(channelNumber));
-
-        if (m.hasContent()) {
-            byte[] body = getContentBody();
+        synchronized (assembler) {
+            Method m = this.assembler.getMethod();
+            connection.writeFrame(m.toFrame(channelNumber));
+            if (m.hasContent()) {
+                byte[] body = this.assembler.getContentBody();
 
-            connection.writeFrame(getContentHeader().toFrame(channelNumber, body.length));
+                connection.writeFrame(this.assembler.getContentHeader()
+                        .toFrame(channelNumber, body.length));
 
-            int frameMax = connection.getFrameMax();
-            int bodyPayloadMax = (frameMax == 0) ? body.length : frameMax - EMPTY_CONTENT_BODY_FRAME_SIZE;
+                int frameMax = connection.getFrameMax();
+                int bodyPayloadMax = (frameMax == 0) ? body.length : frameMax
+                        - EMPTY_CONTENT_BODY_FRAME_SIZE;
 
-            for (int offset = 0; offset < body.length; offset += bodyPayloadMax) {
-                int remaining = body.length - offset;
+                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);
+                    int fragmentLength = (remaining < bodyPayloadMax) ? remaining
+                            : bodyPayloadMax;
+                    Frame frame = Frame.fromBodyFragment(channelNumber, body,
+                            offset, fragmentLength);
+                    connection.writeFrame(frame);
+                }
             }
         }
     }
@@ -128,15 +128,29 @@ public void transmit(AMQChannel channel) throws IOException {
     }
 
     public String toString(boolean suppressBody){
-        byte[] body = getContentBody();
-        String contentStr;
+        synchronized (assembler) {
+            return new StringBuilder()
+                .append('{')
+                .append(this.assembler.getMethod())
+                .append(", ")
+                .append(this.assembler.getContentHeader())
+                .append(", ")
+                .append(contentBodyStringBuilder(
+                        this.assembler.getContentBody(), suppressBody))
+                .append('}').toString();
+        }
+    }
+
+    private static StringBuilder contentBodyStringBuilder(byte[] body, boolean suppressBody) {
         try {
-            contentStr = suppressBody ? (body.length + " bytes of payload") :
-                ("\"" + new String(body) + "\"");
+            if (suppressBody) {
+                return new StringBuilder().append(body.length).append(" bytes of payload");
+            } else {
+                return new StringBuilder().append('\"').append(body).append('\"');
+            }
         } catch (Exception e) {
-            contentStr = "|" + body.length + "|";
+            return new StringBuilder().append('|').append(body.length).append('|');
         }
-        return "{" + getMethod() + "," + getContentHeader() + "," + contentStr + "}";
     }
 
     /** Called to check internal code assumptions. */

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

@@ -26,105 +26,88 @@
 /**
  * Class responsible for piecing together a command from a series of {@link Frame}s.
  * <p/><b>Concurrency</b><br/>
- * This class is <i>not</i> thread-safe.
+ * This class is thread-safe, since all methods are synchronised. Callers should not
+ * synchronise on objects of this class unless they are sole owners.
  * @see AMQCommand
  */
 final class CommandAssembler {
     private static final byte[] EMPTY_BYTE_ARRAY = new byte[0];
 
     /** Current state, used to decide how to handle each incoming frame. */
-    private int state;
-    
-    private static final int STATE_EXPECTING_METHOD = 0;
-    private static final int STATE_EXPECTING_CONTENT_HEADER = 1;
-    private static final int STATE_EXPECTING_CONTENT_BODY = 2;
-    private static final int STATE_COMPLETE = 3;
+    private enum CAState {
+        EXPECTING_METHOD, EXPECTING_CONTENT_HEADER, EXPECTING_CONTENT_BODY, COMPLETE;
+    }
+    private CAState state;
 
     /** The method for this command */
-    private Method _method;
+    private Method method;
 
     /** The content header for this command */
-    private AMQContentHeader _contentHeader;
+    private AMQContentHeader contentHeader;
 
-    /** The first (and usually only) fragment of the content body */
-    private byte[] _body0;
-    /** The remaining fragments of this command's content body - a list of byte[] */
-    private List<byte[]> _bodyN;
+    /** The fragments of this command's content body - a list of byte[] */
+    private final List<byte[]> bodyN;
     /** sum of the lengths of all fragments */
     private int bodyLength;
 
-    /**
-     * How many more bytes of content body are expected to arrive
-     * from the broker.
-     */
+    /** No bytes of content body not yet accumulated */
     private long remainingBodyBytes;
 
-    public CommandAssembler() {
-        this(null, null, null);
-    }
-
     public CommandAssembler(Method method, AMQContentHeader contentHeader, byte[] body) {
-        this._method = method;
-        this._contentHeader = contentHeader;
-        setContentBody(body);
-        resetState();
-    }
-
-    private void resetState() {
+        this.method = method;
+        this.contentHeader = contentHeader;
+        this.bodyN = new ArrayList<byte[]>(2);
+        this.bodyLength = 0;
         this.remainingBodyBytes = 0;
-        if (this._method == null) {
-            this.state = STATE_EXPECTING_METHOD;
-        } else if (this._contentHeader == null) {
-            this.state = this._method.hasContent() ? STATE_EXPECTING_CONTENT_HEADER : STATE_COMPLETE;
+        appendBodyFragment(body);
+        if (method == null) {
+            this.state = CAState.EXPECTING_METHOD;
+        } else if (contentHeader == null) {
+            this.state = method.hasContent() ? CAState.EXPECTING_CONTENT_HEADER : CAState.COMPLETE;
         } else {
-            this.remainingBodyBytes = this._contentHeader.getBodySize() - contentBodySize();
+            this.remainingBodyBytes = contentHeader.getBodySize() - this.bodyLength;
+            updateContentBodyState();
         }
     }
 
-    private int contentBodySize() {
-        return this.bodyLength;
-    }
-
-    public Method getMethod() {
-        return _method;
+    public synchronized Method getMethod() {
+        return this.method;
     }
 
-    public AMQContentHeader getContentHeader() {
-        return _contentHeader;
+    public synchronized AMQContentHeader getContentHeader() {
+        return this.contentHeader;
     }
 
     /** @return true if the command is complete */
-    public boolean isComplete() {
-        return (this.state == STATE_COMPLETE);
+    public synchronized boolean isComplete() {
+        return (this.state == CAState.COMPLETE);
     }
 
     /** Decides whether more body frames are expected */
     private void updateContentBodyState() {
-        this.state = (this.remainingBodyBytes > 0) ? STATE_EXPECTING_CONTENT_BODY : STATE_COMPLETE;
+        this.state = (this.remainingBodyBytes > 0) ? CAState.EXPECTING_CONTENT_BODY : CAState.COMPLETE;
     }
 
-    private boolean consumeMethodFrame(Frame f) throws IOException {
+    private void consumeMethodFrame(Frame f) throws IOException {
         if (f.type == AMQP.FRAME_METHOD) {
-            this._method = AMQImpl.readMethodFrom(f.getInputStream());
-            state = this._method.hasContent() ? STATE_EXPECTING_CONTENT_HEADER : STATE_COMPLETE;
-            return isComplete();
+            this.method = AMQImpl.readMethodFrom(f.getInputStream());
+            this.state = this.method.hasContent() ? CAState.EXPECTING_CONTENT_HEADER : CAState.COMPLETE;
         } else {
             throw new UnexpectedFrameError(f, AMQP.FRAME_METHOD);
         }
     }
 
-    private boolean consumeHeaderFrame(Frame f) throws IOException {
+    private void consumeHeaderFrame(Frame f) throws IOException {
         if (f.type == AMQP.FRAME_HEADER) {
-            this._contentHeader = AMQImpl.readContentHeaderFrom(f.getInputStream());
-            this.remainingBodyBytes = this._contentHeader.getBodySize();
+            this.contentHeader = AMQImpl.readContentHeaderFrom(f.getInputStream());
+            this.remainingBodyBytes = this.contentHeader.getBodySize();
             updateContentBodyState();
-            return isComplete();
         } else {
             throw new UnexpectedFrameError(f, AMQP.FRAME_HEADER);
         }
     }
 
-    private boolean consumeBodyFrame(Frame f) {
+    private void consumeBodyFrame(Frame f) {
         if (f.type == AMQP.FRAME_BODY) {
             byte[] fragment = f.getPayload();
             this.remainingBodyBytes -= fragment.length;
@@ -133,81 +116,56 @@ private boolean consumeBodyFrame(Frame f) {
                 throw new UnsupportedOperationException("%%%%%% FIXME unimplemented");
             }
             appendBodyFragment(fragment);
-            return isComplete();
         } else {
             throw new UnexpectedFrameError(f, AMQP.FRAME_BODY);
         }
     }
 
     /** Stitches together a fragmented content body into a single byte array */
-    private void coalesceContentBody() {
-        List<byte[]> oldFragments = _bodyN;
-        byte[] firstFragment = _body0 == null ? EMPTY_BYTE_ARRAY : _body0;
+    private byte[] coalesceContentBody() {
+        if (this.bodyLength == 0) return EMPTY_BYTE_ARRAY;
+        if (this.bodyN.size() == 1) return this.bodyN.get(0);
 
         byte[] body = new byte[bodyLength];
-        System.arraycopy(firstFragment, 0, body, 0, firstFragment.length);
-        int offset = firstFragment.length;
-        for (byte[] fragment : oldFragments) {
+        int offset = 0;
+        for (byte[] fragment : this.bodyN) {
             System.arraycopy(fragment, 0, body, offset, fragment.length);
             offset += fragment.length;
         }
-
-        setContentBody(body);
-    }
-
-    public byte[] getContentBody() {
-        if (_bodyN != null) {
-            coalesceContentBody();
-        }
-        return _body0 == null ? EMPTY_BYTE_ARRAY : _body0;
+        this.bodyN.clear();
+        this.bodyN.add(body);
+        return body;
     }
 
-    /**
-     * Set the Command's content body.
-     * @param body the body data
-     */
-    private void setContentBody(byte[] body) {
-        _body0 = body;
-        _bodyN = null;
-        bodyLength = byteArrayLength(_body0);
+    public synchronized byte[] getContentBody() {
+        return coalesceContentBody();
     }
 
     private static int byteArrayLength(byte[] ba) {
         return (ba == null) ? 0 : ba.length;
     }
 
     private void appendBodyFragment(byte[] fragment) {
-        if (_body0 == null) {
-            _body0 = fragment;
-            bodyLength = byteArrayLength(fragment);
-        } else {
-            if (_bodyN == null) {
-                _bodyN = new ArrayList<byte[]>();
-            }
-            _bodyN.add(fragment);
-            bodyLength += byteArrayLength(fragment);
-        }
+        if (fragment == null || fragment.length == 0) return;
+        bodyN.add(fragment);
+        bodyLength += byteArrayLength(fragment);
     }
 
     /**
      * @param f frame to be incorporated
-     * @return true if command is complete
-     * @throws IOException if error reading in frame
+     * @return true if command becomes complete
+     * @throws IOException if error reading frame
      */
-    public boolean handleFrame(Frame f) throws IOException
+    public synchronized boolean handleFrame(Frame f) throws IOException
     {
         switch (this.state) {
-          case STATE_EXPECTING_METHOD:
-              return consumeMethodFrame(f);
-
-          case STATE_EXPECTING_CONTENT_HEADER:
-              return consumeHeaderFrame(f);
-
-          case STATE_EXPECTING_CONTENT_BODY:
-              return consumeBodyFrame(f);
+          case EXPECTING_METHOD:          consumeMethodFrame(f); break;
+          case EXPECTING_CONTENT_HEADER:  consumeHeaderFrame(f); break;
+          case EXPECTING_CONTENT_BODY:    consumeBodyFrame(f);   break;
 
           default:
               throw new AssertionError("Bad Command State " + this.state);
         }
+        return isComplete();
     }
 }