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

Applying patch from http://issues.apache.org/activemq/browse/AMQ-1293 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@550449 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Robert Davies 2007-06-25 10:45:55 +00:00
parent db51eba74f
commit d36c0d4de7
71 changed files with 877 additions and 856 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -76,6 +76,7 @@ public class PropertiesBrokerFactory implements BrokerFactoryHandler {
}
}
properties.load(inputStream);
inputStream.close();
// should we append any system properties?
try {

2
activemq-core/src/main/java/org/apache/activemq/broker/jmx/ManagementContext.java
View File

@ -387,7 +387,7 @@ public class ManagementContext implements Service{
mbeanServer.registerMBean(cl.newInstance(),namingServiceObjectName);
// mbeanServer.createMBean("mx4j.tools.naming.NamingService", namingServiceObjectName, null);
// set the naming port
Attribute attr=new Attribute("Port",new Integer(connectorPort));
Attribute attr=new Attribute("Port",Integer.valueOf(connectorPort));
mbeanServer.setAttribute(namingServiceObjectName,attr);
}catch(Throwable e){
log.debug("Failed to create local registry",e);

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

@ -124,9 +124,9 @@ public class OpenTypeSupport {
rc.put("JMSReplyTo", ""+m.getJMSReplyTo());
rc.put("JMSType", m.getJMSType());
rc.put("JMSDeliveryMode", m.getJMSDeliveryMode()==DeliveryMode.PERSISTENT ? "PERSISTENT" : "NON-PERSISTENT");
rc.put("JMSExpiration", new Long(m.getJMSExpiration()));
rc.put("JMSPriority", new Integer(m.getJMSPriority()));
rc.put("JMSRedelivered", new Boolean(m.getJMSRedelivered()));
rc.put("JMSExpiration", Long.valueOf(m.getJMSExpiration()));
rc.put("JMSPriority", Integer.valueOf(m.getJMSPriority()));
rc.put("JMSRedelivered", Boolean.valueOf(m.getJMSRedelivered()));
rc.put("JMSTimestamp", new Date(m.getJMSTimestamp()));
try {
rc.put("Properties", ""+m.getProperties());
@ -155,9 +155,9 @@ public class OpenTypeSupport {
long length=0;
try {
length = m.getBodyLength();
rc.put("BodyLength", new Long(length));
rc.put("BodyLength", Long.valueOf(length));
} catch (JMSException e) {
rc.put("BodyLength", new Long(0));
rc.put("BodyLength", Long.valueOf(0));
}
try {
byte preview[] = new byte[ (int)Math.min(length, 255) ];

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

@ -119,9 +119,9 @@ public class DurableTopicSubscription extends PrefetchSubscription implements Us
MessageReference node=(MessageReference)iter.next();
Integer count=(Integer)redeliveredMessages.get(node.getMessageId());
if(count!=null){
redeliveredMessages.put(node.getMessageId(),new Integer(count.intValue()+1));
redeliveredMessages.put(node.getMessageId(),Integer.valueOf(count.intValue()+1));
}else{
redeliveredMessages.put(node.getMessageId(),new Integer(1));
redeliveredMessages.put(node.getMessageId(),Integer.valueOf(1));
}
if(keepDurableSubsActive){
synchronized(pending){

10
activemq-core/src/main/java/org/apache/activemq/command/ActiveMQMapMessage.java
View File

@ -477,7 +477,7 @@ public class ActiveMQMapMessage extends ActiveMQMessage implements MapMessage {
*/
public void setByte(String name, byte value) throws JMSException {
initializeWriting();
put(name, new Byte(value));
put(name, Byte.valueOf(value));
}
/**
@ -491,7 +491,7 @@ public class ActiveMQMapMessage extends ActiveMQMessage implements MapMessage {
*/
public void setShort(String name, short value) throws JMSException {
initializeWriting();
put(name, new Short(value));
put(name, Short.valueOf(value));
}
/**
@ -505,7 +505,7 @@ public class ActiveMQMapMessage extends ActiveMQMessage implements MapMessage {
*/
public void setChar(String name, char value) throws JMSException {
initializeWriting();
put(name, new Character(value));
put(name, Character.valueOf(value));
}
/**
@ -519,7 +519,7 @@ public class ActiveMQMapMessage extends ActiveMQMessage implements MapMessage {
*/
public void setInt(String name, int value) throws JMSException {
initializeWriting();
put(name, new Integer(value));
put(name, Integer.valueOf(value));
}
/**
@ -533,7 +533,7 @@ public class ActiveMQMapMessage extends ActiveMQMessage implements MapMessage {
*/
public void setLong(String name, long value) throws JMSException {
initializeWriting();
put(name, new Long(value));
put(name, Long.valueOf(value));
}
/**

11
activemq-core/src/main/java/org/apache/activemq/command/ActiveMQMessage.java
View File

@ -53,7 +53,6 @@ public class ActiveMQMessage extends Message implements org.apache.activemq.Mess
}
transient protected Callback acknowledgeCallback;
transient int hashCode;
public Message copy() {
ActiveMQMessage copy = new ActiveMQMessage();
@ -545,23 +544,23 @@ public class ActiveMQMessage extends Message implements org.apache.activemq.Mess
setBooleanProperty(name,value,true);
}
public void setBooleanProperty(String name, boolean value,boolean checkReadOnly) throws JMSException {
setObjectProperty(name, value ? Boolean.TRUE : Boolean.FALSE,checkReadOnly);
setObjectProperty(name, Boolean.valueOf(value), checkReadOnly);
}
public void setByteProperty(String name, byte value) throws JMSException {
setObjectProperty(name, new Byte(value));
setObjectProperty(name, Byte.valueOf(value));
}
public void setShortProperty(String name, short value) throws JMSException {
setObjectProperty(name, new Short(value));
setObjectProperty(name, Short.valueOf(value));
}
public void setIntProperty(String name, int value) throws JMSException {
setObjectProperty(name, new Integer(value));
setObjectProperty(name, Integer.valueOf(value));
}
public void setLongProperty(String name, long value) throws JMSException {
setObjectProperty(name, new Long(value));
setObjectProperty(name, Long.valueOf(value));
}
public void setFloatProperty(String name, float value) throws JMSException {

10
activemq-core/src/main/java/org/apache/activemq/command/ActiveMQStreamMessage.java
View File

@ -819,16 +819,16 @@ public class ActiveMQStreamMessage extends ActiveMQMessage implements StreamMess
return this.dataIn.readUTF();
}
if (type == MarshallingSupport.LONG_TYPE) {
return new Long(this.dataIn.readLong());
return Long.valueOf(this.dataIn.readLong());
}
if (type == MarshallingSupport.INTEGER_TYPE) {
return new Integer(this.dataIn.readInt());
return Integer.valueOf(this.dataIn.readInt());
}
if (type == MarshallingSupport.SHORT_TYPE) {
return new Short(this.dataIn.readShort());
return Short.valueOf(this.dataIn.readShort());
}
if (type == MarshallingSupport.BYTE_TYPE) {
return new Byte(this.dataIn.readByte());
return Byte.valueOf(this.dataIn.readByte());
}
if (type == MarshallingSupport.FLOAT_TYPE) {
return new Float(this.dataIn.readFloat());
@ -840,7 +840,7 @@ public class ActiveMQStreamMessage extends ActiveMQMessage implements StreamMess
return this.dataIn.readBoolean() ? Boolean.TRUE : Boolean.FALSE;
}
if (type == MarshallingSupport.CHAR_TYPE) {
return new Character(this.dataIn.readChar());
return Character.valueOf(this.dataIn.readChar());
}
if (type == MarshallingSupport.BYTE_ARRAY_TYPE) {
int len = this.dataIn.readInt();

54
activemq-core/src/main/java/org/apache/activemq/filter/ComparisonExpression.java
View File

@ -41,26 +41,26 @@ public abstract class ComparisonExpression extends BinaryExpression implements B
static final private HashSet REGEXP_CONTROL_CHARS = new HashSet();
static {
REGEXP_CONTROL_CHARS.add(new Character('.'));
REGEXP_CONTROL_CHARS.add(new Character('\\'));
REGEXP_CONTROL_CHARS.add(new Character('['));
REGEXP_CONTROL_CHARS.add(new Character(']'));
REGEXP_CONTROL_CHARS.add(new Character('^'));
REGEXP_CONTROL_CHARS.add(new Character('$'));
REGEXP_CONTROL_CHARS.add(new Character('?'));
REGEXP_CONTROL_CHARS.add(new Character('*'));
REGEXP_CONTROL_CHARS.add(new Character('+'));
REGEXP_CONTROL_CHARS.add(new Character('{'));
REGEXP_CONTROL_CHARS.add(new Character('}'));
REGEXP_CONTROL_CHARS.add(new Character('|'));
REGEXP_CONTROL_CHARS.add(new Character('('));
REGEXP_CONTROL_CHARS.add(new Character(')'));
REGEXP_CONTROL_CHARS.add(new Character(':'));
REGEXP_CONTROL_CHARS.add(new Character('&'));
REGEXP_CONTROL_CHARS.add(new Character('<'));
REGEXP_CONTROL_CHARS.add(new Character('>'));
REGEXP_CONTROL_CHARS.add(new Character('='));
REGEXP_CONTROL_CHARS.add(new Character('!'));
REGEXP_CONTROL_CHARS.add(Character.valueOf('.'));
REGEXP_CONTROL_CHARS.add(Character.valueOf('\\'));
REGEXP_CONTROL_CHARS.add(Character.valueOf('['));
REGEXP_CONTROL_CHARS.add(Character.valueOf(']'));
REGEXP_CONTROL_CHARS.add(Character.valueOf('^'));
REGEXP_CONTROL_CHARS.add(Character.valueOf('$'));
REGEXP_CONTROL_CHARS.add(Character.valueOf('?'));
REGEXP_CONTROL_CHARS.add(Character.valueOf('*'));
REGEXP_CONTROL_CHARS.add(Character.valueOf('+'));
REGEXP_CONTROL_CHARS.add(Character.valueOf('{'));
REGEXP_CONTROL_CHARS.add(Character.valueOf('}'));
REGEXP_CONTROL_CHARS.add(Character.valueOf('|'));
REGEXP_CONTROL_CHARS.add(Character.valueOf('('));
REGEXP_CONTROL_CHARS.add(Character.valueOf(')'));
REGEXP_CONTROL_CHARS.add(Character.valueOf(':'));
REGEXP_CONTROL_CHARS.add(Character.valueOf('&'));
REGEXP_CONTROL_CHARS.add(Character.valueOf('<'));
REGEXP_CONTROL_CHARS.add(Character.valueOf('>'));
REGEXP_CONTROL_CHARS.add(Character.valueOf('='));
REGEXP_CONTROL_CHARS.add(Character.valueOf('!'));
}
static class LikeExpression extends UnaryExpression implements BooleanExpression {
@ -354,13 +354,13 @@ public abstract class ComparisonExpression extends BinaryExpression implements B
if (lc != rc) {
if (lc == Byte.class) {
if (rc == Short.class) {
lv = new Short(((Number) lv).shortValue());
lv = Short.valueOf(((Number) lv).shortValue());
}
else if (rc == Integer.class) {
lv = new Integer(((Number) lv).intValue());
lv = Integer.valueOf(((Number) lv).intValue());
}
else if (rc == Long.class) {
lv = new Long(((Number) lv).longValue());
lv = Long.valueOf(((Number) lv).longValue());
}
else if (rc == Float.class) {
lv = new Float(((Number) lv).floatValue());
@ -373,10 +373,10 @@ public abstract class ComparisonExpression extends BinaryExpression implements B
}
} else if (lc == Short.class) {
if (rc == Integer.class) {
lv = new Integer(((Number) lv).intValue());
lv = Integer.valueOf(((Number) lv).intValue());
}
else if (rc == Long.class) {
lv = new Long(((Number) lv).longValue());
lv = Long.valueOf(((Number) lv).longValue());
}
else if (rc == Float.class) {
lv = new Float(((Number) lv).floatValue());
@ -389,7 +389,7 @@ public abstract class ComparisonExpression extends BinaryExpression implements B
}
} else if (lc == Integer.class) {
if (rc == Long.class) {
lv = new Long(((Number) lv).longValue());
lv = Long.valueOf(((Number) lv).longValue());
}
else if (rc == Float.class) {
lv = new Float(((Number) lv).floatValue());
@ -403,7 +403,7 @@ public abstract class ComparisonExpression extends BinaryExpression implements B
}
else if (lc == Long.class) {
if (rc == Integer.class) {
rv = new Long(((Number) rv).longValue());
rv = Long.valueOf(((Number) rv).longValue());
}
else if (rc == Float.class) {
lv = new Float(((Number) lv).floatValue());

10
activemq-core/src/main/java/org/apache/activemq/filter/ConstantExpression.java
View File

@ -60,25 +60,25 @@ public class ConstantExpression implements Expression {
long l = value.longValue();
if (Integer.MIN_VALUE <= l && l <= Integer.MAX_VALUE) {
value = new Integer(value.intValue());
value = Integer.valueOf(value.intValue());
}
return new ConstantExpression(value);
}
public static ConstantExpression createFromHex(String text) {
Number value = new Long(Long.parseLong(text.substring(2), 16));
Number value = Long.valueOf(Long.parseLong(text.substring(2), 16));
long l = value.longValue();
if (Integer.MIN_VALUE <= l && l <= Integer.MAX_VALUE) {
value = new Integer(value.intValue());
value = Integer.valueOf(value.intValue());
}
return new ConstantExpression(value);
}
public static ConstantExpression createFromOctal(String text) {
Number value = new Long(Long.parseLong(text, 8));
Number value = Long.valueOf(Long.parseLong(text, 8));
long l = value.longValue();
if (Integer.MIN_VALUE <= l && l <= Integer.MAX_VALUE) {
value = new Integer(value.intValue());
value = Integer.valueOf(value.intValue());
}
return new ConstantExpression(value);
}

16
activemq-core/src/main/java/org/apache/activemq/filter/PropertyExpression.java
View File

@ -66,12 +66,12 @@ public class PropertyExpression implements Expression {
});
JMS_PROPERTY_EXPRESSIONS.put("JMSDeliveryMode", new SubExpression() {
public Object evaluate(Message message) {
return new Integer(message.isPersistent() ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT );
return Integer.valueOf(message.isPersistent() ? DeliveryMode.PERSISTENT : DeliveryMode.NON_PERSISTENT );
}
});
JMS_PROPERTY_EXPRESSIONS.put("JMSPriority", new SubExpression() {
public Object evaluate(Message message) {
return new Integer(message.getPriority());
return Integer.valueOf(message.getPriority());
}
});
JMS_PROPERTY_EXPRESSIONS.put("JMSMessageID", new SubExpression() {
@ -83,7 +83,7 @@ public class PropertyExpression implements Expression {
});
JMS_PROPERTY_EXPRESSIONS.put("JMSTimestamp", new SubExpression() {
public Object evaluate(Message message) {
return new Long(message.getTimestamp());
return Long.valueOf(message.getTimestamp());
}
});
JMS_PROPERTY_EXPRESSIONS.put("JMSCorrelationID", new SubExpression() {
@ -93,28 +93,28 @@ public class PropertyExpression implements Expression {
});
JMS_PROPERTY_EXPRESSIONS.put("JMSExpiration", new SubExpression() {
public Object evaluate(Message message) {
return new Long(message.getExpiration());
return Long.valueOf(message.getExpiration());
}
});
JMS_PROPERTY_EXPRESSIONS.put("JMSPriority", new SubExpression() {
public Object evaluate(Message message) {
return new Integer(message.getPriority());
return Integer.valueOf(message.getPriority());
}
});
JMS_PROPERTY_EXPRESSIONS.put("JMSTimestamp", new SubExpression() {
public Object evaluate(Message message) {
return new Long(message.getTimestamp());
return Long.valueOf(message.getTimestamp());
}
});
JMS_PROPERTY_EXPRESSIONS.put("JMSRedelivered", new SubExpression() {
public Object evaluate(Message message) {
return new Boolean(message.isRedelivered());
return Boolean.valueOf(message.isRedelivered());
}
});
JMS_PROPERTY_EXPRESSIONS.put("JMSXDeliveryCount", new SubExpression() {
public Object evaluate(Message message) {
return new Integer(message.getRedeliveryCounter()+1);
return Integer.valueOf(message.getRedeliveryCounter()+1);
}
});
JMS_PROPERTY_EXPRESSIONS.put("JMSXGroupID", new SubExpression() {

2
activemq-core/src/main/java/org/apache/activemq/filter/UnaryExpression.java
View File

@ -195,7 +195,7 @@ public abstract class UnaryExpression implements Expression {
bd = bd.negate();
if( BD_LONG_MIN_VALUE.compareTo(bd)==0 ) {
return new Long(Long.MIN_VALUE);
return Long.valueOf(Long.MIN_VALUE);
}
return bd;
}

12
activemq-core/src/main/java/org/apache/activemq/kaha/impl/async/AsyncDataManager.java
View File

@ -51,7 +51,7 @@ public final class AsyncDataManager {
private static final Log log=LogFactory.getLog(AsyncDataManager.class);
public static int CONTROL_RECORD_MAX_LENGTH=1024;
public static final int CONTROL_RECORD_MAX_LENGTH=1024;
public static final int ITEM_HEAD_RESERVED_SPACE=21;
// ITEM_HEAD_SPACE = length + type+ reserved space + SOR
@ -67,9 +67,9 @@ public final class AsyncDataManager {
public static final byte DATA_ITEM_TYPE=1;
public static final byte REDO_ITEM_TYPE=2;
public static String DEFAULT_DIRECTORY="data";
public static String DEFAULT_FILE_PREFIX="data-";
public static int DEFAULT_MAX_FILE_LENGTH=1024*1024*32;
public static final String DEFAULT_DIRECTORY="data";
public static final String DEFAULT_FILE_PREFIX="data-";
public static final int DEFAULT_MAX_FILE_LENGTH=1024*1024*32;
private File directory = new File(DEFAULT_DIRECTORY);
private String filePrefix=DEFAULT_FILE_PREFIX;
@ -314,7 +314,7 @@ public final class AsyncDataManager {
public synchronized void addInterestInFile(int file) throws IOException{
if(file>=0){
Integer key=new Integer(file);
Integer key=Integer.valueOf(file);
DataFile dataFile=(DataFile) fileMap.get(key);
if(dataFile==null){
throw new IOException("That data file does not exist");
@ -331,7 +331,7 @@ public final class AsyncDataManager {
public synchronized void removeInterestInFile(int file) throws IOException{
if(file>=0){
Integer key=new Integer(file);
Integer key=Integer.valueOf(file);
DataFile dataFile=(DataFile) fileMap.get(key);
removeInterestInFile(dataFile);
}

2
activemq-core/src/main/java/org/apache/activemq/kaha/impl/async/DataFileAppender.java
View File

@ -352,7 +352,7 @@ class DataFileAppender {
write = (WriteCommand) write.getNext();
}
}
buff.close();
} catch (IOException e) {
synchronized( enqueueMutex ) {
firstAsyncException = e;

6
activemq-core/src/main/java/org/apache/activemq/kaha/impl/data/DataManagerImpl.java
View File

@ -41,7 +41,7 @@ import org.apache.commons.logging.LogFactory;
public final class DataManagerImpl implements DataManager {
private static final Log log=LogFactory.getLog(DataManagerImpl.class);
public static long MAX_FILE_LENGTH=1024*1024*32;
public static final long MAX_FILE_LENGTH=1024*1024*32;
private static final String NAME_PREFIX="data-";
private final File dir;
private final String name;
@ -239,7 +239,7 @@ public final class DataManagerImpl implements DataManager {
*/
public synchronized void addInterestInFile(int file) throws IOException{
if(file>=0){
Integer key=new Integer(file);
Integer key=Integer.valueOf(file);
DataFile dataFile=(DataFile) fileMap.get(key);
if(dataFile==null){
dataFile=createAndAddDataFile(file);
@ -259,7 +259,7 @@ public final class DataManagerImpl implements DataManager {
*/
public synchronized void removeInterestInFile(int file) throws IOException{
if(file>=0){
Integer key=new Integer(file);
Integer key=Integer.valueOf(file);
DataFile dataFile=(DataFile) fileMap.get(key);
removeInterestInFile(dataFile);
}

2
activemq-core/src/main/java/org/apache/activemq/kaha/impl/index/IndexManager.java
View File

@ -151,7 +151,7 @@ public final class IndexManager{
return result;
}
long getLength(){
synchronized long getLength(){
return length;
}

3
activemq-core/src/main/java/org/apache/activemq/network/ForwardingBridge.java
View File

@ -72,9 +72,6 @@ public class ForwardingBridge implements Service{
private boolean dispatchAsync;
private String destinationFilter = ">";
private int queueDispatched;
private int topicDispatched;
BrokerId localBrokerId;
BrokerId remoteBrokerId;
private NetworkBridgeFailedListener bridgeFailedListener;

4
activemq-core/src/main/java/org/apache/activemq/openwire/v2/BaseDataStreamMarshaller.java
View File

@ -183,7 +183,7 @@ abstract public class BaseDataStreamMarshaller implements DataStreamMarshaller {
tightUnmarshalString(dataIn, bs),
tightUnmarshalString(dataIn, bs),
tightUnmarshalString(dataIn, bs),
new Integer(dataIn.readInt())
Integer.valueOf(dataIn.readInt())
});
} catch (IOException e) {
throw e;
@ -484,7 +484,7 @@ abstract public class BaseDataStreamMarshaller implements DataStreamMarshaller {
looseUnmarshalString(dataIn),
looseUnmarshalString(dataIn),
looseUnmarshalString(dataIn),
new Integer(dataIn.readInt())
Integer.valueOf(dataIn.readInt())
});
} catch (IOException e) {
throw e;

4
activemq-core/src/main/java/org/apache/activemq/openwire/v3/BaseDataStreamMarshaller.java
View File

@ -182,7 +182,7 @@ abstract public class BaseDataStreamMarshaller implements DataStreamMarshaller {
tightUnmarshalString(dataIn, bs),
tightUnmarshalString(dataIn, bs),
tightUnmarshalString(dataIn, bs),
new Integer(dataIn.readInt())
Integer.valueOf(dataIn.readInt())
});
} catch (IOException e) {
throw e;
@ -483,7 +483,7 @@ abstract public class BaseDataStreamMarshaller implements DataStreamMarshaller {
looseUnmarshalString(dataIn),
looseUnmarshalString(dataIn),
looseUnmarshalString(dataIn),
new Integer(dataIn.readInt())
Integer.valueOf(dataIn.readInt())
});
} catch (IOException e) {
throw e;

2
activemq-core/src/main/java/org/apache/activemq/security/AuthorizationEntry.java
View File

@ -108,7 +108,7 @@ public class AuthorizationEntry extends DestinationMapEntry {
paramClass[0] = String.class;
Object[] param = new Object[1];
param[0] = new String(name);
param[0] = name;
try {
Class cls = Class.forName(groupClass);

4
activemq-core/src/main/java/org/apache/activemq/security/LDAPAuthorizationMap.java
View File

@ -133,8 +133,8 @@ public class LDAPAuthorizationMap implements AuthorizationMap {
String queueSearchSubtree = (String) options.get(QUEUE_SEARCH_SUBTREE);
topicSearchMatchingFormat = new MessageFormat(topicSearchMatching);
queueSearchMatchingFormat = new MessageFormat(queueSearchMatching);
topicSearchSubtreeBool = new Boolean(topicSearchSubtree).booleanValue();
queueSearchSubtreeBool = new Boolean(queueSearchSubtree).booleanValue();
topicSearchSubtreeBool = Boolean.valueOf(topicSearchSubtree).booleanValue();
queueSearchSubtreeBool = Boolean.valueOf(queueSearchSubtree).booleanValue();
}
public Set getTempDestinationAdminACLs() {

4
activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaReferenceStoreAdapter.java
View File

@ -162,7 +162,7 @@ public class KahaReferenceStoreAdapter extends KahaPersistenceAdapter implements
}
synchronized void addInterestInRecordFile(int recordNumber) {
Integer key = new Integer(recordNumber);
Integer key = Integer.valueOf(recordNumber);
AtomicInteger rr = recordReferences.get(key);
if (rr == null) {
rr = new AtomicInteger();
@ -172,7 +172,7 @@ public class KahaReferenceStoreAdapter extends KahaPersistenceAdapter implements
}
synchronized void removeInterestInRecordFile(int recordNumber) {
Integer key = new Integer(recordNumber);
Integer key = Integer.valueOf(recordNumber);
AtomicInteger rr = recordReferences.get(key);
if (rr != null && rr.decrementAndGet() <= 0) {
recordReferences.remove(key);

2
activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaTopicReferenceStore.java
View File

@ -217,7 +217,7 @@ public class KahaTopicReferenceStore extends KahaReferenceStore implements Topic
if(msg!=null){
recoverReference(listener,msg);
count++;
container.setBatchEntry(msg.getMessageId().toString(),entry);
container.setBatchEntry(msg.getMessageId(),entry);
}else {
container.reset();
}

1
activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/TopicSubContainer.java
View File

@ -41,6 +41,7 @@ public class TopicSubContainer {
}
/**
* @param id
* @param batchEntry the batchEntry to set
*/
public void setBatchEntry(String id,StoreEntry batchEntry) {

2
activemq-core/src/main/java/org/apache/activemq/thread/Scheduler.java
View File

@ -30,7 +30,7 @@ import java.util.concurrent.TimeUnit;
public class Scheduler {
static public ScheduledThreadPoolExecutor clockDaemon = new ScheduledThreadPoolExecutor(5, new ThreadFactory(){
public static final ScheduledThreadPoolExecutor clockDaemon = new ScheduledThreadPoolExecutor(5, new ThreadFactory(){
public Thread newThread(Runnable runnable) {
Thread thread = new Thread(runnable,"ActiveMQ Scheduler");
thread.setDaemon(true);

2
activemq-core/src/main/java/org/apache/activemq/transport/ResponseCorrelator.java
View File

@ -84,7 +84,7 @@ public class ResponseCorrelator extends TransportFilter{
Response response=(Response)command;
FutureResponse future=null;
synchronized(requestMap){
future=(FutureResponse)requestMap.remove(new Integer(response.getCorrelationId()));
future=(FutureResponse)requestMap.remove(Integer.valueOf(response.getCorrelationId()));
}
if(future!=null){
future.set(response);

6
activemq-core/src/main/java/org/apache/activemq/transport/reliable/DefaultReplayBuffer.java
View File

@ -51,10 +51,10 @@ public class DefaultReplayBuffer implements ReplayBuffer {
int max = size - 1;
while (map.size() >= max) {
// lets find things to evict
Object evictedBuffer = map.remove(new Integer(++lowestCommandId));
Object evictedBuffer = map.remove(Integer.valueOf(++lowestCommandId));
onEvictedBuffer(lowestCommandId, evictedBuffer);
}
map.put(new Integer(commandId), buffer);
map.put(Integer.valueOf(commandId), buffer);
}
}
@ -72,7 +72,7 @@ public class DefaultReplayBuffer implements ReplayBuffer {
for (int i = fromCommandId; i <= toCommandId; i++) {
Object buffer = null;
synchronized (lock) {
buffer = map.get(new Integer(i));
buffer = map.get(Integer.valueOf(i));
}
replayer.sendBuffer(i, buffer);
}

4
activemq-core/src/main/java/org/apache/activemq/transport/stomp/ProtocolConverter.java
View File

@ -114,7 +114,7 @@ public class ProtocolConverter {
command.setCommandId(generateCommandId());
if(handler!=null) {
command.setResponseRequired(true);
resposeHandlers.put(new Integer(command.getCommandId()), handler);
resposeHandlers.put(Integer.valueOf(command.getCommandId()), handler);
}
transportFilter.sendToActiveMQ(command);
}
@ -472,7 +472,7 @@ public class ProtocolConverter {
if ( command.isResponse() ) {
Response response = (Response) command;
ResponseHandler rh = (ResponseHandler) resposeHandlers.remove(new Integer(response.getCorrelationId()));
ResponseHandler rh = (ResponseHandler) resposeHandlers.remove(Integer.valueOf(response.getCorrelationId()));
if( rh !=null ) {
rh.onResponse(this, response);
}

1
activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompWireFormat.java
View File

@ -189,6 +189,7 @@ public class StompWireFormat implements WireFormat {
throw new ProtocolException(errorMessage, true);
baos.write(b);
}
baos.close();
ByteSequence sequence = baos.toByteSequence();
return new String(sequence.getData(),sequence.getOffset(),sequence.getLength(),"UTF-8");
}

6
activemq-core/src/main/java/org/apache/activemq/transport/tcp/TcpTransportServer.java
View File

@ -165,9 +165,9 @@ public class TcpTransportServer extends TransportServerThreadSupport {
}
else {
HashMap options = new HashMap();
options.put("maxInactivityDuration", new Long(maxInactivityDuration));
options.put("minmumWireFormatVersion", new Integer(minmumWireFormatVersion));
options.put("trace", new Boolean(trace));
options.put("maxInactivityDuration", Long.valueOf(maxInactivityDuration));
options.put("minmumWireFormatVersion", Integer.valueOf(minmumWireFormatVersion));
options.put("trace", Boolean.valueOf(trace));
options.putAll(transportOptions);
WireFormat format = wireFormatFactory.createWireFormat();
Transport transport = createTransport(socket, format);

12
activemq-core/src/main/java/org/apache/activemq/util/MarshallingSupport.java
View File

@ -155,22 +155,22 @@ public class MarshallingSupport {
Object value=null;
switch( in.readByte() ) {
case BYTE_TYPE:
value = new Byte(in.readByte());
value = Byte.valueOf(in.readByte());
break;
case BOOLEAN_TYPE:
value = in.readBoolean() ? Boolean.TRUE : Boolean.FALSE;
break;
case CHAR_TYPE:
value = new Character(in.readChar());
value = Character.valueOf(in.readChar());
break;
case SHORT_TYPE:
value = new Short(in.readShort());
value = Short.valueOf(in.readShort());
break;
case INTEGER_TYPE:
value = new Integer(in.readInt());
value = Integer.valueOf(in.readInt());
break;
case LONG_TYPE:
value = new Long(in.readLong());
value = Long.valueOf(in.readLong());
break;
case FLOAT_TYPE:
value = new Float(in.readFloat());
@ -378,6 +378,7 @@ public class MarshallingSupport {
DataByteArrayOutputStream dataOut=new DataByteArrayOutputStream();
props.store(dataOut,"");
result=new String(dataOut.getData(),0,dataOut.size());
dataOut.close();
}
return result;
}
@ -387,6 +388,7 @@ public class MarshallingSupport {
if (str != null && str.length() > 0 ) {
DataByteArrayInputStream dataIn = new DataByteArrayInputStream(str.getBytes());
result.load(dataIn);
dataIn.close();
}
return result;
}

8
activemq-core/src/main/java/org/apache/activemq/util/MemoryIntPropertyEditor.java
View File

@ -32,28 +32,28 @@ public class MemoryIntPropertyEditor extends PropertyEditorSupport {
Pattern p = Pattern.compile("^\\s*(\\d+)\\s*(b)?\\s*$",Pattern.CASE_INSENSITIVE);
Matcher m = p.matcher(text);
if (m.matches()) {
setValue(new Integer(Integer.parseInt(m.group(1))));
setValue(Integer.valueOf(Integer.parseInt(m.group(1))));
return;
}
p = Pattern.compile("^\\s*(\\d+)\\s*k(b)?\\s*$",Pattern.CASE_INSENSITIVE);
m = p.matcher(text);
if (m.matches()) {
setValue(new Integer(Integer.parseInt(m.group(1)) * 1024));
setValue(Integer.valueOf(Integer.parseInt(m.group(1)) * 1024));
return;
}
p = Pattern.compile("^\\s*(\\d+)\\s*m(b)?\\s*$", Pattern.CASE_INSENSITIVE);
m = p.matcher(text);
if (m.matches()) {
setValue(new Integer(Integer.parseInt(m.group(1)) * 1024 * 1024 ));
setValue(Integer.valueOf(Integer.parseInt(m.group(1)) * 1024 * 1024 ));
return;
}
p = Pattern.compile("^\\s*(\\d+)\\s*g(b)?\\s*$", Pattern.CASE_INSENSITIVE);
m = p.matcher(text);
if (m.matches()) {
setValue(new Integer(Integer.parseInt(m.group(1)) * 1024 * 1024 * 1024 ));
setValue(Integer.valueOf(Integer.parseInt(m.group(1)) * 1024 * 1024 * 1024 ));
return;
}

8
activemq-core/src/main/java/org/apache/activemq/util/MemoryPropertyEditor.java
View File

@ -32,28 +32,28 @@ public class MemoryPropertyEditor extends PropertyEditorSupport {
Pattern p = Pattern.compile("^\\s*(\\d+)\\s*(b)?\\s*$",Pattern.CASE_INSENSITIVE);
Matcher m = p.matcher(text);
if (m.matches()) {
setValue(new Long(Long.parseLong(m.group(1))));
setValue(Long.valueOf(Long.parseLong(m.group(1))));
return;
}
p = Pattern.compile("^\\s*(\\d+)\\s*k(b)?\\s*$",Pattern.CASE_INSENSITIVE);
m = p.matcher(text);
if (m.matches()) {
setValue(new Long(Long.parseLong(m.group(1)) * 1024));
setValue(Long.valueOf(Long.parseLong(m.group(1)) * 1024));
return;
}
p = Pattern.compile("^\\s*(\\d+)\\s*m(b)?\\s*$", Pattern.CASE_INSENSITIVE);
m = p.matcher(text);
if (m.matches()) {
setValue(new Long(Long.parseLong(m.group(1)) * 1024 * 1024 ));
setValue(Long.valueOf(Long.parseLong(m.group(1)) * 1024 * 1024 ));
return;
}
p = Pattern.compile("^\\s*(\\d+)\\s*g(b)?\\s*$", Pattern.CASE_INSENSITIVE);
m = p.matcher(text);
if (m.matches()) {
setValue(new Long(Long.parseLong(m.group(1)) * 1024 * 1024 * 1024 ));
setValue(Long.valueOf(Long.parseLong(m.group(1)) * 1024 * 1024 * 1024 ));
return;
}

8
activemq-core/src/main/java/org/apache/activemq/util/TypeConversionSupport.java
View File

@ -102,7 +102,7 @@ public class TypeConversionSupport {
Converter longConverter = new Converter() {
public Object convert(Object value) {
return new Long(((Number) value).longValue());
return Long.valueOf(((Number) value).longValue());
}
};
CONVERSION_MAP.put(new ConversionKey(Byte.class, Long.class), longConverter);
@ -110,13 +110,13 @@ public class TypeConversionSupport {
CONVERSION_MAP.put(new ConversionKey(Integer.class, Long.class), longConverter);
CONVERSION_MAP.put(new ConversionKey(Date.class, Long.class), new Converter() {
public Object convert(Object value) {
return new Long(((Date) value).getTime());
return Long.valueOf(((Date) value).getTime());
}
});
Converter intConverter = new Converter() {
public Object convert(Object value) {
return new Integer(((Number) value).intValue());
return Integer.valueOf(((Number) value).intValue());
}
};
CONVERSION_MAP.put(new ConversionKey(Byte.class, Integer.class), intConverter);
@ -124,7 +124,7 @@ public class TypeConversionSupport {
CONVERSION_MAP.put(new ConversionKey(Byte.class, Short.class), new Converter() {
public Object convert(Object value) {
return new Short(((Number) value).shortValue());
return Short.valueOf(((Number) value).shortValue());
}
});

12
activemq-core/src/test/eclipse-resources/log4j.properties
View File

@ -18,12 +18,20 @@
#
# The logging properties used for eclipse testing, We want to see debug output on the console.
#
log4j.rootLogger=WARN, out
log4j.rootLogger=INFO, out
log4j.logger.org.apache.activemq=DEBUG
# CONSOLE appender not used by default
log4j.appender.out=org.apache.log4j.ConsoleAppender
log4j.appender.out.layout=org.apache.log4j.PatternLayout
log4j.appender.out.layout.ConversionPattern=[%30.30t] %-30.30c{1} %-5p %m%n
#log4j.appender.out.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n
# File appender
log4j.appender.fout=org.apache.log4j.FileAppender
log4j.appender.fout.layout=org.apache.log4j.PatternLayout
log4j.appender.fout.layout.ConversionPattern=%d [%-15.15t] %-5p %-30.30c{1} - %m%n
log4j.appender.fout.file=target/amq-testlog.log
log4j.appender.fout.append=true

876
activemq-core/src/test/java/org/apache/activemq/AMQDeadlockTest3.java
View File

@ -1,438 +1,438 @@
package org.apache.activemq;
import java.net.URI;
import java.util.ArrayList;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import javax.jms.BytesMessage;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Session;
import junit.framework.Assert;
import junit.framework.TestCase;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.TransportConnector;
import org.apache.activemq.broker.region.policy.PolicyEntry;
import org.apache.activemq.broker.region.policy.PolicyMap;
import org.apache.activemq.memory.UsageManager;
import org.apache.activemq.network.DiscoveryNetworkConnector;
import org.apache.activemq.network.NetworkConnector;
import org.apache.activemq.pool.PooledConnectionFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
public class AMQDeadlockTest3 extends TestCase {
private static final transient Log log = LogFactory.getLog(AMQDeadlockTest3.class);
private static final String URL1 = "tcp://localhost:61616";
private static final String URL2 = "tcp://localhost:61617";
private static final String QUEUE1_NAME = "test.queue.1";
private static final String QUEUE2_NAME = "test.queue.2";
private static final int MAX_CONSUMERS = 1;
private static final int MAX_PRODUCERS = 1;
private static final int NUM_MESSAGE_TO_SEND = 10;
private AtomicInteger messageCount = new AtomicInteger();
private CountDownLatch doneLatch;
public void setUp() throws Exception {
}
public void tearDown() throws Exception {
}
// This should fail with incubator-activemq-fuse-4.1.0.5
public void testQueueLimitsWithOneBrokerSameConnection() throws Exception {
BrokerService brokerService1 = null;
ActiveMQConnectionFactory acf = null;
PooledConnectionFactory pcf = null;
DefaultMessageListenerContainer container1 = null;
try {
brokerService1 = createBrokerService("broker1", URL1, null);
brokerService1.start();
acf = createConnectionFactory(URL1);
pcf = new PooledConnectionFactory(acf);
// Only listen on the first queue.. let the 2nd queue fill up.
doneLatch = new CountDownLatch(NUM_MESSAGE_TO_SEND);
container1 = createDefaultMessageListenerContainer(acf, new TestMessageListener1(500), QUEUE1_NAME);
container1.afterPropertiesSet();
Thread.sleep(2000);
final ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < MAX_PRODUCERS; i++) {
executor.submit(new PooledProducerTask(pcf, QUEUE2_NAME));
Thread.sleep(1000);
executor.submit(new PooledProducerTask(pcf, QUEUE1_NAME));
}
// Wait for all message to arrive.
assertTrue(doneLatch.await(20, TimeUnit.SECONDS));
executor.shutdownNow();
Assert.assertEquals(NUM_MESSAGE_TO_SEND, messageCount.get());
} finally {
container1.stop();
container1.destroy();
container1 = null;
brokerService1.stop();
brokerService1 = null;
}
}
// This should fail with incubator-activemq-fuse-4.1.0.5
public void testQueueLimitsWithTwoBrokerProduceandConsumeonDifferentBrokersWithOneConnectionForProducing()
throws Exception {
BrokerService brokerService1 = null;
BrokerService brokerService2 = null;
ActiveMQConnectionFactory acf1 = null;
ActiveMQConnectionFactory acf2 = null;
PooledConnectionFactory pcf = null;
DefaultMessageListenerContainer container1 = null;
try {
brokerService1 = createBrokerService("broker1", URL1, URL2);
brokerService1.start();
brokerService2 = createBrokerService("broker2", URL2, URL1);
brokerService2.start();
acf1 = createConnectionFactory(URL1);
acf2 = createConnectionFactory(URL2);
pcf = new PooledConnectionFactory(acf1);
Thread.sleep(1000);
doneLatch = new CountDownLatch(MAX_PRODUCERS * NUM_MESSAGE_TO_SEND);
container1 = createDefaultMessageListenerContainer(acf2, new TestMessageListener1(500), QUEUE1_NAME);
container1.afterPropertiesSet();
final ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < MAX_PRODUCERS; i++) {
executor.submit(new PooledProducerTask(pcf, QUEUE2_NAME));
Thread.sleep(1000);
executor.submit(new PooledProducerTask(pcf, QUEUE1_NAME));
}
assertTrue(doneLatch.await(20, TimeUnit.SECONDS));
executor.shutdownNow();
Assert.assertEquals(MAX_PRODUCERS * NUM_MESSAGE_TO_SEND,
messageCount.get());
} finally {
container1.stop();
container1.destroy();
container1 = null;
brokerService1.stop();
brokerService1 = null;
brokerService2.stop();
brokerService2 = null;
}
}
// This should fail with incubator-activemq-fuse-4.1.0.5
public void testQueueLimitsWithTwoBrokerProduceandConsumeonDifferentBrokersWithSeperateConnectionsForProducing()
throws Exception {
BrokerService brokerService1 = null;
BrokerService brokerService2 = null;
ActiveMQConnectionFactory acf1 = null;
ActiveMQConnectionFactory acf2 = null;
DefaultMessageListenerContainer container1 = null;
DefaultMessageListenerContainer container2 = null;
try {
brokerService1 = createBrokerService("broker1", URL1, URL2);
brokerService1.start();
brokerService2 = createBrokerService("broker2", URL2, URL1);
brokerService2.start();
acf1 = createConnectionFactory(URL1);
acf2 = createConnectionFactory(URL2);
Thread.sleep(1000);
doneLatch = new CountDownLatch(NUM_MESSAGE_TO_SEND*MAX_PRODUCERS);
container1 = createDefaultMessageListenerContainer(acf2, new TestMessageListener1(500), QUEUE1_NAME);
container1.afterPropertiesSet();
container2 = createDefaultMessageListenerContainer(acf2, new TestMessageListener1(30000), QUEUE2_NAME);
container2.afterPropertiesSet();
final ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < MAX_PRODUCERS; i++) {
executor.submit(new NonPooledProducerTask(acf1, QUEUE2_NAME));
Thread.sleep(1000);
executor.submit(new NonPooledProducerTask(acf1, QUEUE1_NAME));
}
assertTrue(doneLatch.await(20, TimeUnit.SECONDS));
executor.shutdownNow();
Assert.assertEquals(MAX_PRODUCERS * NUM_MESSAGE_TO_SEND, messageCount.get());
} finally {
container1.stop();
container1.destroy();
container1 = null;
container2.stop();
container2.destroy();
container2 = null;
brokerService1.stop();
brokerService1 = null;
brokerService2.stop();
brokerService2 = null;
}
}
private BrokerService createBrokerService(final String brokerName,
final String uri1, final String uri2) throws Exception {
final BrokerService brokerService = new BrokerService();
brokerService.setBrokerName(brokerName);
brokerService.setPersistent(false);
brokerService.setUseJmx(true);
final UsageManager memoryManager = new UsageManager();
memoryManager.setLimit(5000000);
brokerService.setMemoryManager(memoryManager);
final ArrayList policyEntries = new ArrayList();
final PolicyEntry entry = new PolicyEntry();
entry.setQueue(">");
// entry.setQueue(QUEUE1_NAME);
entry.setMemoryLimit(1000);
policyEntries.add(entry);
final PolicyMap policyMap = new PolicyMap();
policyMap.setPolicyEntries(policyEntries);
brokerService.setDestinationPolicy(policyMap);
final TransportConnector tConnector = new TransportConnector();
tConnector.setUri(new URI(uri1));
tConnector.setBrokerName(brokerName);
tConnector.setName(brokerName + ".transportConnector");
brokerService.addConnector(tConnector);
if (uri2 != null) {
final NetworkConnector nc = new DiscoveryNetworkConnector(new URI("static:" + uri2));
nc.setBridgeTempDestinations(true);
nc.setBrokerName(brokerName);
brokerService.addNetworkConnector(nc);
}
return brokerService;
}
public DefaultMessageListenerContainer createDefaultMessageListenerContainer(
final ConnectionFactory acf, final MessageListener listener,
final String queue) {
final DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
container.setConnectionFactory(acf);
container.setDestinationName(queue);
container.setMessageListener(listener);
container.setSessionTransacted(false);
container.setSessionAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);
container.setConcurrentConsumers(MAX_CONSUMERS);
return container;
}
public ActiveMQConnectionFactory createConnectionFactory(final String url) {
final ActiveMQConnectionFactory acf = new ActiveMQConnectionFactory(url);
acf.setCopyMessageOnSend(false);
acf.setUseAsyncSend(false);
acf.setDispatchAsync(true);
acf.setUseCompression(false);
acf.setOptimizeAcknowledge(false);
acf.setOptimizedMessageDispatch(true);
acf.setAlwaysSyncSend(true);
return acf;
}
private class TestMessageListener1 implements MessageListener {
private final long waitTime;
public TestMessageListener1(long waitTime) {
this.waitTime = waitTime;
}
public void onMessage(Message msg) {
try {
log.info("Listener1 Consumed message "+ msg.getIntProperty("count"));
messageCount.incrementAndGet();
doneLatch.countDown();
Thread.sleep(waitTime);
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
private class PooledProducerTask implements Runnable {
private final String queueName;
private final PooledConnectionFactory pcf;
public PooledProducerTask(final PooledConnectionFactory pcf,
final String queueName) {
this.pcf = pcf;
this.queueName = queueName;
}
public void run() {
try {
final JmsTemplate jmsTemplate = new JmsTemplate(pcf);
jmsTemplate.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
jmsTemplate.setExplicitQosEnabled(true);
jmsTemplate.setMessageIdEnabled(false);
jmsTemplate.setMessageTimestampEnabled(false);
jmsTemplate.afterPropertiesSet();
final byte[] bytes = new byte[2048];
final Random r = new Random();
r.nextBytes(bytes);
Thread.sleep(2000);
final AtomicInteger count = new AtomicInteger();
for (int i = 0; i < NUM_MESSAGE_TO_SEND; i++) {
jmsTemplate.send(queueName, new MessageCreator() {
public Message createMessage(Session session)
throws JMSException {
final BytesMessage message = session.createBytesMessage();
message.writeBytes(bytes);
message.setIntProperty("count", count.incrementAndGet());
message.setStringProperty("producer", "pooled");
return message;
}
});
log.info("PooledProducer sent message: "+ count.get());
// Thread.sleep(1000);
}
} catch (final Throwable e) {
log.error("Producer 1 is exiting", e);
}
}
}
private class NonPooledProducerTask implements Runnable {
private final String queueName;
private final ConnectionFactory cf;
public NonPooledProducerTask(final ConnectionFactory cf,
final String queueName) {
this.cf = cf;
this.queueName = queueName;
}
public void run() {
try {
final JmsTemplate jmsTemplate = new JmsTemplate(cf);
jmsTemplate.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
jmsTemplate.setExplicitQosEnabled(true);
jmsTemplate.setMessageIdEnabled(false);
jmsTemplate.setMessageTimestampEnabled(false);
jmsTemplate.afterPropertiesSet();
final byte[] bytes = new byte[2048];
final Random r = new Random();
r.nextBytes(bytes);
Thread.sleep(2000);
final AtomicInteger count = new AtomicInteger();
for (int i = 0; i < NUM_MESSAGE_TO_SEND; i++) {
jmsTemplate.send(queueName, new MessageCreator() {
public Message createMessage(Session session)
throws JMSException {
final BytesMessage message = session
.createBytesMessage();
message.writeBytes(bytes);
message.setIntProperty("count", count
.incrementAndGet());
message.setStringProperty("producer", "non-pooled");
return message;
}
});
log.info("Non-PooledProducer sent message: " + count.get());
// Thread.sleep(1000);
}
} catch (final Throwable e) {
log.error("Producer 1 is exiting", e);
}
}
}
}
package org.apache.activemq;
import java.net.URI;
import java.util.ArrayList;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import javax.jms.BytesMessage;
import javax.jms.ConnectionFactory;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import javax.jms.Session;
import junit.framework.Assert;
import junit.framework.TestCase;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.TransportConnector;
import org.apache.activemq.broker.region.policy.PolicyEntry;
import org.apache.activemq.broker.region.policy.PolicyMap;
import org.apache.activemq.memory.UsageManager;
import org.apache.activemq.network.DiscoveryNetworkConnector;
import org.apache.activemq.network.NetworkConnector;
import org.apache.activemq.pool.PooledConnectionFactory;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.jms.core.MessageCreator;
import org.springframework.jms.listener.DefaultMessageListenerContainer;
public class AMQDeadlockTest3 extends TestCase {
private static final transient Log log = LogFactory.getLog(AMQDeadlockTest3.class);
private static final String URL1 = "tcp://localhost:61616";
private static final String URL2 = "tcp://localhost:61617";
private static final String QUEUE1_NAME = "test.queue.1";
private static final String QUEUE2_NAME = "test.queue.2";
private static final int MAX_CONSUMERS = 1;
private static final int MAX_PRODUCERS = 1;
private static final int NUM_MESSAGE_TO_SEND = 10;
private AtomicInteger messageCount = new AtomicInteger();
private CountDownLatch doneLatch;
public void setUp() throws Exception {
}
public void tearDown() throws Exception {
}
// This should fail with incubator-activemq-fuse-4.1.0.5
public void testQueueLimitsWithOneBrokerSameConnection() throws Exception {
BrokerService brokerService1 = null;
ActiveMQConnectionFactory acf = null;
PooledConnectionFactory pcf = null;
DefaultMessageListenerContainer container1 = null;
try {
brokerService1 = createBrokerService("broker1", URL1, null);
brokerService1.start();
acf = createConnectionFactory(URL1);
pcf = new PooledConnectionFactory(acf);
// Only listen on the first queue.. let the 2nd queue fill up.
doneLatch = new CountDownLatch(NUM_MESSAGE_TO_SEND);
container1 = createDefaultMessageListenerContainer(acf, new TestMessageListener1(500), QUEUE1_NAME);
container1.afterPropertiesSet();
Thread.sleep(2000);
final ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < MAX_PRODUCERS; i++) {
executor.submit(new PooledProducerTask(pcf, QUEUE2_NAME));
Thread.sleep(1000);
executor.submit(new PooledProducerTask(pcf, QUEUE1_NAME));
}
// Wait for all message to arrive.
assertTrue(doneLatch.await(20, TimeUnit.SECONDS));
executor.shutdownNow();
Assert.assertEquals(NUM_MESSAGE_TO_SEND, messageCount.get());
} finally {
container1.stop();
container1.destroy();
container1 = null;
brokerService1.stop();
brokerService1 = null;
}
}
// This should fail with incubator-activemq-fuse-4.1.0.5
public void testQueueLimitsWithTwoBrokerProduceandConsumeonDifferentBrokersWithOneConnectionForProducing()
throws Exception {
BrokerService brokerService1 = null;
BrokerService brokerService2 = null;
ActiveMQConnectionFactory acf1 = null;
ActiveMQConnectionFactory acf2 = null;
PooledConnectionFactory pcf = null;
DefaultMessageListenerContainer container1 = null;
try {
brokerService1 = createBrokerService("broker1", URL1, URL2);
brokerService1.start();
brokerService2 = createBrokerService("broker2", URL2, URL1);
brokerService2.start();
acf1 = createConnectionFactory(URL1);
acf2 = createConnectionFactory(URL2);
pcf = new PooledConnectionFactory(acf1);
Thread.sleep(1000);
doneLatch = new CountDownLatch(MAX_PRODUCERS * NUM_MESSAGE_TO_SEND);
container1 = createDefaultMessageListenerContainer(acf2, new TestMessageListener1(500), QUEUE1_NAME);
container1.afterPropertiesSet();
final ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < MAX_PRODUCERS; i++) {
executor.submit(new PooledProducerTask(pcf, QUEUE2_NAME));
Thread.sleep(1000);
executor.submit(new PooledProducerTask(pcf, QUEUE1_NAME));
}
assertTrue(doneLatch.await(20, TimeUnit.SECONDS));
executor.shutdownNow();
Assert.assertEquals(MAX_PRODUCERS * NUM_MESSAGE_TO_SEND,
messageCount.get());
} finally {
container1.stop();
container1.destroy();
container1 = null;
brokerService1.stop();
brokerService1 = null;
brokerService2.stop();
brokerService2 = null;
}
}
// This should fail with incubator-activemq-fuse-4.1.0.5
public void testQueueLimitsWithTwoBrokerProduceandConsumeonDifferentBrokersWithSeperateConnectionsForProducing()
throws Exception {
BrokerService brokerService1 = null;
BrokerService brokerService2 = null;
ActiveMQConnectionFactory acf1 = null;
ActiveMQConnectionFactory acf2 = null;
DefaultMessageListenerContainer container1 = null;
DefaultMessageListenerContainer container2 = null;
try {
brokerService1 = createBrokerService("broker1", URL1, URL2);
brokerService1.start();
brokerService2 = createBrokerService("broker2", URL2, URL1);
brokerService2.start();
acf1 = createConnectionFactory(URL1);
acf2 = createConnectionFactory(URL2);
Thread.sleep(1000);
doneLatch = new CountDownLatch(NUM_MESSAGE_TO_SEND*MAX_PRODUCERS);
container1 = createDefaultMessageListenerContainer(acf2, new TestMessageListener1(500), QUEUE1_NAME);
container1.afterPropertiesSet();
container2 = createDefaultMessageListenerContainer(acf2, new TestMessageListener1(30000), QUEUE2_NAME);
container2.afterPropertiesSet();
final ExecutorService executor = Executors.newCachedThreadPool();
for (int i = 0; i < MAX_PRODUCERS; i++) {
executor.submit(new NonPooledProducerTask(acf1, QUEUE2_NAME));
Thread.sleep(1000);
executor.submit(new NonPooledProducerTask(acf1, QUEUE1_NAME));
}
assertTrue(doneLatch.await(20, TimeUnit.SECONDS));
executor.shutdownNow();
Assert.assertEquals(MAX_PRODUCERS * NUM_MESSAGE_TO_SEND, messageCount.get());
} finally {
container1.stop();
container1.destroy();
container1 = null;
container2.stop();
container2.destroy();
container2 = null;
brokerService1.stop();
brokerService1 = null;
brokerService2.stop();
brokerService2 = null;
}
}
private BrokerService createBrokerService(final String brokerName,
final String uri1, final String uri2) throws Exception {
final BrokerService brokerService = new BrokerService();
brokerService.setBrokerName(brokerName);
brokerService.setPersistent(false);
brokerService.setUseJmx(true);
final UsageManager memoryManager = new UsageManager();
memoryManager.setLimit(5000000);
brokerService.setMemoryManager(memoryManager);
final ArrayList policyEntries = new ArrayList();
final PolicyEntry entry = new PolicyEntry();
entry.setQueue(">");
// entry.setQueue(QUEUE1_NAME);
entry.setMemoryLimit(1000);
policyEntries.add(entry);
final PolicyMap policyMap = new PolicyMap();
policyMap.setPolicyEntries(policyEntries);
brokerService.setDestinationPolicy(policyMap);
final TransportConnector tConnector = new TransportConnector();
tConnector.setUri(new URI(uri1));
tConnector.setBrokerName(brokerName);
tConnector.setName(brokerName + ".transportConnector");
brokerService.addConnector(tConnector);
if (uri2 != null) {
final NetworkConnector nc = new DiscoveryNetworkConnector(new URI("static:" + uri2));
nc.setBridgeTempDestinations(true);
nc.setBrokerName(brokerName);
brokerService.addNetworkConnector(nc);
}
return brokerService;
}
public DefaultMessageListenerContainer createDefaultMessageListenerContainer(
final ConnectionFactory acf, final MessageListener listener,
final String queue) {
final DefaultMessageListenerContainer container = new DefaultMessageListenerContainer();
container.setConnectionFactory(acf);
container.setDestinationName(queue);
container.setMessageListener(listener);
container.setSessionTransacted(false);
container.setSessionAcknowledgeMode(Session.AUTO_ACKNOWLEDGE);
container.setConcurrentConsumers(MAX_CONSUMERS);
return container;
}
public ActiveMQConnectionFactory createConnectionFactory(final String url) {
final ActiveMQConnectionFactory acf = new ActiveMQConnectionFactory(url);
acf.setCopyMessageOnSend(false);
acf.setUseAsyncSend(false);
acf.setDispatchAsync(true);
acf.setUseCompression(false);
acf.setOptimizeAcknowledge(false);
acf.setOptimizedMessageDispatch(true);
acf.setAlwaysSyncSend(true);
return acf;
}
private class TestMessageListener1 implements MessageListener {
private final long waitTime;
public TestMessageListener1(long waitTime) {
this.waitTime = waitTime;
}
public void onMessage(Message msg) {
try {
log.info("Listener1 Consumed message "+ msg.getIntProperty("count"));
messageCount.incrementAndGet();
doneLatch.countDown();
Thread.sleep(waitTime);
} catch (JMSException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
}
private static class PooledProducerTask implements Runnable {
private final String queueName;
private final PooledConnectionFactory pcf;
public PooledProducerTask(final PooledConnectionFactory pcf,
final String queueName) {
this.pcf = pcf;
this.queueName = queueName;
}
public void run() {
try {
final JmsTemplate jmsTemplate = new JmsTemplate(pcf);
jmsTemplate.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
jmsTemplate.setExplicitQosEnabled(true);
jmsTemplate.setMessageIdEnabled(false);
jmsTemplate.setMessageTimestampEnabled(false);
jmsTemplate.afterPropertiesSet();
final byte[] bytes = new byte[2048];
final Random r = new Random();
r.nextBytes(bytes);
Thread.sleep(2000);
final AtomicInteger count = new AtomicInteger();
for (int i = 0; i < NUM_MESSAGE_TO_SEND; i++) {
jmsTemplate.send(queueName, new MessageCreator() {
public Message createMessage(Session session)
throws JMSException {
final BytesMessage message = session.createBytesMessage();
message.writeBytes(bytes);
message.setIntProperty("count", count.incrementAndGet());
message.setStringProperty("producer", "pooled");
return message;
}
});
log.info("PooledProducer sent message: "+ count.get());
// Thread.sleep(1000);
}
} catch (final Throwable e) {
log.error("Producer 1 is exiting", e);
}
}
}
private static class NonPooledProducerTask implements Runnable {
private final String queueName;
private final ConnectionFactory cf;
public NonPooledProducerTask(final ConnectionFactory cf,
final String queueName) {
this.cf = cf;
this.queueName = queueName;
}
public void run() {
try {
final JmsTemplate jmsTemplate = new JmsTemplate(cf);
jmsTemplate.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
jmsTemplate.setExplicitQosEnabled(true);
jmsTemplate.setMessageIdEnabled(false);
jmsTemplate.setMessageTimestampEnabled(false);
jmsTemplate.afterPropertiesSet();
final byte[] bytes = new byte[2048];
final Random r = new Random();
r.nextBytes(bytes);
Thread.sleep(2000);
final AtomicInteger count = new AtomicInteger();
for (int i = 0; i < NUM_MESSAGE_TO_SEND; i++) {
jmsTemplate.send(queueName, new MessageCreator() {
public Message createMessage(Session session)
throws JMSException {
final BytesMessage message = session
.createBytesMessage();
message.writeBytes(bytes);
message.setIntProperty("count", count
.incrementAndGet());
message.setStringProperty("producer", "non-pooled");
return message;
}
});
log.info("Non-PooledProducer sent message: " + count.get());
// Thread.sleep(1000);
}
} catch (final Throwable e) {
log.error("Producer 1 is exiting", e);
}
}
}
}

130
activemq-core/src/test/java/org/apache/activemq/JMSConsumerTest.java
View File

@ -65,13 +65,13 @@ public class JMSConsumerTest extends JmsTestSupport {
public void initCombosForTestMessageListenerWithConsumerCanBeStopped() {
addCombinationValues("deliveryMode", new Object[] {
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT) });
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT) });
addCombinationValues("destinationType", new Object[] {
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE) });
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE) });
}
public void testMessageListenerWithConsumerCanBeStopped() throws Exception {
@ -115,17 +115,17 @@ public class JMSConsumerTest extends JmsTestSupport {
public void initCombosForTestMutiReceiveWithPrefetch1() {
addCombinationValues("deliveryMode", new Object[] {
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT) });
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT) });
addCombinationValues("ackMode", new Object[] {
new Integer(Session.AUTO_ACKNOWLEDGE),
new Integer(Session.DUPS_OK_ACKNOWLEDGE),
new Integer(Session.CLIENT_ACKNOWLEDGE) });
Integer.valueOf(Session.AUTO_ACKNOWLEDGE),
Integer.valueOf(Session.DUPS_OK_ACKNOWLEDGE),
Integer.valueOf(Session.CLIENT_ACKNOWLEDGE) });
addCombinationValues("destinationType", new Object[] {
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE)
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)
});
}
@ -155,10 +155,10 @@ public class JMSConsumerTest extends JmsTestSupport {
public void initCombosForTestDurableConsumerSelectorChange() {
addCombinationValues("deliveryMode", new Object[] {
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT) });
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT) });
addCombinationValues("destinationType", new Object[] {
new Byte(ActiveMQDestination.TOPIC_TYPE)});
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE)});
}
public void testDurableConsumerSelectorChange() throws Exception {
@ -200,11 +200,11 @@ public class JMSConsumerTest extends JmsTestSupport {
}
public void initCombosForTestSendReceiveBytesMessage() {
addCombinationValues("deliveryMode", new Object[] { new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT) });
addCombinationValues("destinationType", new Object[] { new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE), new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE) });
addCombinationValues("deliveryMode", new Object[] { Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT) });
addCombinationValues("destinationType", new Object[] { Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE), Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE) });
}
public void testSendReceiveBytesMessage() throws Exception {
@ -233,13 +233,13 @@ public class JMSConsumerTest extends JmsTestSupport {
public void initCombosForTestSetMessageListenerAfterStart() {
addCombinationValues("deliveryMode", new Object[] {
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT) });
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT) });
addCombinationValues("destinationType", new Object[] {
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE) });
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE) });
}
public void testSetMessageListenerAfterStart() throws Exception {
@ -273,15 +273,15 @@ public class JMSConsumerTest extends JmsTestSupport {
public void initCombosForTestMessageListenerUnackedWithPrefetch1StayInQueue() {
addCombinationValues("deliveryMode", new Object[] {
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)
});
addCombinationValues("ackMode", new Object[] {
new Integer(Session.AUTO_ACKNOWLEDGE),
new Integer(Session.DUPS_OK_ACKNOWLEDGE),
new Integer(Session.CLIENT_ACKNOWLEDGE)
Integer.valueOf(Session.AUTO_ACKNOWLEDGE),
Integer.valueOf(Session.DUPS_OK_ACKNOWLEDGE),
Integer.valueOf(Session.CLIENT_ACKNOWLEDGE)
});
addCombinationValues("destinationType", new Object[] { new Byte(ActiveMQDestination.QUEUE_TYPE), });
addCombinationValues("destinationType", new Object[] { Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), });
}
public void testMessageListenerUnackedWithPrefetch1StayInQueue() throws Exception {
@ -364,13 +364,13 @@ public class JMSConsumerTest extends JmsTestSupport {
public void initCombosForTestMessageListenerWithConsumerWithPrefetch1() {
addCombinationValues("deliveryMode", new Object[] {
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT) });
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT) });
addCombinationValues("destinationType", new Object[] {
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE) });
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE) });
}
public void testMessageListenerWithConsumerWithPrefetch1() throws Exception {
@ -404,13 +404,13 @@ public class JMSConsumerTest extends JmsTestSupport {
public void initCombosForTestMessageListenerWithConsumer() {
addCombinationValues("deliveryMode", new Object[] {
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT) });
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT) });
addCombinationValues("destinationType", new Object[] {
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE) });
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE) });
}
public void testMessageListenerWithConsumer() throws Exception {
@ -441,11 +441,11 @@ public class JMSConsumerTest extends JmsTestSupport {
}
public void initCombosForTestUnackedWithPrefetch1StayInQueue() {
addCombinationValues("deliveryMode", new Object[] { new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT) });
addCombinationValues("ackMode", new Object[] { new Integer(Session.AUTO_ACKNOWLEDGE),
new Integer(Session.DUPS_OK_ACKNOWLEDGE), new Integer(Session.CLIENT_ACKNOWLEDGE) });
addCombinationValues("destinationType", new Object[] { new Byte(ActiveMQDestination.QUEUE_TYPE), });
addCombinationValues("deliveryMode", new Object[] { Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT) });
addCombinationValues("ackMode", new Object[] { Integer.valueOf(Session.AUTO_ACKNOWLEDGE),
Integer.valueOf(Session.DUPS_OK_ACKNOWLEDGE), Integer.valueOf(Session.CLIENT_ACKNOWLEDGE) });
addCombinationValues("destinationType", new Object[] { Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), });
}
public void testUnackedWithPrefetch1StayInQueue() throws Exception {
@ -490,8 +490,8 @@ public class JMSConsumerTest extends JmsTestSupport {
}
public void initCombosForTestPrefetch1MessageNotDispatched() {
addCombinationValues("deliveryMode", new Object[] { new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT) });
addCombinationValues("deliveryMode", new Object[] { Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT) });
}
public void testPrefetch1MessageNotDispatched() throws Exception {
@ -532,9 +532,9 @@ public class JMSConsumerTest extends JmsTestSupport {
}
public void initCombosForTestDontStart() {
addCombinationValues("deliveryMode", new Object[] { new Integer(DeliveryMode.NON_PERSISTENT), });
addCombinationValues("destinationType", new Object[] { new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE), });
addCombinationValues("deliveryMode", new Object[] { Integer.valueOf(DeliveryMode.NON_PERSISTENT), });
addCombinationValues("destinationType", new Object[] { Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE), });
}
public void testDontStart() throws Exception {
@ -551,9 +551,9 @@ public class JMSConsumerTest extends JmsTestSupport {
}
public void initCombosForTestStartAfterSend() {
addCombinationValues("deliveryMode", new Object[] { new Integer(DeliveryMode.NON_PERSISTENT), });
addCombinationValues("destinationType", new Object[] { new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE), });
addCombinationValues("deliveryMode", new Object[] { Integer.valueOf(DeliveryMode.NON_PERSISTENT), });
addCombinationValues("destinationType", new Object[] { Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE), });
}
public void testStartAfterSend() throws Exception {
@ -574,11 +574,11 @@ public class JMSConsumerTest extends JmsTestSupport {
}
public void initCombosForTestReceiveMessageWithConsumer() {
addCombinationValues("deliveryMode", new Object[] { new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT) });
addCombinationValues("destinationType", new Object[] { new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE), new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE) });
addCombinationValues("deliveryMode", new Object[] { Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT) });
addCombinationValues("destinationType", new Object[] { Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE), Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE) });
}
public void testReceiveMessageWithConsumer() throws Exception {

8
activemq-core/src/test/java/org/apache/activemq/JMSExclusiveConsumerTest.java
View File

@ -45,8 +45,8 @@ public class JMSExclusiveConsumerTest extends JmsTestSupport {
public int deliveryMode;
public void initCombosForTestRoundRobinDispatchOnNonExclusive() {
addCombinationValues("deliveryMode", new Object[] { new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT) });
addCombinationValues("deliveryMode", new Object[] { Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT) });
}
/**
@ -82,8 +82,8 @@ public class JMSExclusiveConsumerTest extends JmsTestSupport {
}
public void initCombosForTestDispatchExclusive() {
addCombinationValues("deliveryMode", new Object[] { new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT) });
addCombinationValues("deliveryMode", new Object[] { Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT) });
}
/**

6
activemq-core/src/test/java/org/apache/activemq/JMSMessageTest.java
View File

@ -65,10 +65,10 @@ public class JMSMessageTest extends JmsTestSupport {
"vm://localhost?marshal=true"
});
addCombinationValues("deliveryMode", new Object[] {
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT) });
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT) });
addCombinationValues("destinationType", new Object[] {
new Byte(ActiveMQDestination.QUEUE_TYPE)});
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE)});
}
public void testTextMessage() throws Exception {

32
activemq-core/src/test/java/org/apache/activemq/JMSUsecaseTest.java
View File

@ -50,11 +50,11 @@ public class JMSUsecaseTest extends JmsTestSupport {
public void initCombosForTestQueueBrowser() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
} );
}
public void testQueueBrowser() throws Exception {
@ -80,13 +80,13 @@ public class JMSUsecaseTest extends JmsTestSupport {
public void initCombosForTestSendReceive() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
}
public void testSendReceive() throws Exception {
// Send a message to the broker.
@ -105,13 +105,13 @@ public class JMSUsecaseTest extends JmsTestSupport {
public void initCombosForTestSendReceiveTransacted() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
}
public void testSendReceiveTransacted() throws Exception {
// Send a message to the broker.

4
activemq-core/src/test/java/org/apache/activemq/JmsTransactionTestSupport.java
View File

@ -54,8 +54,8 @@ abstract public class JmsTransactionTestSupport extends TestSupport implements M
protected Destination destination;
// for message listener test
private final int messageCount = 5;
private final String messageText = "message";
private static final int messageCount = 5;
private static final String messageText = "message";
private List unackMessages = new ArrayList(messageCount);
private List ackMessages = new ArrayList(messageCount);
private boolean resendPhase = false;

4
activemq-core/src/test/java/org/apache/activemq/LargeStreamletTest.java
View File

@ -94,7 +94,7 @@ public final class LargeStreamletTest extends TestCase {
});
final Thread writerThread = new Thread(new Runnable() {
private final Random random = new Random();
public void run() {
totalWritten.set(0);
int count = MESSAGE_COUNT;
@ -103,7 +103,7 @@ public final class LargeStreamletTest extends TestCase {
.createOutputStream(destination);
try {
final byte[] buf = new byte[BUFFER_SIZE];
new Random().nextBytes(buf);
random.nextBytes(buf);
while (count > 0 && !stopThreads.get()) {
outputStream.write(buf);
totalWritten.addAndGet(buf.length);

24
activemq-core/src/test/java/org/apache/activemq/LoadTestBurnIn.java
View File

@ -97,25 +97,25 @@ public class LoadTestBurnIn extends JmsTestSupport {
public void initCombosForTestSendReceive() {
addCombinationValues("deliveryMode", new Object[] {
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT) });
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT) });
addCombinationValues("destinationType", new Object[] {
new Byte(ActiveMQDestination.TOPIC_TYPE),
// new Byte(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
// Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
});
addCombinationValues("durableConsumer", new Object[] {
Boolean.TRUE,
// Boolean.FALSE,
});
addCombinationValues("messageSize", new Object[] {
new Integer(101),
new Integer(102),
new Integer(103),
new Integer(104),
new Integer(105),
new Integer(106),
new Integer(107),
new Integer(108),
Integer.valueOf(101),
Integer.valueOf(102),
Integer.valueOf(103),
Integer.valueOf(104),
Integer.valueOf(105),
Integer.valueOf(106),
Integer.valueOf(107),
Integer.valueOf(108),
});
}

228
activemq-core/src/test/java/org/apache/activemq/broker/BrokerTest.java
View File

@ -47,8 +47,8 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestQueuBrowserWith2Consumers() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
}
/**
@ -114,13 +114,13 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestConsumerPrefetchAndStandardAck() {
addCombinationValues( "deliveryMode", new Object[]{
// new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
// Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE)
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)
} );
}
@ -168,13 +168,13 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestTransactedAckWithPrefetchOfOne() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE)
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)
} );
}
@ -222,13 +222,13 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestTransactedSend() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
}
public void testTransactedSend() throws Exception {
@ -276,11 +276,11 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestQueueTransactedAck() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
} );
}
@ -329,8 +329,8 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestConsumerCloseCausesRedelivery() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destination", new Object[]{
new ActiveMQQueue("TEST")} );
}
@ -442,8 +442,8 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestGroupedMessagesDeliveredToOnlyOneConsumer() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
}
public void testGroupedMessagesDeliveredToOnlyOneConsumer() throws Exception {
@ -503,8 +503,8 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestTopicConsumerOnlySeeMessagesAfterCreation() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "durableConsumer", new Object[]{
Boolean.TRUE,
Boolean.FALSE});
@ -551,8 +551,8 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestTopicRetroactiveConsumerSeeMessagesBeforeCreation() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "durableConsumer", new Object[]{
Boolean.TRUE,
Boolean.FALSE});
@ -611,11 +611,11 @@ public class BrokerTest extends BrokerTestSupport {
//
// public void initCombosForTestTempDestinationsRemovedOnConnectionClose() {
// addCombinationValues( "deliveryMode", new Object[]{
// new Integer(DeliveryMode.NON_PERSISTENT),
// new Integer(DeliveryMode.PERSISTENT)} );
// Integer.valueOf(DeliveryMode.NON_PERSISTENT),
// Integer.valueOf(DeliveryMode.PERSISTENT)} );
// addCombinationValues( "destinationType", new Object[]{
// new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
// new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
// Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
// Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
// }
//
// public void testTempDestinationsRemovedOnConnectionClose() throws Exception {
@ -657,11 +657,11 @@ public class BrokerTest extends BrokerTestSupport {
// public void initCombosForTestTempDestinationsAreNotAutoCreated() {
// addCombinationValues( "deliveryMode", new Object[]{
// new Integer(DeliveryMode.NON_PERSISTENT),
// new Integer(DeliveryMode.PERSISTENT)} );
// Integer.valueOf(DeliveryMode.NON_PERSISTENT),
// Integer.valueOf(DeliveryMode.PERSISTENT)} );
// addCombinationValues( "destinationType", new Object[]{
// new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
// new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
// Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
// Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
// }
//
//
@ -697,11 +697,11 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestTempDestinationsOnlyAllowsLocalConsumers() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
}
public void testTempDestinationsOnlyAllowsLocalConsumers() throws Exception {
@ -739,8 +739,8 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestExclusiveQueueDeliversToOnlyOneConsumer() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
}
public void testExclusiveQueueDeliversToOnlyOneConsumer() throws Exception {
@ -803,11 +803,11 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestWildcardConsume() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE)} );
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE)} );
}
public void testWildcardConsume() throws Exception {
@ -850,11 +850,11 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestCompositeConsume() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE)} );
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE)} );
}
public void testCompositeConsume() throws Exception {
@ -895,11 +895,11 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestCompositeSend() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE)} );
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE)} );
}
public void testCompositeSend() throws Exception {
@ -964,8 +964,8 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestConnectionCloseCascades() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destination", new Object[]{
new ActiveMQTopic("TEST"),
new ActiveMQQueue("TEST")} );
@ -1018,8 +1018,8 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestSessionCloseCascades() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destination", new Object[]{
new ActiveMQTopic("TEST"),
new ActiveMQQueue("TEST")} );
@ -1072,8 +1072,8 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestConsumerClose() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destination", new Object[]{
new ActiveMQTopic("TEST"),
new ActiveMQQueue("TEST")} );
@ -1125,8 +1125,8 @@ public class BrokerTest extends BrokerTestSupport {
}
public void initCombosForTestTopicNoLocal() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
}
public void testTopicNoLocal() throws Exception {
@ -1192,8 +1192,8 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTopicDispatchIsBroadcast() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
}
public void testTopicDispatchIsBroadcast() throws Exception {
@ -1242,11 +1242,11 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestQueueDispatchedAreRedeliveredOnConsumerClose() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
} );
}
@ -1301,11 +1301,11 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestQueueBrowseMessages() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
} );
}
public void testQueueBrowseMessages() throws Exception {
@ -1343,8 +1343,8 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestQueueOnlyOnceDeliveryWith2Consumers() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
}
public void testQueueOnlyOnceDeliveryWith2Consumers() throws Exception {
@ -1397,11 +1397,11 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestQueueSendThenAddConsumer() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
} );
}
public void testQueueSendThenAddConsumer() throws Exception {
@ -1432,11 +1432,11 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestQueueAckRemovesMessage() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
} );
}
@ -1477,10 +1477,10 @@ public class BrokerTest extends BrokerTestSupport {
new ActiveMQTopic("TEST_TOPIC"),
new ActiveMQQueue("TEST_QUEUE")} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
}
public void testSelectorSkipsMessages() throws Exception {
@ -1518,13 +1518,13 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestAddConsumerThenSend() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
}
public void testAddConsumerThenSend() throws Exception {
@ -1552,13 +1552,13 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestConsumerPrefetchAtOne() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
}
public void testConsumerPrefetchAtOne() throws Exception {
@ -1591,13 +1591,13 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestConsumerPrefetchAtTwo() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
}
public void testConsumerPrefetchAtTwo() throws Exception {
@ -1633,13 +1633,13 @@ public class BrokerTest extends BrokerTestSupport {
public void initCombosForTestConsumerPrefetchAndDeliveredAck() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)} );
}
public void testConsumerPrefetchAndDeliveredAck() throws Exception {

2
activemq-core/src/test/java/org/apache/activemq/broker/BrokerTestSupport.java
View File

@ -65,7 +65,7 @@ public class BrokerTestSupport extends CombinationTestSupport {
/**
* Setting this to false makes the test run faster but they may be less accurate.
*/
public static boolean FAST_NO_MESSAGE_LEFT_ASSERT = System.getProperty("FAST_NO_MESSAGE_LEFT_ASSERT", "true").equals("true");
public static final boolean FAST_NO_MESSAGE_LEFT_ASSERT = System.getProperty("FAST_NO_MESSAGE_LEFT_ASSERT", "true").equals("true");
protected RegionBroker regionBroker;
protected BrokerService broker;

48
activemq-core/src/test/java/org/apache/activemq/broker/MessageExpirationTest.java
View File

@ -54,13 +54,13 @@ public class MessageExpirationTest extends BrokerTestSupport {
public void initCombosForTestMessagesWaitingForUssageDecreaseExpire() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE),
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE),
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
} );
}
@ -162,14 +162,14 @@ public class MessageExpirationTest extends BrokerTestSupport {
*/
public void initCombosForTestMessagesInLongTransactionExpire() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
//new Integer(DeliveryMode.PERSISTENT)
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
//Integer.valueOf(DeliveryMode.PERSISTENT)
} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE)
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)
} );
}
@ -232,25 +232,25 @@ public class MessageExpirationTest extends BrokerTestSupport {
public void TestMessagesInSubscriptionPendingListExpire() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE)
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)
} );
}
public void initCombosForTestMessagesInSubscriptionPendingListExpire() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)} );
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.QUEUE_TYPE),
new Byte(ActiveMQDestination.TOPIC_TYPE),
new Byte(ActiveMQDestination.TEMP_QUEUE_TYPE),
new Byte(ActiveMQDestination.TEMP_TOPIC_TYPE)
Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)
} );
}

2
activemq-core/src/test/java/org/apache/activemq/broker/StubBroker.java
View File

@ -58,7 +58,7 @@ public class StubBroker implements Broker {
}
}
public class RemoveConnectionData {
public static class RemoveConnectionData {
public final ConnectionContext connectionContext;
public final ConnectionInfo connectionInfo;
public final Throwable error;

2
activemq-core/src/test/java/org/apache/activemq/broker/virtual/CompositeQueueTest.java
View File

@ -92,7 +92,7 @@ public class CompositeQueueTest extends EmbeddedBrokerTestSupport {
protected TextMessage createMessage(Session session, int i) throws JMSException {
TextMessage textMessage = session.createTextMessage("message: " + i);
if (i % 2 == 1) {
if (i % 2 != 0) {
textMessage.setStringProperty("odd", "yes");
}
textMessage.setIntProperty("i", i);

6
activemq-core/src/test/java/org/apache/activemq/bugs/JmsDurableTopicSlowReceiveTest.java
View File

@ -46,12 +46,12 @@ public class JmsDurableTopicSlowReceiveTest extends JmsTopicSendReceiveTest{
protected MessageProducer producer2;
protected Destination consumerDestination2;
BrokerService broker;
final int NMSG=100;
final int MSIZE=256000;
static final int NMSG=100;
static final int MSIZE=256000;
private Connection connection3;
private Session consumeSession3;
private TopicSubscriber consumer3;
private final String countProperyName = "count";
private static final String countProperyName = "count";
/**
* Set up a durable suscriber test.

10
activemq-core/src/test/java/org/apache/activemq/command/ActiveMQBytesMessageTest.java
View File

@ -243,11 +243,11 @@ public class ActiveMQBytesMessageTest extends TestCase {
try {
msg.writeObject("fred");
msg.writeObject(Boolean.TRUE);
msg.writeObject(new Character('q'));
msg.writeObject(new Byte((byte) 1));
msg.writeObject(new Short((short) 3));
msg.writeObject(new Integer(3));
msg.writeObject(new Long(300l));
msg.writeObject(Character.valueOf('q'));
msg.writeObject(Byte.valueOf((byte) 1));
msg.writeObject(Short.valueOf((short) 3));
msg.writeObject(Integer.valueOf(3));
msg.writeObject(Long.valueOf(300l));
msg.writeObject(new Float(3.3f));
msg.writeObject(new Double(3.3));
msg.writeObject(new byte[3]);

2
activemq-core/src/test/java/org/apache/activemq/command/ActiveMQStreamMessageTest.java
View File

@ -364,7 +364,7 @@ public class ActiveMQStreamMessageTest extends TestCase {
msg.reset();
assertTrue(msg.readLong() == test);
msg.reset();
assertTrue(msg.readString().equals(new Long(test).toString()));
assertTrue(msg.readString().equals(Long.valueOf(test).toString()));
msg.reset();
try {
msg.readBoolean();

42
activemq-core/src/test/java/org/apache/activemq/network/SimpleNetworkTest.java
View File

@ -61,6 +61,7 @@ public class SimpleNetworkTest extends TestCase{
public void testRequestReply() throws Exception{
System.err.println("START TEST!");
final MessageProducer remoteProducer=remoteSession.createProducer(null);
MessageConsumer remoteConsumer=remoteSession.createConsumer(included);
remoteConsumer.setMessageListener(new MessageListener(){
@ -88,9 +89,11 @@ public class SimpleNetworkTest extends TestCase{
assertNotNull(result);
log.info(result.getText());
}
System.err.println("FIN TEST!");
}
public void testFiltering() throws Exception{
public void XtestFiltering() throws Exception{
MessageConsumer includedConsumer=remoteSession.createConsumer(included);
MessageConsumer excludedConsumer=remoteSession.createConsumer(excluded);
MessageProducer includedProducer=localSession.createProducer(included);
@ -101,9 +104,10 @@ public class SimpleNetworkTest extends TestCase{
excludedProducer.send(test);
assertNull(excludedConsumer.receive(500));
assertNotNull(includedConsumer.receive(500));
System.err.println("FIN TEST!");
}
public void testConduitBridge() throws Exception{
public void XtestConduitBridge() throws Exception{
MessageConsumer consumer1=remoteSession.createConsumer(included);
MessageConsumer consumer2=remoteSession.createConsumer(included);
MessageProducer producer=localSession.createProducer(included);
@ -120,7 +124,7 @@ public class SimpleNetworkTest extends TestCase{
assertNull(consumer2.receive(500));
}
public void testDurableStoreAndForward() throws Exception{
public void XtestDurableStoreAndForward() throws Exception{
// create a remote durable consumer
MessageConsumer remoteConsumer=remoteSession.createDurableSubscriber(included,consumerName);
Thread.sleep(1000);
@ -165,19 +169,10 @@ public class SimpleNetworkTest extends TestCase{
}
protected void doSetUp() throws Exception{
Resource resource=new ClassPathResource(getRemoteBrokerURI());
BrokerFactoryBean factory=new BrokerFactoryBean(resource);
factory.afterPropertiesSet();
remoteBroker=factory.getBroker();
remoteBroker=createRemoteBroker();
remoteBroker.start();
resource=new ClassPathResource(getLocalBrokerURI());
factory=new BrokerFactoryBean(resource);
factory.afterPropertiesSet();
localBroker=factory.getBroker();
localBroker=createLocalBroker();
localBroker.start();
URI localURI=localBroker.getVmConnectorURI();
ActiveMQConnectionFactory fac=new ActiveMQConnectionFactory(localURI);
localConnection=fac.createConnection();
@ -193,6 +188,7 @@ public class SimpleNetworkTest extends TestCase{
localSession=localConnection.createSession(false,Session.AUTO_ACKNOWLEDGE);
remoteSession=remoteConnection.createSession(false,Session.AUTO_ACKNOWLEDGE);
}
protected String getRemoteBrokerURI() {
return "org/apache/activemq/network/remoteBroker.xml";
@ -201,4 +197,22 @@ public class SimpleNetworkTest extends TestCase{
protected String getLocalBrokerURI() {
return "org/apache/activemq/network/localBroker.xml";
}
protected BrokerService createBroker(String URI) throws Exception {
Resource resource=new ClassPathResource(URI);
BrokerFactoryBean factory=new BrokerFactoryBean(resource);
resource=new ClassPathResource(URI);
factory=new BrokerFactoryBean(resource);
factory.afterPropertiesSet();
BrokerService result=factory.getBroker();
return result;
}
protected BrokerService createLocalBroker() throws Exception {
return createBroker(getLocalBrokerURI());
}
protected BrokerService createRemoteBroker() throws Exception {
return createBroker(getRemoteBrokerURI());
}
}

2
activemq-core/src/test/java/org/apache/activemq/openwire/v1/ActiveMQTextMessageTest.java
View File

@ -39,7 +39,7 @@ import org.apache.activemq.command.*;
public class ActiveMQTextMessageTest extends ActiveMQMessageTest {
public static ActiveMQTextMessageTest SINGLETON = new ActiveMQTextMessageTest();
public static final ActiveMQTextMessageTest SINGLETON = new ActiveMQTextMessageTest();
public Object createObject() throws Exception {
ActiveMQTextMessage info = new ActiveMQTextMessage();

2
activemq-core/src/test/java/org/apache/activemq/openwire/v1/BrokerInfoTest.java
View File

@ -39,7 +39,7 @@ import org.apache.activemq.command.*;
public class BrokerInfoTest extends BaseCommandTestSupport {
public static BrokerInfoTest SINGLETON = new BrokerInfoTest();
public static final BrokerInfoTest SINGLETON = new BrokerInfoTest();
public Object createObject() throws Exception {
BrokerInfo info = new BrokerInfo();

2
activemq-core/src/test/java/org/apache/activemq/openwire/v1/MessageAckTest.java
View File

@ -39,7 +39,7 @@ import org.apache.activemq.command.*;
public class MessageAckTest extends BaseCommandTestSupport {
public static MessageAckTest SINGLETON = new MessageAckTest();
public static final MessageAckTest SINGLETON = new MessageAckTest();
public Object createObject() throws Exception {
MessageAck info = new MessageAck();

2
activemq-core/src/test/java/org/apache/activemq/perf/AMQStoreQueueTest.java
View File

@ -36,7 +36,7 @@ public class AMQStoreQueueTest extends SimpleQueueTest{
answer.setPersistenceAdapter(adaptor);
answer.addConnector(bindAddress);
answer.setDeleteAllMessagesOnStartup(true);
//answer.setDeleteAllMessagesOnStartup(true);
}

2
activemq-core/src/test/java/org/apache/activemq/perf/SimpleTopicTest.java
View File

@ -43,7 +43,7 @@ public class SimpleTopicTest extends TestCase{
protected String DESTINATION_NAME=getClass().getName();
protected int SAMPLE_COUNT=10;
protected long SAMPLE_INTERVAL=1000;
protected int NUMBER_OF_CONSUMERS=1;
protected int NUMBER_OF_CONSUMERS=10;
protected int NUMBER_OF_PRODUCERS=1;
protected int PAYLOAD_SIZE=1024;
protected byte[] array=null;

6
activemq-core/src/test/java/org/apache/activemq/proxy/ProxyConnectorTest.java
View File

@ -51,11 +51,11 @@ public class ProxyConnectorTest extends ProxyTestSupport {
public void initCombosForTestSendAndConsume() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)
} );
addCombinationValues( "destinationType", new Object[]{
new Byte(ActiveMQDestination.TOPIC_TYPE),
Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
} );
}
public void testSendAndConsume() throws Exception {

1
activemq-core/src/test/java/org/apache/activemq/security/LDAPAuthorizationMapTest.java
View File

@ -48,7 +48,6 @@ import junit.framework.TestCase;
*
*/
public class LDAPAuthorizationMapTest extends TestCase {
private HashMap options;
private LDAPAuthorizationMap authMap;
protected void setUp() throws Exception {

2
activemq-core/src/test/java/org/apache/activemq/security/SimpleSecurityBrokerSystemTest.java
View File

@ -104,7 +104,7 @@ public class SimpleSecurityBrokerSystemTest extends SecurityTestSupport {
return new SimpleAuthorizationMap(writeAccess, readAccess, adminAccess);
}
class SimpleAuthenticationFactory implements BrokerPlugin {
static class SimpleAuthenticationFactory implements BrokerPlugin {
public Broker installPlugin(Broker broker) {
HashMap u = new HashMap();

8
activemq-core/src/test/java/org/apache/activemq/test/message/NestedMapMessageTest.java
View File

@ -49,8 +49,8 @@ public class NestedMapMessageTest extends JmsTopicSendReceiveWithTwoConnectionsA
Map map = (Map) mapMessage.getObject("mapField");
assertNotNull(map);
assertEquals("mapField.a", "foo", map.get("a"));
assertEquals("mapField.b", new Integer(23), map.get("b"));
assertEquals("mapField.c", new Long(45), map.get("c"));
assertEquals("mapField.b", Integer.valueOf(23), map.get("b"));
assertEquals("mapField.c", Long.valueOf(45), map.get("c"));
value = map.get("d");
assertTrue("mapField.d should be a Map", value instanceof Map);
@ -84,8 +84,8 @@ public class NestedMapMessageTest extends JmsTopicSendReceiveWithTwoConnectionsA
Map nestedMap = new HashMap();
nestedMap.put("a", "foo");
nestedMap.put("b", new Integer(23));
nestedMap.put("c", new Long(45));
nestedMap.put("b", Integer.valueOf(23));
nestedMap.put("c", Long.valueOf(45));
nestedMap.put("d", grandChildMap);
answer.setObject("mapField", nestedMap);

2
activemq-core/src/test/java/org/apache/activemq/test/retroactive/DummyMessageQuery.java
View File

@ -34,7 +34,7 @@ public class DummyMessageQuery implements MessageQuery {
protected static final Log log = LogFactory.getLog(DummyMessageQuery.class);
public static int messageCount = 10;
public static final int messageCount = 10;
public void execute(ActiveMQDestination destination, MessageListener listener) throws Exception {
log.info("Initial query is creating: " + messageCount + " messages");

4
activemq-core/src/test/java/org/apache/activemq/transport/TopicClusterTest.java
View File

@ -54,8 +54,8 @@ public class TopicClusterTest extends TestCase implements MessageListener {
protected Destination destination;
protected boolean topic = true;
protected AtomicInteger receivedMessageCount = new AtomicInteger(0);
protected static int MESSAGE_COUNT = 50;
protected static int NUMBER_IN_CLUSTER = 3;
protected static final int MESSAGE_COUNT = 50;
protected static final int NUMBER_IN_CLUSTER = 3;
protected int deliveryMode = DeliveryMode.NON_PERSISTENT;
protected MessageProducer[] producers;
protected Connection[] connections;

4
activemq-core/src/test/java/org/apache/activemq/transport/failover/FailoverTransportBrokerTest.java
View File

@ -47,8 +47,8 @@ public class FailoverTransportBrokerTest extends NetworkTestSupport {
public void initCombosForTestPublisherFailsOver() {
addCombinationValues( "deliveryMode", new Object[]{
new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT)
Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT)
} );
addCombinationValues( "destination", new Object[]{
new ActiveMQQueue("TEST"),

8
activemq-core/src/test/java/org/apache/activemq/transport/fanout/FanoutTransportBrokerTest.java
View File

@ -61,8 +61,8 @@ public class FanoutTransportBrokerTest extends NetworkTestSupport {
}
public void initCombosForTestPublisherFansout() {
addCombinationValues("deliveryMode", new Object[] { new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT) });
addCombinationValues("deliveryMode", new Object[] { Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT) });
addCombinationValues("destination", new Object[] { new ActiveMQQueue("TEST"), new ActiveMQTopic("TEST"), });
}
@ -109,8 +109,8 @@ public class FanoutTransportBrokerTest extends NetworkTestSupport {
public void initCombosForTestPublisherWaitsForServerToBeUp() {
addCombinationValues("deliveryMode", new Object[] { new Integer(DeliveryMode.NON_PERSISTENT),
new Integer(DeliveryMode.PERSISTENT) });
addCombinationValues("deliveryMode", new Object[] { Integer.valueOf(DeliveryMode.NON_PERSISTENT),
Integer.valueOf(DeliveryMode.PERSISTENT) });
addCombinationValues("destination", new Object[] { new ActiveMQQueue("TEST"), new ActiveMQTopic("TEST"), });
}
public void testPublisherWaitsForServerToBeUp() throws Exception {

4
activemq-core/src/test/java/org/apache/activemq/transport/peer/PeerTransportTest.java
View File

@ -48,8 +48,8 @@ public class PeerTransportTest extends TestCase {
protected Log log = LogFactory.getLog(getClass());
protected ActiveMQDestination destination;
protected boolean topic = true;
protected static int MESSAGE_COUNT = 50;
protected static int NUMBER_IN_CLUSTER = 3;
protected static final int MESSAGE_COUNT = 50;
protected static final int NUMBER_IN_CLUSTER = 3;
protected int deliveryMode = DeliveryMode.NON_PERSISTENT;
protected MessageProducer[] producers;
protected Connection[] connections;

4
activemq-core/src/test/java/org/apache/activemq/transport/tcp/InactivityMonitorTest.java
View File

@ -204,8 +204,8 @@ public class InactivityMonitorTest extends CombinationTestSupport implements Tra
startClient();
addCombinationValues("clientInactivityLimit", new Object[] { new Long(1000)});
addCombinationValues("serverInactivityLimit", new Object[] { new Long(1000)});
addCombinationValues("clientInactivityLimit", new Object[] { Long.valueOf(1000)});
addCombinationValues("serverInactivityLimit", new Object[] { Long.valueOf(1000)});
addCombinationValues("serverRunOnCommand", new Object[] { new Runnable() {
public void run() {
try {

1
activemq-core/src/test/java/org/apache/activemq/transport/tcp/SslTransportTest.java
View File

@ -37,7 +37,6 @@ import org.apache.activemq.wireformat.ObjectStreamWireFormat;
public class SslTransportTest extends TestCase {
SSLSocket sslSocket;
SslTransport transport;
StubTransportListener stubListener;
String username;

2
activemq-core/src/test/java/org/apache/activemq/usecases/AMQDeadlockTestW4Brokers.java
View File

@ -241,7 +241,7 @@ public class AMQDeadlockTestW4Brokers extends TestCase {
return acf;
}
private class TestMessageListener1 implements MessageListener {
private static class TestMessageListener1 implements MessageListener {
private final long waitTime;
final AtomicInteger count = new AtomicInteger(0);

2
activemq-core/src/test/java/org/apache/activemq/usecases/AMQFailoverIssue.java
View File

@ -174,7 +174,7 @@ public class AMQFailoverIssue extends TestCase{
}
}
private class PooledProducerTask implements Runnable{
private static class PooledProducerTask implements Runnable{
private final String queueName;
private final PooledConnectionFactory pcf;

2
activemq-core/src/test/java/org/apache/activemq/usecases/ChangeSentMessageTest.java
View File

@ -49,7 +49,7 @@ public class ChangeSentMessageTest extends TestSupport {
HashMap map = new HashMap();
ObjectMessage message = publisherSession.createObjectMessage();
for (int i = 0;i < COUNT;i++) {
map.put(VALUE_NAME, new Integer(i));
map.put(VALUE_NAME, Integer.valueOf(i));
message.setObject(map);
producer.send(message);
assertTrue(message.getObject()==map);

2
activemq-core/src/test/java/org/apache/activemq/usecases/TopicRedeliverTest.java
View File

@ -45,7 +45,7 @@ public class TopicRedeliverTest extends TestSupport {
super(n);
}
protected void setup() throws Exception{
protected void setUp() throws Exception{
super.setUp();
topic = true;
}