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

http://issues.apache.org/activemq/browse/AMQ-714 

git-svn-id: https://svn.apache.org/repos/asf/incubator/activemq/trunk@413674 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Hiram R. Chirino 2006-06-12 15:08:33 +00:00
parent 855076cee3
commit c1a0cd6875
8 changed files with 42 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
activemq-core/src/main/java/org/apache/activemq/broker/region/policy/FixedCountSubscriptionRecoveryPolicy.java
View File

@ -34,6 +34,12 @@ public class FixedCountSubscriptionRecoveryPolicy implements SubscriptionRecover
volatile private MessageReference messages[];
private int maximumSize=100;
private int tail=0;
public SubscriptionRecoveryPolicy copy() {
FixedCountSubscriptionRecoveryPolicy rc = new FixedCountSubscriptionRecoveryPolicy();
rc.setMaximumSize(maximumSize);
return rc;
}
synchronized public boolean add(ConnectionContext context,MessageReference node) throws Exception{
messages[tail++]=node;
@ -109,4 +115,5 @@ public class FixedCountSubscriptionRecoveryPolicy implements SubscriptionRecover
}
return (Message[]) result.toArray(new Message[result.size()]);
}
}

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

@ -48,6 +48,13 @@ public class FixedSizedSubscriptionRecoveryPolicy implements SubscriptionRecover
private int maximumSize = 100 * 64 * 1024;
private boolean useSharedBuffer = true;
public SubscriptionRecoveryPolicy copy() {
FixedSizedSubscriptionRecoveryPolicy rc = new FixedSizedSubscriptionRecoveryPolicy();
rc.setMaximumSize(maximumSize);
rc.setUseSharedBuffer(useSharedBuffer);
return rc;
}
public boolean add(ConnectionContext context, MessageReference message) throws Exception {
buffer.add(message);
return true;
@ -125,4 +132,5 @@ public class FixedSizedSubscriptionRecoveryPolicy implements SubscriptionRecover
return new DestinationBasedMessageList(maximumSize);
}
}
}

5
activemq-core/src/main/java/org/apache/activemq/broker/region/policy/LastImageSubscriptionRecoveryPolicy.java
View File

@ -18,6 +18,7 @@ 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.region.MessageReference;
import org.apache.activemq.broker.region.Subscription;
@ -76,4 +77,8 @@ public class LastImageSubscriptionRecoveryPolicy implements SubscriptionRecovery
return (Message[])result.toArray(new Message[result.size()]);
}
public SubscriptionRecoveryPolicy copy() {
return new LastImageSubscriptionRecoveryPolicy();
}
}

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

@ -32,6 +32,12 @@ import org.apache.activemq.command.Message;
*/
public class NoSubscriptionRecoveryPolicy implements SubscriptionRecoveryPolicy {
public SubscriptionRecoveryPolicy copy() {
// This object is immutable
return this;
}
public boolean add(ConnectionContext context, MessageReference node) throws Exception {
return true;
}

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

@ -65,7 +65,7 @@ public class PolicyEntry extends DestinationMapEntry {
topic.setDeadLetterStrategy(deadLetterStrategy);
}
if (subscriptionRecoveryPolicy != null) {
topic.setSubscriptionRecoveryPolicy(subscriptionRecoveryPolicy);
topic.setSubscriptionRecoveryPolicy(subscriptionRecoveryPolicy.copy());
}
topic.setSendAdvisoryIfNoConsumers(sendAdvisoryIfNoConsumers);
if( memoryLimit>0 ) {

5
activemq-core/src/main/java/org/apache/activemq/broker/region/policy/QueryBasedSubscriptionRecoveryPolicy.java
View File

@ -55,7 +55,10 @@ public class QueryBasedSubscriptionRecoveryPolicy implements SubscriptionRecover
private IdGenerator idGenerator = new IdGenerator();
private ProducerId producerId = createProducerId();
public QueryBasedSubscriptionRecoveryPolicy() {
public SubscriptionRecoveryPolicy copy() {
QueryBasedSubscriptionRecoveryPolicy rc = new QueryBasedSubscriptionRecoveryPolicy();
rc.setQuery(query);
return rc;
}
public boolean add(ConnectionContext context, MessageReference message) throws Exception {

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

@ -64,4 +64,8 @@ public interface SubscriptionRecoveryPolicy extends Service {
*/
Message[] browse(ActiveMQDestination dest) throws Exception;
/**
* Used to copy the policy object.
*/
SubscriptionRecoveryPolicy copy();
}

7
activemq-core/src/main/java/org/apache/activemq/broker/region/policy/TimedSubscriptionRecoveryPolicy.java
View File

@ -21,6 +21,7 @@ import java.util.Collections;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.broker.region.Subscription;
@ -67,6 +68,12 @@ public class TimedSubscriptionRecoveryPolicy implements SubscriptionRecoveryPoli
}
};
public SubscriptionRecoveryPolicy copy() {
TimedSubscriptionRecoveryPolicy rc = new TimedSubscriptionRecoveryPolicy();
rc.setRecoverDuration(recoverDuration);
return rc;
}
public boolean add(ConnectionContext context, MessageReference message) throws Exception {
buffer.add(new TimestampWrapper(message, lastGCRun));
return true;