Fix TextMessage body thread safety

activemq-client/src/main/java/org/apache/activemq/command/ActiveMQTextMessage.java

@@ -45,20 +45,19 @@ public class ActiveMQTextMessage extends ActiveMQMessage implements TextMessage
 
     public static final byte DATA_STRUCTURE_TYPE = CommandTypes.ACTIVEMQ_TEXT_MESSAGE;
 
-    protected String text;
+    // This is package scope (instead of private) for testing purposes
+    volatile String text;
 
     @Override
     public Message copy() {
         ActiveMQTextMessage copy = new ActiveMQTextMessage();
-        copy(copy);
+        synchronized (this) {
+            super.copy(copy);
+            copy.text = text;
+        }
         return copy;
     }
 
-    private void copy(ActiveMQTextMessage copy) {
-        super.copy(copy);
-        copy.text = text;
-    }
-
     @Override
     public byte getDataStructureType() {
         return DATA_STRUCTURE_TYPE;
@@ -72,19 +71,35 @@ public String getJMSXMimeType() {
     @Override
     public void setText(String text) throws MessageNotWriteableException {
         checkReadOnlyBody();
-        this.text = text;
-        setContent(null);
+        synchronized (this) {
+            this.text = text;
+            setContent(null);
+        }
+    }
+
+    // Synchronize this to prevent setting content if another mutation operation
+    // is happening concurrently.
+    @Override
+    public synchronized void setContent(ByteSequence content) {
+        super.setContent(content);
     }
 
     @Override
     public String getText() throws JMSException {
-        ByteSequence content = getContent();
+        String text = this.text;
 
-        if (text == null && content != null) {
-            text = decodeContent(content);
-            setContent(null);
-            setCompressed(false);
+        if (text == null) {
+            synchronized (this) {
+                text = this.text;
+                // Double-checked locking, re-check under lock if we need to decode
+                if (text == null && content != null) {
+                    this.text = text = decodeContent(content);
+                    setContent(null);
+                    setCompressed(false);
+                }
+            }
         }
+
         return text;
     }
 
@@ -124,27 +139,36 @@ public void beforeMarshall(WireFormat wireFormat) throws IOException {
 
     @Override
     public void storeContentAndClear() {
-        storeContent();
-        text=null;
+        // always lock to simplify things because if this method is being called
+        // it's right before send so it's very likely to need to mutate state
+        // This should generally be uncontested lock so it will be fast
+        synchronized (this) {
+            storeContent();
+            text = null;
+        }
     }
 
     @Override
     public void storeContent() {
         try {
-            ByteSequence content = getContent();
-            String text = this.text;
-            if (content == null && text != null) {
-                ByteArrayOutputStream bytesOut = new ByteArrayOutputStream();
-                OutputStream os = bytesOut;
-                ActiveMQConnection connection = getConnection();
-                if (connection != null && connection.isUseCompression()) {
-                    compressed = true;
-                    os = new DeflaterOutputStream(os);
+            // Content is volatile so if it's not null we can skip and do nothing
+            if (content == null) {
+                synchronized (this) {
+                    // Double-checked locking, re-check state under lock
+                    if (content == null && text != null) {
+                        ByteArrayOutputStream bytesOut = new ByteArrayOutputStream();
+                        OutputStream os = bytesOut;
+                        ActiveMQConnection connection = getConnection();
+                        if (connection != null && connection.isUseCompression()) {
+                            compressed = true;
+                            os = new DeflaterOutputStream(os);
+                        }
+                        DataOutputStream dataOut = new DataOutputStream(os);
+                        MarshallingSupport.writeUTF8(dataOut, text);
+                        dataOut.close();
+                        setContent(bytesOut.toByteSequence());
+                    }
                 }
-                DataOutputStream dataOut = new DataOutputStream(os);
-                MarshallingSupport.writeUTF8(dataOut, text);
-                dataOut.close();
-                setContent(bytesOut.toByteSequence());
             }
         } catch (IOException e) {
             throw new RuntimeException(e);
@@ -156,11 +180,19 @@ public void storeContent() {
     @Override
     public void clearUnMarshalledState() throws JMSException {
         super.clearUnMarshalledState();
-        this.text = null;
+        if (this.text != null) {
+            synchronized (this) {
+                // This is volatile but another locking makes sure another thread
+                // isn't attempting to read this value to marshal to the content at the
+                // same time
+                this.text = null;
+            }
+        }
     }
 
+    // We need to sync because both variables need to be read independently
     @Override
-    public boolean isContentMarshalled() {
+    public synchronized boolean isContentMarshalled() {
         return content != null || text == null;
     }
 
@@ -177,19 +209,31 @@ public boolean isContentMarshalled() {
      */
     @Override
     public void clearBody() throws JMSException {
-        super.clearBody();
-        this.text = null;
+        synchronized (this) {
+            super.clearBody();
+            this.text = null;
+        }
     }
 
     @Override
     public int getSize() {
-        String text = this.text;
-        if (size == 0 && content == null && text != null) {
-            size = getMinimumMessageSize();
-            if (marshalledProperties != null) {
-                size += marshalledProperties.getLength();
+        int size = this.size;
+        if (size == 0) {
+            synchronized (this) {
+                size = this.size;
+                String text = this.text;
+                ByteSequence content = getContent();
+                if (size == 0 && content == null && text != null) {
+                    size = getMinimumMessageSize();
+                    ByteSequence marshalledProperties = this.marshalledProperties;
+                    if (marshalledProperties != null) {
+                        size += marshalledProperties.getLength();
+                    }
+                    size += text.length() * 2;
+                    this.size = size;
+                }
+                return super.getSize();
             }
-            size += text.length() * 2;
         }
         return super.getSize();
     }
@@ -198,7 +242,7 @@ public int getSize() {
     public String toString() {
         try {
             String text = this.text;
-            if( text == null ) {
+            if (text == null) {
                 text = decodeContent(getContent());
             }
             if (text != null) {

activemq-client/src/main/java/org/apache/activemq/command/Message.java

@@ -81,12 +81,12 @@ public abstract class Message extends BaseCommand implements MarshallAware, Mess
     protected boolean compressed;
     protected String userID;
 
-    protected ByteSequence content;
+    protected volatile ByteSequence content;
     protected volatile ByteSequence marshalledProperties;
     protected DataStructure dataStructure;
     protected int redeliveryCounter;
 
-    protected int size;
+    protected volatile int size;
     protected Map<String, Object> properties;
     protected boolean readOnlyProperties;
     protected boolean readOnlyBody;
@@ -715,14 +715,18 @@ public int decrementReferenceCount() {
     @Override
 	public int getSize() {
         int minimumMessageSize = getMinimumMessageSize();
+        ByteSequence content = this.content;
+        int size = this.size;
         if (size < minimumMessageSize || size == 0) {
             size = minimumMessageSize;
+            ByteSequence marshalledProperties = this.marshalledProperties;
             if (marshalledProperties != null) {
                 size += marshalledProperties.getLength();
             }
             if (content != null) {
                 size += content.getLength();
             }
+            this.size = size;
         }
         return size;
     }