Author: jstrachan
Date: Mon Dec 19 07:54:17 2005
New Revision: 357712
URL: http://svn.apache.org/viewcvs?rev=357712&view=rev
Log:
added a helper class and strategy method for people wishing to implement their
own custom MessageQuery strategies to be fired as a new subscriber is created
Added:
incubator/activemq/trunk/activemq-core/src/main/java/org/activemq/broker/region/policy/MessageQuery.java
(with props)
incubator/activemq/trunk/activemq-core/src/main/java/org/activemq/broker/region/policy/QueryBasedSubscriptionRecoveryPolicy.java
(with props)
Added:
incubator/activemq/trunk/activemq-core/src/main/java/org/activemq/broker/region/policy/MessageQuery.java
URL:
http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/activemq/broker/region/policy/MessageQuery.java?rev=357712&view=auto
==============================================================================
---
incubator/activemq/trunk/activemq-core/src/main/java/org/activemq/broker/region/policy/MessageQuery.java
(added)
+++
incubator/activemq/trunk/activemq-core/src/main/java/org/activemq/broker/region/policy/MessageQuery.java
Mon Dec 19 07:54:17 2005
@@ -0,0 +1,39 @@
+/**
+ *
+ * Copyright 2005 LogicBlaze, Inc. http://www.logicblaze.com
+ *
+ * Licensed under the Apache License, Version 2.0 (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
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ **/
+package org.activemq.broker.region.policy;
+
+import org.activemq.command.ActiveMQDestination;
+
+import javax.jms.MessageListener;
+
+/**
+ * Represents some kind of query which will load messages from some source.
+ *
+ * @version $Revision$
+ */
+public interface MessageQuery {
+
+ /**
+ * Executes the query for messages; each message is passed into the
listener
+ *
+ * @param destination the destination on which the query is to be performed
+ * @param listener is the listener to notify as each message is created or
loaded
+ */
+ public void execute(ActiveMQDestination destination, MessageListener
listener);
+
+}
Propchange:
incubator/activemq/trunk/activemq-core/src/main/java/org/activemq/broker/region/policy/MessageQuery.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
incubator/activemq/trunk/activemq-core/src/main/java/org/activemq/broker/region/policy/MessageQuery.java
------------------------------------------------------------------------------
svn:keywords = Date Author Id Revision HeadURL
Propchange:
incubator/activemq/trunk/activemq-core/src/main/java/org/activemq/broker/region/policy/MessageQuery.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
Added:
incubator/activemq/trunk/activemq-core/src/main/java/org/activemq/broker/region/policy/QueryBasedSubscriptionRecoveryPolicy.java
URL:
http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/activemq/broker/region/policy/QueryBasedSubscriptionRecoveryPolicy.java?rev=357712&view=auto
==============================================================================
---
incubator/activemq/trunk/activemq-core/src/main/java/org/activemq/broker/region/policy/QueryBasedSubscriptionRecoveryPolicy.java
(added)
+++
incubator/activemq/trunk/activemq-core/src/main/java/org/activemq/broker/region/policy/QueryBasedSubscriptionRecoveryPolicy.java
Mon Dec 19 07:54:17 2005
@@ -0,0 +1,101 @@
+/**
+ * <a href="http://activemq.org";>ActiveMQ: The Open Source Message Fabric</a>
+ *
+ * Copyright 2005 (C) LogicBlaze, Inc. http://www.logicblaze.com
+ *
+ * Licensed under the Apache License, Version 2.0 (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
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ **/
+package org.activemq.broker.region.policy;
+
+import org.activemq.ActiveMQMessageTransformation;
+import org.activemq.broker.ConnectionContext;
+import org.activemq.broker.region.MessageReference;
+import org.activemq.broker.region.Subscription;
+import org.activemq.command.ActiveMQDestination;
+import org.activemq.command.ActiveMQMessage;
+import org.activemq.filter.MessageEvaluationContext;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+
+import javax.jms.Message;
+import javax.jms.MessageListener;
+
+/**
+ * This implementation of [EMAIL PROTECTED] SubscriptionRecoveryPolicy} will
perform a user
+ * specific query mechanism to load any messages they may have missed.
+ *
+ * @org.xbean.XBean
+ *
+ * @version $Revision$
+ */
+public class QueryBasedSubscriptionRecoveryPolicy implements
SubscriptionRecoveryPolicy {
+
+ private static final Log log =
LogFactory.getLog(QueryBasedSubscriptionRecoveryPolicy.class);
+
+ private MessageQuery query;
+
+ public void add(ConnectionContext context, MessageReference message)
throws Throwable {
+ }
+
+ public void recover(ConnectionContext context, final Subscription sub)
throws Throwable {
+ if (query != null) {
+ final MessageEvaluationContext msgContext =
context.getMessageEvaluationContext();
+ try {
+ ActiveMQDestination destination =
sub.getConsumerInfo().getDestination();
+ query.execute(destination, new MessageListener() {
+ public void onMessage(Message message) {
+ dispatchInitialMessage(message, msgContext, sub);
+ }
+ });
+ }
+ finally {
+ msgContext.clear();
+ }
+ }
+ }
+
+ public void start() throws Exception {
+ if (query != null) {
+ throw new IllegalArgumentException("No query property configured");
+ }
+ }
+
+ public void stop() throws Exception {
+ }
+
+ public MessageQuery getQuery() {
+ return query;
+ }
+
+ /**
+ * Sets the query strategy to load initial messages
+ */
+ public void setQuery(MessageQuery query) {
+ this.query = query;
+ }
+
+ protected void dispatchInitialMessage(Message message,
MessageEvaluationContext msgContext, Subscription sub) {
+ try {
+ ActiveMQMessage activeMessage =
ActiveMQMessageTransformation.transformMessage(message, null);
+ msgContext.setDestination(activeMessage.getDestination());
+ msgContext.setMessageReference(activeMessage);
+ if (sub.matches(activeMessage, msgContext)) {
+ sub.add(activeMessage);
+ }
+ }
+ catch (Throwable e) {
+ log.warn("Failed to dispatch initial message: " + message + " into
subscription. Reason: " + e, e);
+ }
+ }
+}
Propchange:
incubator/activemq/trunk/activemq-core/src/main/java/org/activemq/broker/region/policy/QueryBasedSubscriptionRecoveryPolicy.java
------------------------------------------------------------------------------
svn:eol-style = native
Propchange:
incubator/activemq/trunk/activemq-core/src/main/java/org/activemq/broker/region/policy/QueryBasedSubscriptionRecoveryPolicy.java
------------------------------------------------------------------------------
svn:keywords = Date Author Id Revision HeadURL
Propchange:
incubator/activemq/trunk/activemq-core/src/main/java/org/activemq/broker/region/policy/QueryBasedSubscriptionRecoveryPolicy.java
------------------------------------------------------------------------------
svn:mime-type = text/plain
