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

Added support for length of time messages are processed by the broker - 

fix for https://issues.apache.org/activemq/browse/AMQ-1160,
https://issues.apache.org/activemq/browse/AMQ-1072,
https://issues.apache.org/activemq/browse/AMQ-936 
and ground work for for https://issues.apache.org/activemq/browse/AMQ-567

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@561026 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Robert Davies 2007-07-30 16:01:37 +00:00
parent b97f7d8479
commit a401575569
18 changed files with 204 additions and 126 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
activemq-core/src/main/java/org/apache/activemq/ActiveMQConnectionMetaData.java
View File

@ -142,6 +142,7 @@ public class ActiveMQConnectionMetaData implements ConnectionMetaData {
jmxProperties.put("JMSXGroupID", "1");
jmxProperties.put("JMSXGroupSeq", "1");
jmxProperties.put("JMSXDeliveryCount","1");
jmxProperties.put("JMSXProducerTXID","1");
return jmxProperties.keys();
}
}

5
activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
View File

@ -896,8 +896,11 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
ActiveMQMessage message=createActiveMQMessage(md);
beforeMessageIsConsumed(md);
try{
boolean expired=message.isExpired();
if(!expired){
listener.onMessage(message);
afterMessageIsConsumed(md,false);
}
afterMessageIsConsumed(md,expired);
}catch(RuntimeException e){
if(session.isDupsOkAcknowledge()||session.isAutoAcknowledge()){
// Redeliver the message

19
activemq-core/src/main/java/org/apache/activemq/broker/Broker.java
View File

@ -186,12 +186,17 @@ public interface Broker extends Region, Service {
*/
BrokerInfo[] getPeerBrokerInfos();
/**
* Notify the Broker that a dispatch is going to happen
* @param messageDispatch
*/
public void preProcessDispatch(MessageDispatch messageDispatch);
/**
* Notify the Broker that a dispatch has happened
* @param messageDispatch
*/
public void processDispatch(MessageDispatch messageDispatch);
public void postProcessDispatch(MessageDispatch messageDispatch);
/**
* @return true if the broker has stopped
@ -263,11 +268,18 @@ public interface Broker extends Region, Service {
*/
Broker getRoot();
/**
* Determine if a message has expired -allows default behaviour to be overriden -
* as the timestamp set by the producer can be out of sync with the broker
* @param messageReference
* @return true if the message is expired
*/
public boolean isExpired(MessageReference messageReference);
/**
* A Message has Expired
* @param context
* @param messageReference
* @throws Exception
*/
public void messageExpired(ConnectionContext context, MessageReference messageReference);
@ -275,7 +287,8 @@ public interface Broker extends Region, Service {
* A message needs to go the a DLQ
* @param context
* @param messageReference
* @throws Exception
*/
public void sendToDeadLetterQueue(ConnectionContext context,MessageReference messageReference);
}

19
activemq-core/src/main/java/org/apache/activemq/broker/BrokerFilter.java
View File

@ -142,7 +142,8 @@ public class BrokerFilter implements Broker {
return next.addDestination(context,destination);
}
public void removeDestination(ConnectionContext context, ActiveMQDestination destination, long timeout) throws Exception {
public void removeDestination(ConnectionContext context,ActiveMQDestination destination,long timeout)
throws Exception{
next.removeDestination(context,destination,timeout);
}
@ -178,7 +179,6 @@ public class BrokerFilter implements Broker {
next.gc();
}
public void addBroker(Connection connection,BrokerInfo info){
next.addBroker(connection,info);
}
@ -187,13 +187,16 @@ public class BrokerFilter implements Broker {
next.removeBroker(connection,info);
}
public BrokerInfo[] getPeerBrokerInfos(){
return next.getPeerBrokerInfos();
}
public void processDispatch(MessageDispatch messageDispatch){
next.processDispatch(messageDispatch);
public void preProcessDispatch(MessageDispatch messageDispatch){
next.preProcessDispatch(messageDispatch);
}
public void postProcessDispatch(MessageDispatch messageDispatch){
next.postProcessDispatch(messageDispatch);
}
public void processDispatchNotification(MessageDispatchNotification messageDispatchNotification) throws Exception{
@ -210,12 +213,10 @@ public class BrokerFilter implements Broker {
public void addDestinationInfo(ConnectionContext context,DestinationInfo info) throws Exception{
next.addDestinationInfo(context,info);
}
public void removeDestinationInfo(ConnectionContext context,DestinationInfo info) throws Exception{
next.removeDestinationInfo(context,info);
}
public boolean isFaultTolerantConfiguration(){
@ -246,6 +247,10 @@ public class BrokerFilter implements Broker {
return next.getBrokerService();
}
public boolean isExpired(MessageReference messageReference){
return next.isExpired(messageReference);
}
public void messageExpired(ConnectionContext context,MessageReference message){
next.messageExpired(context,message);
}

13
activemq-core/src/main/java/org/apache/activemq/broker/EmptyBroker.java
View File

@ -185,13 +185,10 @@ public class EmptyBroker implements Broker {
return null;
}
/**
* Notifiy the Broker that a dispatch has happened
*
* @param messageDispatch
*/
public void processDispatch(MessageDispatch messageDispatch) {
public void preProcessDispatch(MessageDispatch messageDispatch) {
}
public void postProcessDispatch(MessageDispatch messageDispatch) {
}
public void processDispatchNotification(MessageDispatchNotification messageDispatchNotification) throws Exception {
@ -245,6 +242,10 @@ public class EmptyBroker implements Broker {
return null;
}
public boolean isExpired(MessageReference messageReference) {
return false;
}
public void messageExpired(ConnectionContext context,MessageReference message){
}

10
activemq-core/src/main/java/org/apache/activemq/broker/ErrorBroker.java
View File

@ -188,7 +188,11 @@ public class ErrorBroker implements Broker {
throw new BrokerStoppedException(this.message);
}
public void processDispatch(MessageDispatch messageDispatch) {
public void preProcessDispatch(MessageDispatch messageDispatch) {
throw new BrokerStoppedException(this.message);
}
public void postProcessDispatch(MessageDispatch messageDispatch) {
throw new BrokerStoppedException(this.message);
}
@ -245,6 +249,10 @@ public class ErrorBroker implements Broker {
throw new BrokerStoppedException(this.message);
}
public boolean isExpired(MessageReference messageReference) {
throw new BrokerStoppedException(this.message);
}
public void messageExpired(ConnectionContext context,MessageReference message){
throw new BrokerStoppedException(this.message);
}

11
activemq-core/src/main/java/org/apache/activemq/broker/MutableBrokerFilter.java
View File

@ -200,8 +200,12 @@ public class MutableBrokerFilter implements Broker {
return getNext().getPeerBrokerInfos();
}
public void processDispatch(MessageDispatch messageDispatch){
getNext().processDispatch(messageDispatch);
public void preProcessDispatch(MessageDispatch messageDispatch){
getNext().preProcessDispatch(messageDispatch);
}
public void postProcessDispatch(MessageDispatch messageDispatch){
getNext().postProcessDispatch(messageDispatch);
}
public void processDispatchNotification(MessageDispatchNotification messageDispatchNotification) throws Exception{
@ -259,6 +263,9 @@ public class MutableBrokerFilter implements Broker {
return getNext().getBrokerService();
}
public boolean isExpired(MessageReference messageReference) {
return getNext().isExpired(messageReference);
}
public void messageExpired(ConnectionContext context,MessageReference message){
getNext().messageExpired(context,message);

17
activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java
View File

@ -741,7 +741,7 @@ public class TransportConnection implements Service,Connection,Task,CommandVisit
if(message.isMessageDispatch()) {
MessageDispatch md=(MessageDispatch) message;
Runnable sub=md.getTransmitCallback();
broker.processDispatch(md);
broker.postProcessDispatch(md);
if(sub!=null){
sub.run();
}
@ -750,21 +750,22 @@ public class TransportConnection implements Service,Connection,Task,CommandVisit
}
protected void processDispatch(Command command) throws IOException{
final MessageDispatch messageDispatch=(MessageDispatch)(command.isMessageDispatch()?command:null);
try{
if(!disposed.get()){
if(messageDispatch!=null){
broker.preProcessDispatch(messageDispatch);
}
dispatch(command);
}
}finally{
if(command.isMessageDispatch()){
MessageDispatch md=(MessageDispatch) command;
Runnable sub=md.getTransmitCallback();
broker.processDispatch(md);
if(messageDispatch!=null){
Runnable sub=messageDispatch.getTransmitCallback();
broker.postProcessDispatch(messageDispatch);
if(sub!=null){
sub.run();
}
}
getStatistics().getDequeues().increment();
}
}
@ -918,7 +919,7 @@ public class TransportConnection implements Service,Connection,Task,CommandVisit
if(command.isMessageDispatch()) {
MessageDispatch md=(MessageDispatch) command;
Runnable sub=md.getTransmitCallback();
broker.processDispatch(md);
broker.postProcessDispatch(md);
if(sub!=null){
sub.run();
}

4
activemq-core/src/main/java/org/apache/activemq/broker/ft/MasterBroker.java
View File

@ -283,7 +283,7 @@ public class MasterBroker extends InsertableMutableBrokerFilter{
*
* @param messageDispatch
*/
public void processDispatch(MessageDispatch messageDispatch){
public void postProcessDispatch(MessageDispatch messageDispatch){
MessageDispatchNotification mdn=new MessageDispatchNotification();
mdn.setConsumerId(messageDispatch.getConsumerId());
mdn.setDeliverySequenceId(messageDispatch.getDeliverySequenceId());
@ -293,7 +293,7 @@ public class MasterBroker extends InsertableMutableBrokerFilter{
mdn.setMessageId(msg.getMessageId());
sendAsyncToSlave(mdn);
}
super.processDispatch(messageDispatch);
super.postProcessDispatch(messageDispatch);
}
/**

2
activemq-core/src/main/java/org/apache/activemq/broker/region/PrefetchSubscription.java
View File

@ -382,7 +382,7 @@ abstract public class PrefetchSubscription extends AbstractSubscription{
pending.remove();
// Message may have been sitting in the pending list a while
// waiting for the consumer to ak the message.
if(node!=QueueMessageReference.NULL_MESSAGE&&node.isExpired()){
if(node!=QueueMessageReference.NULL_MESSAGE&&broker.isExpired(node)){
broker.messageExpired(getContext(),node);
dequeueCounter++;
continue;

8
activemq-core/src/main/java/org/apache/activemq/broker/region/Queue.java
View File

@ -137,7 +137,7 @@ public class Queue implements Destination, Task {
public boolean recoverMessage(Message message){
// Message could have expired while it was being loaded..
if(message.isExpired()){
if(broker.isExpired(message)){
broker.messageExpired(createConnectionContext(),message);
destinationStatistics.getMessages().decrement();
return true;
@ -379,7 +379,7 @@ public class Queue implements Destination, Task {
try {
// While waiting for space to free up... the message may have expired.
if(message.isExpired()) {
if(broker.isExpired(message)) {
broker.messageExpired(context,message);
destinationStatistics.getMessages().decrement();
} else {
@ -455,7 +455,7 @@ public class Queue implements Destination, Task {
try {
// It could take while before we receive the commit
// op, by that time the message could have expired..
if(message.isExpired()){
if(broker.isExpired(message)){
broker.messageExpired(context,message);
destinationStatistics.getMessages().decrement();
return;
@ -1014,7 +1014,7 @@ public class Queue implements Destination, Task {
while(messages.hasNext()&&count<toPageIn){
MessageReference node=messages.next();
messages.remove();
if(!node.isExpired()){
if(!broker.isExpired(node)){
node=createMessageReference(node.getMessage());
result.add(node);
count++;

13
activemq-core/src/main/java/org/apache/activemq/broker/region/RegionBroker.java
View File

@ -385,6 +385,7 @@ public class RegionBroker implements Broker {
public void send(ProducerBrokerExchange producerExchange,Message message) throws Exception{
long si=sequenceGenerator.getNextSequenceId();
message.getMessageId().setBrokerSequenceId(si);
message.setBrokerInTime(System.currentTimeMillis());
if(producerExchange.isMutable()||producerExchange.getRegion()==null){
ActiveMQDestination destination=message.getDestination();
// ensure the destination is registered with the RegionBroker
@ -538,8 +539,14 @@ public class RegionBroker implements Broker {
return result;
}
public void processDispatch(MessageDispatch messageDispatch){
public void preProcessDispatch(MessageDispatch messageDispatch){
Message message = messageDispatch.getMessage();
if(message != null) {
message.setBrokerOutTime(System.currentTimeMillis());
}
}
public void postProcessDispatch(MessageDispatch messageDispatch){
}
public void processDispatchNotification(MessageDispatchNotification messageDispatchNotification) throws Exception {
@ -625,6 +632,10 @@ public class RegionBroker implements Broker {
return brokerService;
}
public boolean isExpired(MessageReference messageReference) {
return messageReference.isExpired();
}
public void messageExpired(ConnectionContext context,MessageReference node){
if(log.isDebugEnabled()){
log.debug("Message expired "+node);

6
activemq-core/src/main/java/org/apache/activemq/broker/region/Topic.java
View File

@ -263,7 +263,7 @@ public class Topic implements Destination {
// There is delay between the client sending it and it arriving at the
// destination.. it may have expired.
if( message.isExpired() ) {
if( broker.isExpired(message) ) {
broker.messageExpired(context,message);
destinationStatistics.getMessages().decrement();
if( ( !message.isResponseRequired() || producerExchange.getProducerState().getInfo().getWindowSize() > 0 ) && !context.isInRecoveryMode() ) {
@ -286,7 +286,7 @@ public class Topic implements Destination {
public void run() {
// While waiting for space to free up... the message may have expired.
if(message.isExpired()){
if(broker.isExpired(message)){
broker.messageExpired(context,message);
destinationStatistics.getMessages().decrement();
@ -357,7 +357,7 @@ public class Topic implements Destination {
public void afterCommit() throws Exception {
// It could take while before we receive the commit
// operration.. by that time the message could have expired..
if( message.isExpired() ) {
if(broker.isExpired(message) ) {
broker.messageExpired(context,message);
message.decrementReferenceCount();
destinationStatistics.getMessages().decrement();

4
activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java
View File

@ -129,7 +129,7 @@ public class TopicSubscription extends AbstractSubscription{
matched.reset();
while(matched.hasNext()){
MessageReference node=matched.next();
if(node.isExpired()){
if(broker.isExpired(node)){
matched.remove();
dispatchedCounter.incrementAndGet();
node.decrementReferenceCount();
@ -361,7 +361,7 @@ public class TopicSubscription extends AbstractSubscription{
matched.remove();
// Message may have been sitting in the matched list a while
// waiting for the consumer to ak the message.
if(message.isExpired()){
if(broker.isExpired(message)){
message.decrementReferenceCount();
broker.messageExpired(getContext(),message);
dequeueCounter.incrementAndGet();

1
activemq-core/src/main/java/org/apache/activemq/broker/region/policy/AbstractDeadLetterStrategy.java
View File

@ -28,6 +28,7 @@ public abstract class AbstractDeadLetterStrategy implements DeadLetterStrategy
private boolean processNonPersistent=true;
private boolean processExpired=true;
public boolean isSendToDeadLetterQueue(Message message){
boolean result=false;
if(message!=null){

1
activemq-core/src/main/java/org/apache/activemq/broker/util/TimeStampingBrokerPlugin.java
View File

@ -18,7 +18,6 @@
package org.apache.activemq.broker.util;
import org.apache.activemq.broker.BrokerPluginSupport;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.ProducerBrokerExchange;
import org.apache.activemq.command.Message;

4
activemq-core/src/main/java/org/apache/activemq/broker/util/UDPTraceBrokerPlugin.java
View File

@ -189,9 +189,9 @@ public class UDPTraceBrokerPlugin extends BrokerPluginSupport {
return super.prepareTransaction(context, xid);
}
public void processDispatch(MessageDispatch messageDispatch) {
public void postProcessDispatch(MessageDispatch messageDispatch) {
trace(messageDispatch);
super.processDispatch(messageDispatch);
super.postProcessDispatch(messageDispatch);
}
public void processDispatchNotification(MessageDispatchNotification messageDispatchNotification) throws Exception {

28
activemq-core/src/main/java/org/apache/activemq/command/Message.java
View File

@ -54,6 +54,8 @@ abstract public class Message extends BaseCommand implements MarshallAware, Mess
protected long expiration;
protected long timestamp;
protected long arrival;
protected long brokerInTime;
protected long brokerOutTime;
protected String correlationId;
protected ActiveMQDestination replyTo;
protected boolean persistent;
@ -84,6 +86,8 @@ abstract public class Message extends BaseCommand implements MarshallAware, Mess
protected boolean droppable = false;
private BrokerId [] cluster;
abstract public Message copy();
protected void copy(Message copy) {
@ -123,6 +127,8 @@ abstract public class Message extends BaseCommand implements MarshallAware, Mess
copy.arrival = arrival;
copy.connection = connection;
copy.regionDestination = regionDestination;
copy.brokerInTime=brokerInTime;
copy.brokerOutTime=brokerOutTime;
//copying the broker path breaks networks - if a consumer re-uses a consumed
//message and forwards it on
//copy.brokerPath = brokerPath;
@ -630,4 +636,26 @@ abstract public class Message extends BaseCommand implements MarshallAware, Mess
public boolean isMessage() {
return true;
}
/**
* @openwire:property version=3
*/
public long getBrokerInTime(){
return this.brokerInTime;
}
public void setBrokerInTime(long brokerInTime){
this.brokerInTime=brokerInTime;
}
/**
* @openwire:property version=3
*/
public long getBrokerOutTime(){
return this.brokerOutTime;
}
public void setBrokerOutTime(long brokerOutTime){
this.brokerOutTime=brokerOutTime;
}
}