mirror of https://github.com/apache/activemq.git
apply fix for: https://issues.apache.org/jira/browse/AMQ-3500
git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1183143 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
parent
4acd13243a
commit
88f60580e3
|
@ -16,6 +16,8 @@
|
||||||
*/
|
*/
|
||||||
package org.apache.activemq.camel.component;
|
package org.apache.activemq.camel.component;
|
||||||
|
|
||||||
|
import java.util.concurrent.TimeUnit;
|
||||||
|
|
||||||
import org.apache.camel.CamelContext;
|
import org.apache.camel.CamelContext;
|
||||||
import org.apache.camel.builder.RouteBuilder;
|
import org.apache.camel.builder.RouteBuilder;
|
||||||
|
|
||||||
|
@ -67,6 +69,36 @@ public class ActiveMQRouteTest extends CamelTestSupport {
|
||||||
return camelContext;
|
return camelContext;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testInvalidDestinationOptionOnConsumer() throws Exception {
|
||||||
|
getMockEndpoint("mock:result").expectedMessageCount(0);
|
||||||
|
assertMockEndpointsSatisfied(1, TimeUnit.SECONDS);
|
||||||
|
try {
|
||||||
|
new RouteBuilder() {
|
||||||
|
public void configure() throws Exception {
|
||||||
|
from("activemq:queue:foo?destination.consumer.exclusive=true&destination.consumer.unknown=foo")
|
||||||
|
.to("mock:result");
|
||||||
|
}
|
||||||
|
};
|
||||||
|
} catch (Exception e) {
|
||||||
|
fail("Should not have accepted bad destination options.");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
@Test
|
||||||
|
public void testInvalidDestinationOptionOnProducer() throws Exception {
|
||||||
|
try {
|
||||||
|
new RouteBuilder() {
|
||||||
|
public void configure() throws Exception {
|
||||||
|
from("activemq:queue:foo")
|
||||||
|
.to("activemq:queue:bar?destination.producer.exclusive=true");
|
||||||
|
}
|
||||||
|
};
|
||||||
|
} catch (Exception e) {
|
||||||
|
fail("Should not have accepted bad destination options.");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
protected RouteBuilder createRouteBuilder() throws Exception {
|
protected RouteBuilder createRouteBuilder() throws Exception {
|
||||||
return new RouteBuilder() {
|
return new RouteBuilder() {
|
||||||
public void configure() throws Exception {
|
public void configure() throws Exception {
|
||||||
|
|
|
@ -225,6 +225,15 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
|
||||||
if (dest.getOptions() != null) {
|
if (dest.getOptions() != null) {
|
||||||
Map<String, String> options = new HashMap<String, String>(dest.getOptions());
|
Map<String, String> options = new HashMap<String, String>(dest.getOptions());
|
||||||
IntrospectionSupport.setProperties(this.info, options, "consumer.");
|
IntrospectionSupport.setProperties(this.info, options, "consumer.");
|
||||||
|
if (options.size() > 0) {
|
||||||
|
String msg = "There are " + options.size()
|
||||||
|
+ " consumer options that couldn't be set on the consumer."
|
||||||
|
+ " Check the options are spelled correctly."
|
||||||
|
+ " Unknown parameters=[" + options + "]."
|
||||||
|
+ " This consumer cannot be started.";
|
||||||
|
LOG.warn(msg);
|
||||||
|
throw new ConfigurationException(msg);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
this.info.setDestination(dest);
|
this.info.setDestination(dest);
|
||||||
|
@ -459,11 +468,11 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
|
||||||
if (timeout > 0 && !unconsumedMessages.isClosed()) {
|
if (timeout > 0 && !unconsumedMessages.isClosed()) {
|
||||||
timeout = Math.max(deadline - System.currentTimeMillis(), 0);
|
timeout = Math.max(deadline - System.currentTimeMillis(), 0);
|
||||||
} else {
|
} else {
|
||||||
if (failureError != null) {
|
if (failureError != null) {
|
||||||
throw JMSExceptionSupport.create(failureError);
|
throw JMSExceptionSupport.create(failureError);
|
||||||
} else {
|
} else {
|
||||||
return null;
|
return null;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
} else if (md.getMessage() == null) {
|
} else if (md.getMessage() == null) {
|
||||||
return null;
|
return null;
|
||||||
|
@ -524,7 +533,7 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
|
||||||
private ActiveMQMessage createActiveMQMessage(final MessageDispatch md) throws JMSException {
|
private ActiveMQMessage createActiveMQMessage(final MessageDispatch md) throws JMSException {
|
||||||
ActiveMQMessage m = (ActiveMQMessage)md.getMessage().copy();
|
ActiveMQMessage m = (ActiveMQMessage)md.getMessage().copy();
|
||||||
if (m.getDataStructureType()==CommandTypes.ACTIVEMQ_BLOB_MESSAGE) {
|
if (m.getDataStructureType()==CommandTypes.ACTIVEMQ_BLOB_MESSAGE) {
|
||||||
((ActiveMQBlobMessage)m).setBlobDownloader(new BlobDownloader(session.getBlobTransferPolicy()));
|
((ActiveMQBlobMessage)m).setBlobDownloader(new BlobDownloader(session.getBlobTransferPolicy()));
|
||||||
}
|
}
|
||||||
if (transformer != null) {
|
if (transformer != null) {
|
||||||
Message transformedMessage = transformer.consumerTransform(session, this, m);
|
Message transformedMessage = transformer.consumerTransform(session, this, m);
|
||||||
|
@ -706,11 +715,11 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
|
||||||
if (ack != null) {
|
if (ack != null) {
|
||||||
deliveredMessages.clear();
|
deliveredMessages.clear();
|
||||||
ackCounter = 0;
|
ackCounter = 0;
|
||||||
} else {
|
} else {
|
||||||
ack = pendingAck;
|
ack = pendingAck;
|
||||||
pendingAck = null;
|
pendingAck = null;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
} else if (pendingAck != null && pendingAck.isStandardAck()) {
|
} else if (pendingAck != null && pendingAck.isStandardAck()) {
|
||||||
ack = pendingAck;
|
ack = pendingAck;
|
||||||
pendingAck = null;
|
pendingAck = null;
|
||||||
|
@ -874,13 +883,13 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
|
||||||
if (optimizeAcknowledge) {
|
if (optimizeAcknowledge) {
|
||||||
ackCounter++;
|
ackCounter++;
|
||||||
if (ackCounter >= (info.getPrefetchSize() * .65) || (optimizeAcknowledgeTimeOut > 0 && System.currentTimeMillis() >= (optimizeAckTimestamp + optimizeAcknowledgeTimeOut))) {
|
if (ackCounter >= (info.getPrefetchSize() * .65) || (optimizeAcknowledgeTimeOut > 0 && System.currentTimeMillis() >= (optimizeAckTimestamp + optimizeAcknowledgeTimeOut))) {
|
||||||
MessageAck ack = makeAckForAllDeliveredMessages(MessageAck.STANDARD_ACK_TYPE);
|
MessageAck ack = makeAckForAllDeliveredMessages(MessageAck.STANDARD_ACK_TYPE);
|
||||||
if (ack != null) {
|
if (ack != null) {
|
||||||
deliveredMessages.clear();
|
deliveredMessages.clear();
|
||||||
ackCounter = 0;
|
ackCounter = 0;
|
||||||
session.sendAck(ack);
|
session.sendAck(ack);
|
||||||
optimizeAckTimestamp = System.currentTimeMillis();
|
optimizeAckTimestamp = System.currentTimeMillis();
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
} else {
|
} else {
|
||||||
MessageAck ack = makeAckForAllDeliveredMessages(MessageAck.STANDARD_ACK_TYPE);
|
MessageAck ack = makeAckForAllDeliveredMessages(MessageAck.STANDARD_ACK_TYPE);
|
||||||
|
@ -917,17 +926,17 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
|
||||||
* @param type Ack-Type (i.e. MessageAck.STANDARD_ACK_TYPE)
|
* @param type Ack-Type (i.e. MessageAck.STANDARD_ACK_TYPE)
|
||||||
* @return <code>null</code> if nothing to ack.
|
* @return <code>null</code> if nothing to ack.
|
||||||
*/
|
*/
|
||||||
private MessageAck makeAckForAllDeliveredMessages(byte type) {
|
private MessageAck makeAckForAllDeliveredMessages(byte type) {
|
||||||
synchronized (deliveredMessages) {
|
synchronized (deliveredMessages) {
|
||||||
if (deliveredMessages.isEmpty())
|
if (deliveredMessages.isEmpty())
|
||||||
return null;
|
return null;
|
||||||
|
|
||||||
MessageDispatch md = deliveredMessages.getFirst();
|
MessageDispatch md = deliveredMessages.getFirst();
|
||||||
MessageAck ack = new MessageAck(md, type, deliveredMessages.size());
|
MessageAck ack = new MessageAck(md, type, deliveredMessages.size());
|
||||||
ack.setFirstMessageId(deliveredMessages.getLast().getMessage().getMessageId());
|
ack.setFirstMessageId(deliveredMessages.getLast().getMessage().getMessageId());
|
||||||
return ack;
|
return ack;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
private void ackLater(MessageDispatch md, byte ackType) throws JMSException {
|
private void ackLater(MessageDispatch md, byte ackType) throws JMSException {
|
||||||
|
|
||||||
|
@ -1016,7 +1025,7 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
|
||||||
// Acknowledge all messages so far.
|
// Acknowledge all messages so far.
|
||||||
MessageAck ack = makeAckForAllDeliveredMessages(MessageAck.STANDARD_ACK_TYPE);
|
MessageAck ack = makeAckForAllDeliveredMessages(MessageAck.STANDARD_ACK_TYPE);
|
||||||
if (ack == null)
|
if (ack == null)
|
||||||
return; // no msgs
|
return; // no msgs
|
||||||
|
|
||||||
if (session.getTransacted()) {
|
if (session.getTransacted()) {
|
||||||
rollbackOnFailedRecoveryRedelivery();
|
rollbackOnFailedRecoveryRedelivery();
|
||||||
|
@ -1154,7 +1163,7 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
|
||||||
|
|
||||||
MessageAck ack = new MessageAck(lastMd, MessageAck.POSION_ACK_TYPE, deliveredMessages.size());
|
MessageAck ack = new MessageAck(lastMd, MessageAck.POSION_ACK_TYPE, deliveredMessages.size());
|
||||||
ack.setPoisonCause(lastMd.getRollbackCause());
|
ack.setPoisonCause(lastMd.getRollbackCause());
|
||||||
ack.setFirstMessageId(firstMsgId);
|
ack.setFirstMessageId(firstMsgId);
|
||||||
session.sendAck(ack,true);
|
session.sendAck(ack,true);
|
||||||
// Adjust the window size.
|
// Adjust the window size.
|
||||||
additionalWindowSize = Math.max(0, additionalWindowSize - deliveredMessages.size());
|
additionalWindowSize = Math.max(0, additionalWindowSize - deliveredMessages.size());
|
||||||
|
@ -1410,11 +1419,11 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
|
||||||
return lastDeliveredSequenceId;
|
return lastDeliveredSequenceId;
|
||||||
}
|
}
|
||||||
|
|
||||||
public IOException getFailureError() {
|
public IOException getFailureError() {
|
||||||
return failureError;
|
return failureError;
|
||||||
}
|
}
|
||||||
|
|
||||||
public void setFailureError(IOException failureError) {
|
public void setFailureError(IOException failureError) {
|
||||||
this.failureError = failureError;
|
this.failureError = failureError;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
|
@ -33,6 +33,8 @@ import org.apache.activemq.management.StatsCapable;
|
||||||
import org.apache.activemq.management.StatsImpl;
|
import org.apache.activemq.management.StatsImpl;
|
||||||
import org.apache.activemq.usage.MemoryUsage;
|
import org.apache.activemq.usage.MemoryUsage;
|
||||||
import org.apache.activemq.util.IntrospectionSupport;
|
import org.apache.activemq.util.IntrospectionSupport;
|
||||||
|
import org.slf4j.Logger;
|
||||||
|
import org.slf4j.LoggerFactory;
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* A client uses a <CODE>MessageProducer</CODE> object to send messages to a
|
* A client uses a <CODE>MessageProducer</CODE> object to send messages to a
|
||||||
|
@ -68,6 +70,8 @@ import org.apache.activemq.util.IntrospectionSupport;
|
||||||
*/
|
*/
|
||||||
public class ActiveMQMessageProducer extends ActiveMQMessageProducerSupport implements StatsCapable, Disposable {
|
public class ActiveMQMessageProducer extends ActiveMQMessageProducerSupport implements StatsCapable, Disposable {
|
||||||
|
|
||||||
|
private static final Logger LOG = LoggerFactory.getLogger(ActiveMQMessageProducer.class);
|
||||||
|
|
||||||
protected ProducerInfo info;
|
protected ProducerInfo info;
|
||||||
protected boolean closed;
|
protected boolean closed;
|
||||||
|
|
||||||
|
@ -81,10 +85,21 @@ public class ActiveMQMessageProducer extends ActiveMQMessageProducerSupport impl
|
||||||
super(session);
|
super(session);
|
||||||
this.info = new ProducerInfo(producerId);
|
this.info = new ProducerInfo(producerId);
|
||||||
this.info.setWindowSize(session.connection.getProducerWindowSize());
|
this.info.setWindowSize(session.connection.getProducerWindowSize());
|
||||||
|
// Allows the options on the destination to configure the producerInfo
|
||||||
if (destination != null && destination.getOptions() != null) {
|
if (destination != null && destination.getOptions() != null) {
|
||||||
Map<String, String> options = new HashMap<String, String>(destination.getOptions());
|
Map<String, String> options = new HashMap<String, String>(destination.getOptions());
|
||||||
IntrospectionSupport.setProperties(this.info, options, "producer.");
|
IntrospectionSupport.setProperties(this.info, options, "producer.");
|
||||||
|
if (options.size() > 0) {
|
||||||
|
String msg = "There are " + options.size()
|
||||||
|
+ " producer options that couldn't be set on the producer."
|
||||||
|
+ " Check the options are spelled correctly."
|
||||||
|
+ " Unknown parameters=[" + options + "]."
|
||||||
|
+ " This producer cannot be started.";
|
||||||
|
LOG.warn(msg);
|
||||||
|
throw new ConfigurationException(msg);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
this.info.setDestination(destination);
|
this.info.setDestination(destination);
|
||||||
|
|
||||||
// Enable producer window flow control if protocol > 3 and the window
|
// Enable producer window flow control if protocol > 3 and the window
|
||||||
|
|
Loading…
Reference in New Issue