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

go back to less granular synchronization 

git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@492461 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Robert Davies 2007-01-04 08:59:12 +00:00
parent 481fc1ee3a
commit 158dbc66e7
1 changed files with 276 additions and 319 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -1,27 +1,22 @@
/** /**
* *
* Licensed to the Apache Software Foundation (ASF) under one or more * Licensed to the Apache Software Foundation (ASF) under one or more contributor license agreements. See the NOTICE
* contributor license agreements. See the NOTICE file distributed with * file distributed with this work for additional information regarding copyright ownership. The ASF licenses this file
* this work for additional information regarding copyright ownership. * to You under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with the
* The ASF licenses this file to You under the Apache License, Version 2.0 * License. You may obtain a copy of the License at
* (the "License"); you may not use this file except in compliance with *
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0 * http://www.apache.org/licenses/LICENSE-2.0
* *
* Unless required by applicable law or agreed to in writing, software * Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on
* distributed under the License is distributed on an "AS IS" BASIS, * an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. * specific language governing permissions and limitations under the License.
* See the License for the specific language governing permissions and
* limitations under the License.
*/ */
package org.apache.activemq.broker.region; package org.apache.activemq.broker.region;
import java.io.IOException; import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator; import java.util.Iterator;
import java.util.LinkedList; import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import javax.jms.InvalidSelectorException; import javax.jms.InvalidSelectorException;
import javax.jms.JMSException; import javax.jms.JMSException;
@ -45,91 +40,86 @@ import org.apache.activemq.transaction.Synchronization;
import org.apache.activemq.util.BrokerSupport; import org.apache.activemq.util.BrokerSupport;
import org.apache.commons.logging.Log; import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory; import org.apache.commons.logging.LogFactory;
/** /**
* A subscription that honors the pre-fetch option of the ConsumerInfo. * A subscription that honors the pre-fetch option of the ConsumerInfo.
* *
* @version $Revision: 1.15 $ * @version $Revision: 1.15 $
*/ */
abstract public class PrefetchSubscription extends AbstractSubscription{ abstract public class PrefetchSubscription extends AbstractSubscription{
static private final Log log=LogFactory.getLog(PrefetchSubscription.class); static private final Log log=LogFactory.getLog(PrefetchSubscription.class);
final protected PendingMessageCursor pending; final protected PendingMessageCursor pending;
final protected LinkedList dispatched=new LinkedList(); final protected LinkedList dispatched=new LinkedList();
protected int prefetchExtension=0; protected int prefetchExtension=0;
protected long enqueueCounter; protected long enqueueCounter;
protected long dispatchCounter; protected long dispatchCounter;
protected long dequeueCounter; protected long dequeueCounter;
private AtomicBoolean dispatching = new AtomicBoolean(); private AtomicBoolean dispatching=new AtomicBoolean();
public PrefetchSubscription(Broker broker,ConnectionContext context,ConsumerInfo info, PendingMessageCursor cursor) public PrefetchSubscription(Broker broker,ConnectionContext context,ConsumerInfo info,PendingMessageCursor cursor)
throws InvalidSelectorException{ throws InvalidSelectorException{
super(broker,context,info); super(broker,context,info);
pending = cursor; pending=cursor;
} }
public PrefetchSubscription(Broker broker,ConnectionContext context,ConsumerInfo info) public PrefetchSubscription(Broker broker,ConnectionContext context,ConsumerInfo info)
throws InvalidSelectorException{ throws InvalidSelectorException{
this(broker,context,info,new VMPendingMessageCursor()); this(broker,context,info,new VMPendingMessageCursor());
} }
/** /**
* Allows a message to be pulled on demand by a client * Allows a message to be pulled on demand by a client
*/ */
public Response pullMessage(ConnectionContext context, MessagePull pull) throws Exception { public synchronized Response pullMessage(ConnectionContext context,MessagePull pull) throws Exception{
// The slave should not deliver pull messages. TODO: when the slave becomes a master, // The slave should not deliver pull messages. TODO: when the slave becomes a master,
// He should send a NULL message to all the consumers to 'wake them up' in case // He should send a NULL message to all the consumers to 'wake them up' in case
// they were waiting for a message. // they were waiting for a message.
if (getPrefetchSize() == 0 && !isSlaveBroker()) { if(getPrefetchSize()==0&&!isSlaveBroker()){
prefetchExtension++; prefetchExtension++;
final long dispatchCounterBeforePull=dispatchCounter;
final long dispatchCounterBeforePull = dispatchCounter;
dispatchMatched(); dispatchMatched();
// If there was nothing dispatched.. we may need to setup a timeout.
// If there was nothing dispatched.. we may need to setup a timeout. if(dispatchCounterBeforePull==dispatchCounter){
if( dispatchCounterBeforePull == dispatchCounter ) { // imediate timeout used by receiveNoWait()
// imediate timeout used by receiveNoWait() if(pull.getTimeout()==-1){
if( pull.getTimeout() == -1 ) { // Send a NULL message.
// Send a NULL message. add(QueueMessageReference.NULL_MESSAGE);
add(QueueMessageReference.NULL_MESSAGE); dispatchMatched();
dispatchMatched(); }
} if(pull.getTimeout()>0){
if( pull.getTimeout() > 0 ) { Scheduler.executeAfterDelay(new Runnable(){
Scheduler.executeAfterDelay(new Runnable(){
public void run() { public void run(){
pullTimeout(dispatchCounterBeforePull); pullTimeout(dispatchCounterBeforePull);
} }
}, pull.getTimeout()); },pull.getTimeout());
} }
} }
} }
return null; return null;
} }
/** /**
* Occurs when a pull times out. If nothing has been dispatched * Occurs when a pull times out. If nothing has been dispatched since the timeout was setup, then send the NULL
* since the timeout was setup, then send the NULL message. * message.
*/ */
private void pullTimeout(long dispatchCounterBeforePull) { private void pullTimeout(long dispatchCounterBeforePull){
if( dispatchCounterBeforePull == dispatchCounter ) { if(dispatchCounterBeforePull==dispatchCounter){
try { try{
add(QueueMessageReference.NULL_MESSAGE); add(QueueMessageReference.NULL_MESSAGE);
dispatchMatched(); dispatchMatched();
} catch (Exception e) { }catch(Exception e){
context.getConnection().serviceException(e); context.getConnection().serviceException(e);
} }
}
}
public void add(MessageReference node) throws Exception{
boolean pendingEmpty=false;
synchronized(pending){
pendingEmpty=pending.isEmpty();
enqueueCounter++;
} }
}
public synchronized void add(MessageReference node) throws Exception{
boolean pendingEmpty=false;
pendingEmpty=pending.isEmpty();
enqueueCounter++;
if(!isFull()&&pendingEmpty&&!broker.isSlaveBroker()){ if(!isFull()&&pendingEmpty&&!broker.isSlaveBroker()){
dispatch(node); dispatch(node);
}else{ }else{
@ -137,142 +127,133 @@ abstract public class PrefetchSubscription extends AbstractSubscription{
synchronized(pending){ synchronized(pending){
if(pending.isEmpty()&&log.isDebugEnabled()){ if(pending.isEmpty()&&log.isDebugEnabled()){
log.debug("Prefetch limit."); log.debug("Prefetch limit.");
} }
pending.addMessageLast(node); pending.addMessageLast(node);
} }
//we might be able to dispatch messages (i.e. not full() anymore)
dispatchMatched();
} }
} }
public void processMessageDispatchNotification(MessageDispatchNotification mdn) throws Exception{ public synchronized void processMessageDispatchNotification(MessageDispatchNotification mdn) throws Exception{
synchronized(pending){ try{
try{ pending.reset();
pending.reset(); while(pending.hasNext()){
while(pending.hasNext()){ MessageReference node=pending.next();
MessageReference node=pending.next(); if(node.getMessageId().equals(mdn.getMessageId())){
if(node.getMessageId().equals(mdn.getMessageId())){ pending.remove();
pending.remove(); createMessageDispatch(node,node.getMessage());
createMessageDispatch(node,node.getMessage()); dispatched.addLast(node);
dispatched.addLast(node); return;
return;
}
} }
}finally{
pending.release();
} }
throw new JMSException("Slave broker out of sync with master: Dispatched message ("+mdn.getMessageId() }finally{
+") was not in the pending list"); pending.release();
} }
throw new JMSException("Slave broker out of sync with master: Dispatched message ("+mdn.getMessageId()
+") was not in the pending list");
} }
public void acknowledge(final ConnectionContext context,final MessageAck ack) throws Exception{ public synchronized void acknowledge(final ConnectionContext context,final MessageAck ack) throws Exception{
// Handle the standard acknowledgment case. // Handle the standard acknowledgment case.
boolean callDispatchMatched=false; boolean callDispatchMatched=false;
synchronized(dispatched){ if(ack.isStandardAck()){
if(ack.isStandardAck()){ // Acknowledge all dispatched messages up till the message id of the acknowledgment.
// Acknowledge all dispatched messages up till the message id of the acknowledgment. int index=0;
int index=0; boolean inAckRange=false;
boolean inAckRange=false; for(Iterator iter=dispatched.iterator();iter.hasNext();){
for(Iterator iter=dispatched.iterator();iter.hasNext();){ final MessageReference node=(MessageReference)iter.next();
final MessageReference node=(MessageReference)iter.next(); MessageId messageId=node.getMessageId();
MessageId messageId=node.getMessageId(); if(ack.getFirstMessageId()==null||ack.getFirstMessageId().equals(messageId)){
if(ack.getFirstMessageId()==null||ack.getFirstMessageId().equals(messageId)){ inAckRange=true;
inAckRange=true; }
} if(inAckRange){
if(inAckRange){ // Don't remove the nodes until we are committed.
// Don't remove the nodes until we are committed. if(!context.isInTransaction()){
if(!context.isInTransaction()){ dequeueCounter++;
dequeueCounter++; node.getRegionDestination().getDestinationStatistics().getDequeues().increment();
node.getRegionDestination().getDestinationStatistics().getDequeues().increment(); iter.remove();
iter.remove(); }else{
}else{ // setup a Synchronization to remove nodes from the dispatched list.
// setup a Synchronization to remove nodes from the dispatched list. context.getTransaction().addSynchronization(new Synchronization(){
context.getTransaction().addSynchronization(new Synchronization(){
public void afterCommit() throws Exception{ public void afterCommit() throws Exception{
synchronized(PrefetchSubscription.this){ synchronized(PrefetchSubscription.this){
dequeueCounter++; dequeueCounter++;
dispatched.remove(node); dispatched.remove(node);
node.getRegionDestination().getDestinationStatistics().getDequeues() node.getRegionDestination().getDestinationStatistics().getDequeues().increment();
.increment(); prefetchExtension--;
prefetchExtension--;
}
} }
public void afterRollback() throws Exception{
super.afterRollback();
}
});
}
index++;
acknowledge(context,ack,node);
if(ack.getLastMessageId().equals(messageId)){
if(context.isInTransaction()){
// extend prefetch window only if not a pulling consumer
if(getPrefetchSize()!=0){
prefetchExtension=Math.max(prefetchExtension,index+1);
}
}else{
prefetchExtension=Math.max(0,prefetchExtension-(index+1));
} }
callDispatchMatched=true;
break; public void afterRollback() throws Exception{
} super.afterRollback();
}
});
} }
} index++;
// this only happens after a reconnect - get an ack which is not valid acknowledge(context,ack,node);
if(!callDispatchMatched){ if(ack.getLastMessageId().equals(messageId)){
log.info("Could not correlate acknowledgment with dispatched message: "+ack); if(context.isInTransaction()){
} // extend prefetch window only if not a pulling consumer
}else if(ack.isDeliveredAck()){ if(getPrefetchSize()!=0){
// Message was delivered but not acknowledged: update pre-fetch counters. prefetchExtension=Math.max(prefetchExtension,index+1);
// Acknowledge all dispatched messages up till the message id of the acknowledgment. }
int index=0; }else{
for(Iterator iter=dispatched.iterator();iter.hasNext();index++){ prefetchExtension=Math.max(0,prefetchExtension-(index+1));
final MessageReference node=(MessageReference)iter.next(); }
if(ack.getLastMessageId().equals(node.getMessageId())){
prefetchExtension=Math.max(prefetchExtension,index+1);
callDispatchMatched=true; callDispatchMatched=true;
break; break;
} }
} }
if(!callDispatchMatched){ }
throw new JMSException("Could not correlate acknowledgment with dispatched message: "+ack); // this only happens after a reconnect - get an ack which is not valid
if(!callDispatchMatched){
log.info("Could not correlate acknowledgment with dispatched message: "+ack);
}
}else if(ack.isDeliveredAck()){
// Message was delivered but not acknowledged: update pre-fetch counters.
// Acknowledge all dispatched messages up till the message id of the acknowledgment.
int index=0;
for(Iterator iter=dispatched.iterator();iter.hasNext();index++){
final MessageReference node=(MessageReference)iter.next();
if(ack.getLastMessageId().equals(node.getMessageId())){
prefetchExtension=Math.max(prefetchExtension,index+1);
callDispatchMatched=true;
break;
} }
}else if(ack.isPoisonAck()){ }
// TODO: what if the message is already in a DLQ??? if(!callDispatchMatched){
// Handle the poison ACK case: we need to send the message to a DLQ throw new JMSException("Could not correlate acknowledgment with dispatched message: "+ack);
if(ack.isInTransaction()) }
throw new JMSException("Poison ack cannot be transacted: "+ack); }else if(ack.isPoisonAck()){
// Acknowledge all dispatched messages up till the message id of the acknowledgment. // TODO: what if the message is already in a DLQ???
int index=0; // Handle the poison ACK case: we need to send the message to a DLQ
boolean inAckRange=false; if(ack.isInTransaction())
for(Iterator iter=dispatched.iterator();iter.hasNext();){ throw new JMSException("Poison ack cannot be transacted: "+ack);
final MessageReference node=(MessageReference)iter.next(); // Acknowledge all dispatched messages up till the message id of the acknowledgment.
MessageId messageId=node.getMessageId(); int index=0;
if(ack.getFirstMessageId()==null||ack.getFirstMessageId().equals(messageId)){ boolean inAckRange=false;
inAckRange=true; for(Iterator iter=dispatched.iterator();iter.hasNext();){
} final MessageReference node=(MessageReference)iter.next();
if(inAckRange){ MessageId messageId=node.getMessageId();
sendToDLQ(context,node); if(ack.getFirstMessageId()==null||ack.getFirstMessageId().equals(messageId)){
node.getRegionDestination().getDestinationStatistics().getDequeues().increment(); inAckRange=true;
iter.remove(); }
dequeueCounter++; if(inAckRange){
index++; sendToDLQ(context,node);
acknowledge(context,ack,node); node.getRegionDestination().getDestinationStatistics().getDequeues().increment();
if(ack.getLastMessageId().equals(messageId)){ iter.remove();
prefetchExtension=Math.max(0,prefetchExtension-(index+1)); dequeueCounter++;
callDispatchMatched=true; index++;
break; acknowledge(context,ack,node);
} if(ack.getLastMessageId().equals(messageId)){
prefetchExtension=Math.max(0,prefetchExtension-(index+1));
callDispatchMatched=true;
break;
} }
} }
if(!callDispatchMatched){ }
throw new JMSException("Could not correlate acknowledgment with dispatched message: "+ack); if(!callDispatchMatched){
} throw new JMSException("Could not correlate acknowledgment with dispatched message: "+ack);
} }
} }
if(callDispatchMatched){ if(callDispatchMatched){
@ -293,7 +274,7 @@ abstract public class PrefetchSubscription extends AbstractSubscription{
* @throws IOException * @throws IOException
* @throws Exception * @throws Exception
*/ */
protected void sendToDLQ(final ConnectionContext context, final MessageReference node) throws IOException, Exception { protected void sendToDLQ(final ConnectionContext context,final MessageReference node) throws IOException,Exception{
// Send the message to the DLQ // Send the message to the DLQ
Message message=node.getMessage(); Message message=node.getMessage();
if(message!=null){ if(message!=null){
@ -301,142 +282,118 @@ abstract public class PrefetchSubscription extends AbstractSubscription{
// sent, // sent,
// it is only populated if the message is routed to another destination like the DLQ // it is only populated if the message is routed to another destination like the DLQ
DeadLetterStrategy deadLetterStrategy=node.getRegionDestination().getDeadLetterStrategy(); DeadLetterStrategy deadLetterStrategy=node.getRegionDestination().getDeadLetterStrategy();
ActiveMQDestination deadLetterDestination=deadLetterStrategy.getDeadLetterQueueFor(message.getDestination()); ActiveMQDestination deadLetterDestination=deadLetterStrategy
BrokerSupport.resend(context, message, deadLetterDestination); .getDeadLetterQueueFor(message.getDestination());
BrokerSupport.resend(context,message,deadLetterDestination);
} }
} }
/** /**
* Used to determine if the broker can dispatch to the consumer. * Used to determine if the broker can dispatch to the consumer.
*
* @return * @return
*/ */
protected boolean isFull(){ protected synchronized boolean isFull(){
return isSlaveBroker() || dispatched.size()-prefetchExtension>=info.getPrefetchSize(); return isSlaveBroker()||dispatched.size()-prefetchExtension>=info.getPrefetchSize();
} }
/** /**
* @return true when 60% or more room is left for dispatching messages * @return true when 60% or more room is left for dispatching messages
*/ */
public boolean isLowWaterMark(){ public boolean isLowWaterMark(){
return (dispatched.size()-prefetchExtension) <= (info.getPrefetchSize() *.4); return (dispatched.size()-prefetchExtension)<=(info.getPrefetchSize()*.4);
} }
/** /**
* @return true when 10% or less room is left for dispatching messages * @return true when 10% or less room is left for dispatching messages
*/ */
public boolean isHighWaterMark(){ public boolean isHighWaterMark(){
return (dispatched.size()-prefetchExtension) >= (info.getPrefetchSize() *.9); return (dispatched.size()-prefetchExtension)>=(info.getPrefetchSize()*.9);
} }
public int countBeforeFull() { public synchronized int countBeforeFull(){
return info.getPrefetchSize() + prefetchExtension - dispatched.size(); return info.getPrefetchSize()+prefetchExtension-dispatched.size();
} }
public int getPendingQueueSize(){ public int getPendingQueueSize(){
synchronized(pending) { synchronized(pending){
return pending.size(); return pending.size();
} }
} }
public int getDispatchedQueueSize(){ public int getDispatchedQueueSize(){
synchronized(dispatched){ synchronized(dispatched){
return dispatched.size(); return dispatched.size();
} }
} }
synchronized public long getDequeueCounter(){ synchronized public long getDequeueCounter(){
return dequeueCounter; return dequeueCounter;
} }
synchronized public long getDispatchedCounter() { synchronized public long getDispatchedCounter(){
return dispatchCounter; return dispatchCounter;
} }
synchronized public long getEnqueueCounter() { synchronized public long getEnqueueCounter(){
return enqueueCounter; return enqueueCounter;
} }
public boolean isRecoveryRequired(){ public boolean isRecoveryRequired(){
return pending.isRecoveryRequired(); return pending.isRecoveryRequired();
} }
/** /**
* optimize message consumer prefetch if the consumer supports it * optimize message consumer prefetch if the consumer supports it
* *
*/ */
public void optimizePrefetch(){ public void optimizePrefetch(){
/* /*
if(info!=null&&info.isOptimizedAcknowledge()&&context!=null&&context.getConnection()!=null * if(info!=null&&info.isOptimizedAcknowledge()&&context!=null&&context.getConnection()!=null
&&context.getConnection().isManageable()){ * &&context.getConnection().isManageable()){ if(info.getCurrentPrefetchSize()!=info.getPrefetchSize() &&
if(info.getCurrentPrefetchSize()!=info.getPrefetchSize() && isLowWaterMark()){ * isLowWaterMark()){ info.setCurrentPrefetchSize(info.getPrefetchSize());
info.setCurrentPrefetchSize(info.getPrefetchSize()); * updateConsumerPrefetch(info.getPrefetchSize()); }else
updateConsumerPrefetch(info.getPrefetchSize()); * if(info.getCurrentPrefetchSize()==info.getPrefetchSize() && isHighWaterMark()){ // want to purge any
}else if(info.getCurrentPrefetchSize()==info.getPrefetchSize() && isHighWaterMark()){ * outstanding acks held by the consumer info.setCurrentPrefetchSize(1); updateConsumerPrefetch(1); } }
// want to purge any outstanding acks held by the consumer */
info.setCurrentPrefetchSize(1);
updateConsumerPrefetch(1);
}
}
*/
} }
public void add(ConnectionContext context,Destination destination) throws Exception{ public synchronized void add(ConnectionContext context,Destination destination) throws Exception{
super.add(context,destination); super.add(context,destination);
synchronized(pending){ pending.add(context,destination);
pending.add(context,destination);
}
} }
public void remove(ConnectionContext context,Destination destination) throws Exception{ public synchronized void remove(ConnectionContext context,Destination destination) throws Exception{
super.remove(context,destination); super.remove(context,destination);
synchronized(pending){ pending.remove(context,destination);
pending.remove(context,destination);
}
} }
protected void dispatchMatched() throws IOException{ protected void dispatchMatched() throws IOException{
if(!broker.isSlaveBroker() && dispatching.compareAndSet(false,true)){ if(!broker.isSlaveBroker()&&dispatching.compareAndSet(false,true)){
try{ try{
List toDispatch=null; try{
synchronized(pending){ int numberToDispatch=countBeforeFull();
try{ if(numberToDispatch>0){
int numberToDispatch=countBeforeFull(); int count=0;
if(numberToDispatch>0){ pending.reset();
int count=0; while(pending.hasNext()&&!isFull()&&count<numberToDispatch){
pending.reset(); MessageReference node=pending.next();
while(pending.hasNext()&&!isFull()&&count<numberToDispatch){ if(node==null)
MessageReference node=pending.next(); break;
if ( node == null ) if(canDispatch(node)){
break; pending.remove();
// Message may have been sitting in the pending list a while
if(canDispatch(node)){ // waiting for the consumer to ak the message.
pending.remove(); if(node!=QueueMessageReference.NULL_MESSAGE&&node.isExpired()){
// Message may have been sitting in the pending list a while continue; // just drop it.
// waiting for the consumer to ak the message.
if(node!=QueueMessageReference.NULL_MESSAGE&&node.isExpired()){
continue; // just drop it.
}
if(toDispatch==null){
toDispatch=new ArrayList();
}
toDispatch.add(node);
count++;
} }
dispatch(node);
count++;
} }
} }
}finally{
pending.release();
}
}
if(toDispatch!=null){
synchronized(dispatched){
for(int i=0;i<toDispatch.size();i++){
MessageReference node=(MessageReference)toDispatch.get(i);
dispatch(node);
}
} }
}finally{
pending.release();
} }
}finally{ }finally{
dispatching.set(false); dispatching.set(false);
@ -449,45 +406,44 @@ abstract public class PrefetchSubscription extends AbstractSubscription{
if(message==null){ if(message==null){
return false; return false;
} }
synchronized(dispatched){ // Make sure we can dispatch a message.
// Make sure we can dispatch a message. if(canDispatch(node)&&!isSlaveBroker()){
if(canDispatch(node)&&!isSlaveBroker()){ MessageDispatch md=createMessageDispatch(node,message);
MessageDispatch md=createMessageDispatch(node,message); // NULL messages don't count... they don't get Acked.
// NULL messages don't count... they don't get Acked. if(node!=QueueMessageReference.NULL_MESSAGE){
if(node!=QueueMessageReference.NULL_MESSAGE){ dispatchCounter++;
dispatchCounter++; dispatched.addLast(node);
dispatched.addLast(node);
}else{
prefetchExtension=Math.max(0,prefetchExtension-1);
}
if(info.isDispatchAsync()){
md.setConsumer(new Runnable(){
public void run(){
// Since the message gets queued up in async dispatch, we don't want to
// decrease the reference count until it gets put on the wire.
onDispatch(node,message);
}
});
context.getConnection().dispatchAsync(md);
}else{
context.getConnection().dispatchSync(md);
onDispatch(node,message);
}
return true;
}else{ }else{
QueueMessageReference n = (QueueMessageReference) node; prefetchExtension=Math.max(0,prefetchExtension-1);
return false;
} }
if(info.isDispatchAsync()){
md.setConsumer(new Runnable(){
public void run(){
// Since the message gets queued up in async dispatch, we don't want to
// decrease the reference count until it gets put on the wire.
onDispatch(node,message);
}
});
context.getConnection().dispatchAsync(md);
}else{
context.getConnection().dispatchSync(md);
onDispatch(node,message);
}
//System.err.println(broker.getBrokerName() + " " + this + " (" + enqueueCounter + ", " + dispatchCounter +") " + node);
return true;
}else{
QueueMessageReference n=(QueueMessageReference)node;
return false;
} }
} }
protected void onDispatch(final MessageReference node,final Message message){ protected void onDispatch(final MessageReference node,final Message message){
if(node.getRegionDestination()!=null){ if(node.getRegionDestination()!=null){
if( node != QueueMessageReference.NULL_MESSAGE ) { if(node!=QueueMessageReference.NULL_MESSAGE){
node.getRegionDestination().getDestinationStatistics().getDispatched().increment(); node.getRegionDestination().getDestinationStatistics().getDispatched().increment();
context.getConnection().getStatistics().onMessageDequeue(message); context.getConnection().getStatistics().onMessageDequeue(message);
} }
try{ try{
dispatchMatched(); dispatchMatched();
}catch(IOException e){ }catch(IOException e){
@ -495,14 +451,15 @@ abstract public class PrefetchSubscription extends AbstractSubscription{
} }
} }
} }
/** /**
* inform the MessageConsumer on the client to change it's prefetch * inform the MessageConsumer on the client to change it's prefetch
*
* @param newPrefetch * @param newPrefetch
*/ */
public void updateConsumerPrefetch(int newPrefetch){ public void updateConsumerPrefetch(int newPrefetch){
if (context != null && context.getConnection() != null && context.getConnection().isManageable()){ if(context!=null&&context.getConnection()!=null&&context.getConnection().isManageable()){
ConsumerControl cc = new ConsumerControl(); ConsumerControl cc=new ConsumerControl();
cc.setConsumerId(info.getConsumerId()); cc.setConsumerId(info.getConsumerId());
cc.setPrefetch(newPrefetch); cc.setPrefetch(newPrefetch);
context.getConnection().dispatchAsync(cc); context.getConnection().dispatchAsync(cc);
@ -515,20 +472,20 @@ abstract public class PrefetchSubscription extends AbstractSubscription{
* @return MessageDispatch * @return MessageDispatch
*/ */
protected MessageDispatch createMessageDispatch(MessageReference node,Message message){ protected MessageDispatch createMessageDispatch(MessageReference node,Message message){
if( node == QueueMessageReference.NULL_MESSAGE ) { if(node==QueueMessageReference.NULL_MESSAGE){
MessageDispatch md=new MessageDispatch(); MessageDispatch md=new MessageDispatch();
md.setMessage(null); md.setMessage(null);
md.setConsumerId(info.getConsumerId()); md.setConsumerId(info.getConsumerId());
md.setDestination( null ); md.setDestination(null);
return md; return md;
} else { }else{
MessageDispatch md=new MessageDispatch(); MessageDispatch md=new MessageDispatch();
md.setConsumerId(info.getConsumerId()); md.setConsumerId(info.getConsumerId());
md.setDestination(node.getRegionDestination().getActiveMQDestination()); md.setDestination(node.getRegionDestination().getActiveMQDestination());
md.setMessage(message); md.setMessage(message);
md.setRedeliveryCounter(node.getRedeliveryCounter()); md.setRedeliveryCounter(node.getRedeliveryCounter());
return md; return md;
} }
} }
/** /**
@ -537,7 +494,7 @@ abstract public class PrefetchSubscription extends AbstractSubscription{
* @param node * @param node
* @return false if the message should not be dispatched to the client (another sub may have already dispatched it * @return false if the message should not be dispatched to the client (another sub may have already dispatched it
* for example). * for example).
* @throws IOException * @throws IOException
*/ */
abstract protected boolean canDispatch(MessageReference node) throws IOException; abstract protected boolean canDispatch(MessageReference node) throws IOException;
@ -547,6 +504,6 @@ abstract public class PrefetchSubscription extends AbstractSubscription{
* @throws IOException * @throws IOException
*/ */
protected void acknowledge(ConnectionContext context,final MessageAck ack,final MessageReference node) protected void acknowledge(ConnectionContext context,final MessageAck ack,final MessageReference node)
throws IOException{} throws IOException{
}
} }