Added support for browsing Topics

git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@381644 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Robert Davies 2006-02-28 12:55:21 +00:00
parent 8ee9b45484
commit 2748ae1612
25 changed files with 291 additions and 208 deletions

View File

@ -2,110 +2,38 @@
* *
* Copyright 2005-2006 The Apache Software Foundation * Copyright 2005-2006 The Apache Software Foundation
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* you may not use this file except in compliance with the License. * the License. You may obtain a copy of the License at
* 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.jmx; package org.apache.activemq.broker.jmx;
import javax.management.openmbean.CompositeData; import javax.management.openmbean.CompositeData;
import javax.management.openmbean.CompositeDataSupport;
import javax.management.openmbean.CompositeType;
import javax.management.openmbean.OpenDataException; import javax.management.openmbean.OpenDataException;
import javax.management.openmbean.TabularData;
import javax.management.openmbean.TabularDataSupport;
import javax.management.openmbean.TabularType;
import org.apache.activemq.broker.jmx.OpenTypeSupport.OpenTypeFactory;
import org.apache.activemq.broker.region.Queue; import org.apache.activemq.broker.region.Queue;
import org.apache.activemq.command.ActiveMQMessage;
import org.apache.activemq.command.Message; import org.apache.activemq.command.Message;
public class QueueView extends DestinationView implements QueueViewMBean{
public class QueueView implements QueueViewMBean { public QueueView(Queue destination){
super(destination);
private final Queue destination;
public QueueView(Queue destination) {
this.destination = destination;
} }
public void gc() { public CompositeData getMessage(String messageId) throws OpenDataException{
destination.gc(); Message rc=((Queue) destination).getMessage(messageId);
} if(rc==null)
public void resetStatistics() {
destination.getDestinationStatistics().reset();
}
public long getEnqueueCount() {
return destination.getDestinationStatistics().getEnqueues().getCount();
}
public long getDequeueCount() {
return destination.getDestinationStatistics().getDequeues().getCount();
}
public long getConsumerCount() {
return destination.getDestinationStatistics().getConsumers().getCount();
}
public long getMessages() {
return destination.getDestinationStatistics().getMessages().getCount();
}
public long getMessagesCached() {
return destination.getDestinationStatistics().getMessagesCached().getCount();
}
public CompositeData[] browse() throws OpenDataException {
Message[] messages = destination.browse();
CompositeData c[] = new CompositeData[messages.length];
for (int i = 0; i < c.length; i++) {
try {
c[i] = OpenTypeSupport.convert(messages[i]);
} catch (Throwable e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
return c;
}
public TabularData browseAsTable() throws OpenDataException {
OpenTypeFactory factory = OpenTypeSupport.getFactory(ActiveMQMessage.class);
Message[] messages = destination.browse();
CompositeType ct = factory.getCompositeType();
TabularType tt = new TabularType("MessageList", "MessageList", ct, new String[]{"JMSMessageID"});
TabularDataSupport rc = new TabularDataSupport(tt);
for (int i = 0; i < messages.length; i++) {
rc.put(new CompositeDataSupport(ct, factory.getFields(messages[i])));
}
return rc;
}
public CompositeData getMessage(String messageId) throws OpenDataException {
Message rc = destination.getMessage(messageId);
if( rc ==null )
return null; return null;
return OpenTypeSupport.convert(rc); return OpenTypeSupport.convert(rc);
} }
public void removeMessage(String messageId) { public void removeMessage(String messageId){
destination.removeMessage(messageId); ((Queue) destination).removeMessage(messageId);
} }
public void purge() { public void purge(){
destination.purge(); ((Queue) destination).purge();
} }
} }

View File

@ -2,55 +2,21 @@
* *
* Copyright 2005-2006 The Apache Software Foundation * Copyright 2005-2006 The Apache Software Foundation
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* you may not use this file except in compliance with the License. * the License. You may obtain a copy of the License at
* 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.jmx; package org.apache.activemq.broker.jmx;
import org.apache.activemq.broker.region.Topic; import org.apache.activemq.broker.region.Topic;
public class TopicView extends DestinationView implements TopicViewMBean{
public class TopicView implements TopicViewMBean { public TopicView(Topic destination){
super(destination);
private final Topic destination;
public TopicView(Topic destination) {
this.destination = destination;
} }
public void gc() {
destination.gc();
}
public void resetStatistics() {
destination.getDestinationStatistics().reset();
}
public long getEnqueueCount() {
return destination.getDestinationStatistics().getEnqueues().getCount();
}
public long getDequeueCount() {
return destination.getDestinationStatistics().getDequeues().getCount();
}
public long getConsumerCount() {
return destination.getDestinationStatistics().getConsumers().getCount();
}
public long getMessages() {
return destination.getDestinationStatistics().getMessages().getCount();
}
public long getMessagesCached() {
return destination.getDestinationStatistics().getMessagesCached().getCount();
}
} }

View File

@ -16,6 +16,10 @@
*/ */
package org.apache.activemq.broker.jmx; package org.apache.activemq.broker.jmx;
import javax.management.openmbean.CompositeData;
import javax.management.openmbean.OpenDataException;
import javax.management.openmbean.TabularData;
public interface TopicViewMBean { public interface TopicViewMBean {
@ -28,4 +32,7 @@ public interface TopicViewMBean {
public long getMessages(); public long getMessages();
public long getMessagesCached(); public long getMessagesCached();
public CompositeData[] browse() throws OpenDataException;
public TabularData browseAsTable() throws OpenDataException;
} }

View File

@ -52,4 +52,6 @@ public interface Destination extends Service {
DestinationStatistics getDestinationStatistics(); DestinationStatistics getDestinationStatistics();
MessageStore getMessageStore(); MessageStore getMessageStore();
DeadLetterStrategy getDeadLetterStrategy(); DeadLetterStrategy getDeadLetterStrategy();
public Message[] browse();
} }

View File

@ -98,6 +98,9 @@ public class Queue implements Destination {
public void recoverMessageReference(String messageReference) throws Throwable { public void recoverMessageReference(String messageReference) throws Throwable {
throw new RuntimeException("Should not be called."); throw new RuntimeException("Should not be called.");
} }
public void finished(){
}
}); });
} }
} }

View File

@ -17,7 +17,7 @@
package org.apache.activemq.broker.region; package org.apache.activemq.broker.region;
import java.io.IOException; import java.io.IOException;
import java.util.Set;
import org.apache.activemq.advisory.AdvisorySupport; import org.apache.activemq.advisory.AdvisorySupport;
import org.apache.activemq.broker.ConnectionContext; import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.policy.DeadLetterStrategy; import org.apache.activemq.broker.region.policy.DeadLetterStrategy;
@ -41,9 +41,11 @@ import org.apache.activemq.thread.TaskRunnerFactory;
import org.apache.activemq.thread.Valve; import org.apache.activemq.thread.Valve;
import org.apache.activemq.transaction.Synchronization; import org.apache.activemq.transaction.Synchronization;
import org.apache.activemq.util.SubscriptionKey; import org.apache.activemq.util.SubscriptionKey;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentHashMap; import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentHashMap;
import edu.emory.mathcs.backport.java.util.concurrent.CopyOnWriteArrayList; import edu.emory.mathcs.backport.java.util.concurrent.CopyOnWriteArrayList;
import edu.emory.mathcs.backport.java.util.concurrent.CopyOnWriteArraySet;
/** /**
* The Topic is a destination that sends a copy of a message to every active * The Topic is a destination that sends a copy of a message to every active
@ -52,7 +54,7 @@ import edu.emory.mathcs.backport.java.util.concurrent.CopyOnWriteArrayList;
* @version $Revision: 1.21 $ * @version $Revision: 1.21 $
*/ */
public class Topic implements Destination { public class Topic implements Destination {
private static final Log log = LogFactory.getLog(Topic.class);
protected final ActiveMQDestination destination; protected final ActiveMQDestination destination;
protected final CopyOnWriteArrayList consumers = new CopyOnWriteArrayList(); protected final CopyOnWriteArrayList consumers = new CopyOnWriteArrayList();
protected final Valve dispatchValve = new Valve(true); protected final Valve dispatchValve = new Valve(true);
@ -196,6 +198,9 @@ public class Topic implements Destination {
public void recoverMessageReference(String messageReference) throws Throwable { public void recoverMessageReference(String messageReference) throws Throwable {
throw new RuntimeException("Should not be called."); throw new RuntimeException("Should not be called.");
} }
public void finished(){
}
}); });
if( true && subscription.getConsumerInfo().isRetroactive() ) { if( true && subscription.getConsumerInfo().isRetroactive() ) {
@ -291,6 +296,30 @@ public class Topic implements Destination {
this.subscriptionRecoveryPolicy.stop(); this.subscriptionRecoveryPolicy.stop();
} }
public Message[] browse(){
final Set result=new CopyOnWriteArraySet();
try{
store.recover(new MessageRecoveryListener(){
public void recoverMessage(Message message) throws Throwable{
result.add(message);
}
public void recoverMessageReference(String messageReference) throws Throwable{}
public void finished(){}
});
Message[] msgs=subscriptionRecoveryPolicy.browse(getActiveMQDestination());
if(msgs!=null){
for(int i=0;i<msgs.length;i++){
result.add(msgs[i]);
}
}
}catch(Throwable e){
log.warn("Failed to browse Topic: "+getActiveMQDestination().getPhysicalName(),e);
}
return (Message[]) result.toArray(new Message[result.size()]);
}
// Properties // Properties
// ------------------------------------------------------------------------- // -------------------------------------------------------------------------

View File

@ -2,95 +2,111 @@
* *
* Copyright 2005-2006 The Apache Software Foundation * Copyright 2005-2006 The Apache Software Foundation
* *
* Licensed under the Apache License, Version 2.0 (the "License"); * Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance with
* you may not use this file except in compliance with the License. * the License. You may obtain a copy of the License at
* 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.policy; package org.apache.activemq.broker.region.policy;
import java.util.ArrayList;
import java.util.List;
import org.apache.activemq.broker.ConnectionContext; import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.MessageReference; import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.broker.region.Subscription; import org.apache.activemq.broker.region.Subscription;
import org.apache.activemq.broker.region.Topic; import org.apache.activemq.broker.region.Topic;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.Message;
import org.apache.activemq.filter.DestinationFilter;
import org.apache.activemq.filter.MessageEvaluationContext; import org.apache.activemq.filter.MessageEvaluationContext;
/** /**
* This implementation of {@link SubscriptionRecoveryPolicy} will only keep * This implementation of {@link SubscriptionRecoveryPolicy} will only keep the last message.
* the last message.
* *
* @org.apache.xbean.XBean * @org.apache.xbean.XBean
* *
* @version $Revision$ * @version $Revision$
*/ */
public class FixedCountSubscriptionRecoveryPolicy implements SubscriptionRecoveryPolicy { public class FixedCountSubscriptionRecoveryPolicy implements SubscriptionRecoveryPolicy{
volatile private MessageReference messages[]; volatile private MessageReference messages[];
private int maximumSize=100; private int maximumSize=100;
private int tail=0; private int tail=0;
synchronized public boolean add(ConnectionContext context, MessageReference node) throws Throwable { synchronized public boolean add(ConnectionContext context,MessageReference node) throws Throwable{
messages[tail++] = node; messages[tail++]=node;
if( tail >= messages.length ) if(tail>=messages.length)
tail = 0; tail=0;
return true; return true;
} }
synchronized public void recover(ConnectionContext context, Topic topic, Subscription sub) throws Throwable { synchronized public void recover(ConnectionContext context,Topic topic,Subscription sub) throws Throwable{
// Re-dispatch the last message seen. // Re-dispatch the last message seen.
int t = tail; int t=tail;
// The buffer may not have rolled over yet..., start from the front // The buffer may not have rolled over yet..., start from the front
if( messages[t]==null ) if(messages[t]==null)
t=0; t=0;
// Well the buffer is really empty then. // Well the buffer is really empty then.
if( messages[t]==null ) if(messages[t]==null)
return; return;
// Keep dispatching until t hit's tail again. // Keep dispatching until t hit's tail again.
MessageEvaluationContext msgContext = context.getMessageEvaluationContext(); MessageEvaluationContext msgContext=context.getMessageEvaluationContext();
do { do{
MessageReference node = messages[t]; MessageReference node=messages[t];
try { try{
msgContext.setDestination(node.getRegionDestination().getActiveMQDestination()); msgContext.setDestination(node.getRegionDestination().getActiveMQDestination());
msgContext.setMessageReference(node); msgContext.setMessageReference(node);
if (sub.matches(node, msgContext)) { if(sub.matches(node,msgContext)){
sub.add(node); sub.add(node);
} }
} finally { }finally{
msgContext.clear(); msgContext.clear();
} }
t++; t++;
if( t >= messages.length ) if(t>=messages.length)
t = 0; t=0;
} while( t!=tail ); }while(t!=tail);
} }
public void start() throws Exception { public void start() throws Exception{
messages = new MessageReference[maximumSize]; messages=new MessageReference[maximumSize];
} }
public void stop() throws Exception { public void stop() throws Exception{
messages = null; messages=null;
} }
public int getMaximumSize() { public int getMaximumSize(){
return maximumSize; return maximumSize;
} }
/** /**
* Sets the maximum number of messages that this destination will hold around in RAM * Sets the maximum number of messages that this destination will hold around in RAM
*/ */
public void setMaximumSize(int maximumSize) { public void setMaximumSize(int maximumSize){
this.maximumSize = maximumSize; this.maximumSize=maximumSize;
} }
public Message[] browse(ActiveMQDestination destination) throws Throwable{
List result=new ArrayList();
DestinationFilter filter=DestinationFilter.parseFilter(destination);
int t=tail;
if(messages[t]==null)
t=0;
if(messages[t]!=null){
do{
MessageReference ref=messages[t];
Message message=ref.getMessage();
if(filter.matches(message.getDestination())){
result.add(message);
}
t++;
if(t>=messages.length)
t=0;
}while(t!=tail);
}
return (Message[]) result.toArray(new Message[result.size()]);
}
} }

View File

@ -20,11 +20,16 @@ import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.MessageReference; import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.broker.region.Subscription; import org.apache.activemq.broker.region.Subscription;
import org.apache.activemq.broker.region.Topic; import org.apache.activemq.broker.region.Topic;
import org.apache.activemq.broker.region.policy.TimedSubscriptionRecoveryPolicy.TimestampWrapper;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.Message;
import org.apache.activemq.filter.DestinationFilter;
import org.apache.activemq.filter.MessageEvaluationContext; import org.apache.activemq.filter.MessageEvaluationContext;
import org.apache.activemq.memory.list.DestinationBasedMessageList; import org.apache.activemq.memory.list.DestinationBasedMessageList;
import org.apache.activemq.memory.list.MessageList; import org.apache.activemq.memory.list.MessageList;
import org.apache.activemq.memory.list.SimpleMessageList; import org.apache.activemq.memory.list.SimpleMessageList;
import java.util.ArrayList;
import java.util.Iterator; import java.util.Iterator;
import java.util.List; import java.util.List;
@ -105,6 +110,10 @@ public class FixedSizedSubscriptionRecoveryPolicy implements SubscriptionRecover
this.useSharedBuffer = useSharedBuffer; this.useSharedBuffer = useSharedBuffer;
} }
public Message[] browse(ActiveMQDestination destination) throws Throwable{
return buffer.browse(destination);
}
// Implementation methods // Implementation methods
// ------------------------------------------------------------------------- // -------------------------------------------------------------------------

View File

@ -16,10 +16,15 @@
*/ */
package org.apache.activemq.broker.region.policy; package org.apache.activemq.broker.region.policy;
import java.util.ArrayList;
import java.util.List;
import org.apache.activemq.broker.ConnectionContext; import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.MessageReference; import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.broker.region.Subscription; import org.apache.activemq.broker.region.Subscription;
import org.apache.activemq.broker.region.Topic; import org.apache.activemq.broker.region.Topic;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.Message;
import org.apache.activemq.filter.DestinationFilter;
import org.apache.activemq.filter.MessageEvaluationContext; import org.apache.activemq.filter.MessageEvaluationContext;
/** /**
@ -62,4 +67,13 @@ public class LastImageSubscriptionRecoveryPolicy implements SubscriptionRecovery
public void stop() throws Exception { public void stop() throws Exception {
} }
public Message[] browse(ActiveMQDestination destination) throws Throwable{
List result = new ArrayList();
DestinationFilter filter=DestinationFilter.parseFilter(destination);
if (filter.matches(lastImage.getMessage().getDestination())){
result.add(lastImage.getMessage());
}
return (Message[])result.toArray(new Message[result.size()]);
}
} }

View File

@ -20,6 +20,8 @@ import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.MessageReference; import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.broker.region.Subscription; import org.apache.activemq.broker.region.Subscription;
import org.apache.activemq.broker.region.Topic; import org.apache.activemq.broker.region.Topic;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.Message;
/** /**
* This is the default Topic recovery policy which does not recover any messages. * This is the default Topic recovery policy which does not recover any messages.
@ -43,4 +45,8 @@ public class NoSubscriptionRecoveryPolicy implements SubscriptionRecoveryPolicy
public void stop() throws Exception { public void stop() throws Exception {
} }
public Message[] browse(ActiveMQDestination dest) throws Throwable{
return new Message[0];
}
} }

View File

@ -99,6 +99,10 @@ public class QueryBasedSubscriptionRecoveryPolicy implements SubscriptionRecover
this.query = query; this.query = query;
} }
public org.apache.activemq.command.Message[] browse(ActiveMQDestination dest) throws Throwable{
return new org.apache.activemq.command.Message[0];
}
protected void dispatchInitialMessage(Message message, Destination regionDestination, MessageEvaluationContext msgContext, Subscription sub) { protected void dispatchInitialMessage(Message message, Destination regionDestination, MessageEvaluationContext msgContext, Subscription sub) {
try { try {
ActiveMQMessage activeMessage = ActiveMQMessageTransformation.transformMessage(message, null); ActiveMQMessage activeMessage = ActiveMQMessageTransformation.transformMessage(message, null);

View File

@ -16,11 +16,14 @@
*/ */
package org.apache.activemq.broker.region.policy; package org.apache.activemq.broker.region.policy;
import org.apache.activemq.Service; import org.apache.activemq.Service;
import org.apache.activemq.broker.ConnectionContext; import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.MessageReference; import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.broker.region.Subscription; import org.apache.activemq.broker.region.Subscription;
import org.apache.activemq.broker.region.Topic; import org.apache.activemq.broker.region.Topic;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.Message;
/** /**
* Abstraction to allow different recovery policies to be plugged * Abstraction to allow different recovery policies to be plugged
@ -35,8 +38,9 @@ public interface SubscriptionRecoveryPolicy extends Service {
* A message was sent to the destination. * A message was sent to the destination.
* *
* @param context * @param context
* @param message
* @param node * @param node
* @return TODO * @return true if successful
* @throws Throwable * @throws Throwable
*/ */
boolean add(ConnectionContext context, MessageReference message) throws Throwable; boolean add(ConnectionContext context, MessageReference message) throws Throwable;
@ -45,11 +49,19 @@ public interface SubscriptionRecoveryPolicy extends Service {
* Let a subscription recover message held by the policy. * Let a subscription recover message held by the policy.
* *
* @param context * @param context
* @param topic TODO
* @param topic * @param topic
* @param sub
* @param node * @param node
* @throws Throwable * @throws Throwable
*/ */
void recover(ConnectionContext context, Topic topic, Subscription sub) throws Throwable; void recover(ConnectionContext context, Topic topic, Subscription sub) throws Throwable;
/**
* @param dest
* @return messages
* @throws Throwable
*/
Message[] browse(ActiveMQDestination dest) throws Throwable;
} }

View File

@ -16,18 +16,20 @@
*/ */
package org.apache.activemq.broker.region.policy; package org.apache.activemq.broker.region.policy;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.broker.region.Subscription;
import org.apache.activemq.broker.region.Topic;
import org.apache.activemq.filter.MessageEvaluationContext;
import org.apache.activemq.thread.Scheduler;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Collections; import java.util.Collections;
import java.util.Iterator; import java.util.Iterator;
import java.util.LinkedList; import java.util.LinkedList;
import java.util.List; import java.util.List;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.broker.region.Subscription;
import org.apache.activemq.broker.region.Topic;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.Message;
import org.apache.activemq.filter.DestinationFilter;
import org.apache.activemq.filter.MessageEvaluationContext;
import org.apache.activemq.thread.Scheduler;
/** /**
* This implementation of {@link SubscriptionRecoveryPolicy} will keep a timed * This implementation of {@link SubscriptionRecoveryPolicy} will keep a timed
@ -123,4 +125,19 @@ public class TimedSubscriptionRecoveryPolicy implements SubscriptionRecoveryPoli
this.recoverDuration = recoverDuration; this.recoverDuration = recoverDuration;
} }
public Message[] browse(ActiveMQDestination destination) throws Throwable{
List result = new ArrayList();
ArrayList copy = new ArrayList(buffer);
DestinationFilter filter=DestinationFilter.parseFilter(destination);
for (Iterator iter = copy.iterator(); iter.hasNext();) {
TimestampWrapper timestampWrapper = (TimestampWrapper) iter.next();
MessageReference ref = timestampWrapper.message;
Message message=ref.getMessage();
if (filter.matches(message.getDestination())){
result.add(message);
}
}
return (Message[]) result.toArray(new Message[result.size()]);
}
} }

View File

@ -22,11 +22,11 @@ import java.util.Iterator;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.Set; import java.util.Set;
import org.apache.activemq.broker.region.MessageReference; import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.broker.region.Subscription; import org.apache.activemq.broker.region.Subscription;
import org.apache.activemq.command.ActiveMQDestination; import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQMessage; import org.apache.activemq.command.ActiveMQMessage;
import org.apache.activemq.command.Message;
import org.apache.activemq.filter.DestinationMap; import org.apache.activemq.filter.DestinationMap;
import org.apache.activemq.memory.buffer.MessageBuffer; import org.apache.activemq.memory.buffer.MessageBuffer;
import org.apache.activemq.memory.buffer.MessageQueue; import org.apache.activemq.memory.buffer.MessageQueue;
@ -70,9 +70,13 @@ public class DestinationBasedMessageList implements MessageList {
} }
public List getMessages(Subscription sub) { public List getMessages(Subscription sub) {
return getMessages(sub.getConsumerInfo().getDestination());
}
protected List getMessages(ActiveMQDestination destination) {
Set set = null; Set set = null;
synchronized (lock) { synchronized (lock) {
set = subscriptionIndex.get(sub.getConsumerInfo().getDestination()); set = subscriptionIndex.get(destination);
} }
List answer = new ArrayList(); List answer = new ArrayList();
for (Iterator iter = set.iterator(); iter.hasNext();) { for (Iterator iter = set.iterator(); iter.hasNext();) {
@ -82,6 +86,12 @@ public class DestinationBasedMessageList implements MessageList {
return answer; return answer;
} }
public Message[] browse(ActiveMQDestination destination) {
List result = getMessages(destination);
return (Message[])result.toArray(new Message[result.size()]);
}
public void clear() { public void clear() {
messageBuffer.clear(); messageBuffer.clear();
} }

View File

@ -18,6 +18,8 @@ package org.apache.activemq.memory.list;
import org.apache.activemq.broker.region.MessageReference; import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.broker.region.Subscription; import org.apache.activemq.broker.region.Subscription;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.Message;
import java.util.List; import java.util.List;
@ -36,5 +38,13 @@ public interface MessageList {
*/ */
List getMessages(Subscription sub); List getMessages(Subscription sub);
/**
* @param destination
* @return an array of Messages that match the destination
*/
Message[] browse(ActiveMQDestination destination);
void clear(); void clear();
} }

View File

@ -18,8 +18,16 @@ package org.apache.activemq.memory.list;
import org.apache.activemq.broker.region.MessageReference; import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.broker.region.Subscription; import org.apache.activemq.broker.region.Subscription;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.Message;
import org.apache.activemq.filter.DestinationFilter;
import org.apache.activemq.network.DemandForwardingBridge;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.io.IOException;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedList; import java.util.LinkedList;
import java.util.List; import java.util.List;
@ -32,7 +40,7 @@ import java.util.List;
* @version $Revision: 1.1 $ * @version $Revision: 1.1 $
*/ */
public class SimpleMessageList implements MessageList { public class SimpleMessageList implements MessageList {
static final private Log log=LogFactory.getLog(SimpleMessageList.class);
private LinkedList list = new LinkedList(); private LinkedList list = new LinkedList();
private int maximumSize = 100 * 64 * 1024; private int maximumSize = 100 * 64 * 1024;
private int size; private int size;
@ -61,6 +69,27 @@ public class SimpleMessageList implements MessageList {
return getList(); return getList();
} }
public Message[] browse(ActiveMQDestination destination) {
List result = new ArrayList();
DestinationFilter filter=DestinationFilter.parseFilter(destination);
synchronized(lock){
for (Iterator i = list.iterator(); i.hasNext();){
MessageReference ref = (MessageReference)i.next();
Message msg;
try{
msg=ref.getMessage();
if (filter.matches(msg.getDestination())){
result.add(msg);
}
}catch(IOException e){
log.error("Failed to get Message from MessageReference: " + ref,e);
}
}
}
return (Message[])result.toArray(new Message[result.size()]);
}
/** /**
* Returns a copy of the list * Returns a copy of the list
*/ */

View File

@ -24,4 +24,5 @@ import org.apache.activemq.command.Message;
public interface MessageRecoveryListener { public interface MessageRecoveryListener {
void recoverMessage(Message message) throws Throwable; void recoverMessage(Message message) throws Throwable;
void recoverMessageReference(String messageReference) throws Throwable; void recoverMessageReference(String messageReference) throws Throwable;
void finished();
} }

View File

@ -25,4 +25,5 @@ import java.io.IOException;
public interface JDBCMessageRecoveryListener { public interface JDBCMessageRecoveryListener {
void recoverMessage(long sequenceId, byte[] message) throws IOException, Throwable; void recoverMessage(long sequenceId, byte[] message) throws IOException, Throwable;
void recoverMessageReference(String reference) throws IOException, Throwable; void recoverMessageReference(String reference) throws IOException, Throwable;
void finished();
} }

View File

@ -158,6 +158,9 @@ public class JDBCMessageStore implements MessageStore {
public void recoverMessageReference(String reference) throws IOException, Throwable { public void recoverMessageReference(String reference) throws IOException, Throwable {
listener.recoverMessageReference(reference); listener.recoverMessageReference(reference);
} }
public void finished(){
listener.finished();
}
}); });
} catch (SQLException e) { } catch (SQLException e) {
JDBCPersistenceAdapter.log("JDBC Failure: ",e); JDBCPersistenceAdapter.log("JDBC Failure: ",e);

View File

@ -75,6 +75,10 @@ public class JDBCTopicMessageStore extends JDBCMessageStore implements TopicMess
public void recoverMessageReference(String reference) throws IOException, Throwable { public void recoverMessageReference(String reference) throws IOException, Throwable {
listener.recoverMessageReference(reference); listener.recoverMessageReference(reference);
} }
public void finished(){
listener.finished();
}
}); });
} catch (SQLException e) { } catch (SQLException e) {
JDBCPersistenceAdapter.log("JDBC Failure: ",e); JDBCPersistenceAdapter.log("JDBC Failure: ",e);

View File

@ -337,11 +337,11 @@ public class DefaultJDBCAdapter implements JDBCAdapter {
listener.recoverMessage(rs.getLong(1), getBinaryData(rs, 2)); listener.recoverMessage(rs.getLong(1), getBinaryData(rs, 2));
} }
} }
} }
finally { finally {
close(rs); close(rs);
close(s); close(s);
listener.finished();
} }
} }
@ -402,7 +402,9 @@ public class DefaultJDBCAdapter implements JDBCAdapter {
finally { finally {
close(rs); close(rs);
close(s); close(s);
listener.finished();
} }
} }
/** /**

View File

@ -360,6 +360,10 @@ public class QuickJournalMessageStore implements MessageStore {
Message message = (Message) peristenceAdapter.readCommand(loc); Message message = (Message) peristenceAdapter.readCommand(loc);
listener.recoverMessage(message); listener.recoverMessage(message);
} }
public void finished(){
listener.finished();
}
}); });
} }

View File

@ -63,6 +63,10 @@ public class QuickJournalTopicMessageStore extends QuickJournalMessageStore impl
Message message = (Message) peristenceAdapter.readCommand(loc); Message message = (Message) peristenceAdapter.readCommand(loc);
listener.recoverMessage(message); listener.recoverMessage(message);
} }
public void finished(){
listener.finished();
}
}); });
} }

View File

@ -82,6 +82,7 @@ public class MemoryMessageStore implements MessageStore {
listener.recoverMessage((Message) msg); listener.recoverMessage((Message) msg);
} }
} }
listener.finished();
} }
} }

View File

@ -107,6 +107,7 @@ public class MemoryTopicMessageStore extends MemoryMessageStore implements Topic
pastLastAck=entry.getKey().equals(lastAck); pastLastAck=entry.getKey().equals(lastAck);
} }
} }
listener.finished();
} }
} }