diff --git a/.attach_pid78822 b/.attach_pid78822 new file mode 100644 index 00000000000..e69de29bb2d diff --git a/activemq-client/src/main/java/org/apache/activemq/command/ActiveMQTextMessage.java b/activemq-client/src/main/java/org/apache/activemq/command/ActiveMQTextMessage.java index a5df4164f0a..0ace802c8ce 100644 --- a/activemq-client/src/main/java/org/apache/activemq/command/ActiveMQTextMessage.java +++ b/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) { diff --git a/activemq-client/src/main/java/org/apache/activemq/command/Message.java b/activemq-client/src/main/java/org/apache/activemq/command/Message.java index 06e0aebeea6..11ccd306fa4 100644 --- a/activemq-client/src/main/java/org/apache/activemq/command/Message.java +++ b/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 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; }