Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

https://issues.apache.org/jira/browse/AMQ-4920 

And code to prevent concurrent writes to a message when dispatched to
multiple Topic consumers.
This commit is contained in:
Timothy Bish 2013-12-17 15:22:08 -05:00
parent 7c01c9b581
commit c387e842ee
1 changed files with 19 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
activemq-amqp/src/main/java/org/apache/activemq/transport/amqp/AmqpProtocolConverter.java
View File

@ -129,6 +129,7 @@ class AmqpProtocolConverter implements IAmqpProtocolConverter {
updateTracer();
}
@Override
public void updateTracer() {
if (amqpTransport.isTrace()) {
((TransportImpl) protonTransport).setProtocolTracer(new ProtocolTracer() {
@ -849,13 +850,27 @@ class AmqpProtocolConverter implements IAmqpProtocolConverter {
final MessageDispatch md = outbound.removeFirst();
try {
ActiveMQMessage temp = null;
if (md.getMessage() != null) {
org.apache.activemq.command.Message message = md.getMessage();
if (!message.getProperties().containsKey(MESSAGE_FORMAT_KEY)) {
message.setProperty(MESSAGE_FORMAT_KEY, 0);
// Topics can dispatch the same Message to more than one consumer
// so we must copy to prevent concurrent read / write to the same
// message object.
if (md.getDestination().isTopic()) {
synchronized (md.getMessage()) {
temp = (ActiveMQMessage) md.getMessage().copy();
}
} else {
temp = (ActiveMQMessage) md.getMessage();
}
if (!temp.getProperties().containsKey(MESSAGE_FORMAT_KEY)) {
temp.setProperty(MESSAGE_FORMAT_KEY, 0);
}
}
final ActiveMQMessage jms = (ActiveMQMessage) md.getMessage();
final ActiveMQMessage jms = temp;
if (jms == null) {
// It's the end of browse signal.
endOfBrowse = true;