You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ch...@apache.org on 2007/08/10 18:57:10 UTC
svn commit: r564679 [6/8] - in /activemq/trunk:
activemq-core/src/main/java/org/apache/activemq/
activemq-core/src/main/java/org/apache/activemq/broker/
activemq-core/src/main/java/org/apache/activemq/broker/jmx/
activemq-core/src/main/java/org/apache/...
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/SessionState.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/SessionState.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/SessionState.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/SessionState.java Fri Aug 10 09:57:01 2007
@@ -18,6 +18,7 @@
package org.apache.activemq.state;
import java.util.Collection;
+import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -31,8 +32,8 @@
public class SessionState {
final SessionInfo info;
- public final ConcurrentHashMap producers = new ConcurrentHashMap();
- public final ConcurrentHashMap consumers = new ConcurrentHashMap();
+ private final Map<ProducerId, ProducerState> producers = new ConcurrentHashMap<ProducerId, ProducerState>();
+ private final Map<ConsumerId, ConsumerState> consumers = new ConcurrentHashMap<ConsumerId, ConsumerState>();
private final AtomicBoolean shutdown = new AtomicBoolean(false);
public SessionState(SessionInfo info) {
@@ -49,7 +50,7 @@
}
public ProducerState removeProducer(ProducerId id) {
- return (ProducerState)producers.remove(id);
+ return producers.remove(id);
}
public void addConsumer(ConsumerInfo info) {
@@ -58,40 +59,41 @@
}
public ConsumerState removeConsumer(ConsumerId id) {
- return (ConsumerState)consumers.remove(id);
+ return consumers.remove(id);
}
public SessionInfo getInfo() {
return info;
}
- public Set getConsumerIds() {
+ public Set<ConsumerId> getConsumerIds() {
return consumers.keySet();
}
- public Set getProducerIds() {
+ public Set<ProducerId> getProducerIds() {
return producers.keySet();
}
- public Collection getProducerStates() {
+ public Collection<ProducerState> getProducerStates() {
return producers.values();
}
public ProducerState getProducerState(ProducerId producerId) {
- return (ProducerState)producers.get(producerId);
+ return producers.get(producerId);
}
- public Collection getConsumerStates() {
+ public Collection<ConsumerState> getConsumerStates() {
return consumers.values();
}
public ConsumerState getConsumerState(ConsumerId consumerId) {
- return (ConsumerState)consumers.get(consumerId);
+ return consumers.get(consumerId);
}
private void checkShutdown() {
- if (shutdown.get())
+ if (shutdown.get()) {
throw new IllegalStateException("Disposed");
+ }
}
public void shutdown() {
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/TransactionState.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/TransactionState.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/TransactionState.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/TransactionState.java Fri Aug 10 09:57:01 2007
@@ -24,13 +24,11 @@
import org.apache.activemq.command.TransactionId;
public class TransactionState {
- final TransactionId id;
- public final ArrayList commands = new ArrayList();
+ private final List<Command> commands = new ArrayList<Command>();
+ private final TransactionId id;
private final AtomicBoolean shutdown = new AtomicBoolean(false);
-
private boolean prepared;
-
private int preparedResult;
public TransactionState(TransactionId id) {
@@ -46,13 +44,14 @@
commands.add(operation);
}
- public List getCommands() {
+ public List<Command> getCommands() {
return commands;
}
private void checkShutdown() {
- if (shutdown.get())
+ if (shutdown.get()) {
throw new IllegalStateException("Disposed");
+ }
}
public void shutdown() {
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQMessageStore.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQMessageStore.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQMessageStore.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQMessageStore.java Fri Aug 10 09:57:01 2007
@@ -23,7 +23,10 @@
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
import java.util.Map.Entry;
+import java.util.Set;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicReference;
@@ -57,20 +60,22 @@
public class AMQMessageStore implements MessageStore {
private static final Log LOG = LogFactory.getLog(AMQMessageStore.class);
+
protected final AMQPersistenceAdapter peristenceAdapter;
protected final AMQTransactionStore transactionStore;
protected final ReferenceStore referenceStore;
protected final ActiveMQDestination destination;
protected final TransactionTemplate transactionTemplate;
- private LinkedHashMap<MessageId, ReferenceData> messages = new LinkedHashMap<MessageId, ReferenceData>();
- private ArrayList<MessageAck> messageAcks = new ArrayList<MessageAck>();
- /** A MessageStore that we can use to retrieve messages quickly. */
- private LinkedHashMap<MessageId, ReferenceData> cpAddedMessageIds;
protected Location lastLocation;
protected Location lastWrittenLocation;
- protected HashSet<Location> inFlightTxLocations = new HashSet<Location>();
+ protected Set<Location> inFlightTxLocations = new HashSet<Location>();
protected final TaskRunner asyncWriteTask;
protected CountDownLatch flushLatch;
+
+ private Map<MessageId, ReferenceData> messages = new LinkedHashMap<MessageId, ReferenceData>();
+ private List<MessageAck> messageAcks = new ArrayList<MessageAck>();
+ /** A MessageStore that we can use to retrieve messages quickly. */
+ private Map<MessageId, ReferenceData> cpAddedMessageIds;
private final boolean debug = LOG.isDebugEnabled();
private final AtomicReference<Location> mark = new AtomicReference<Location>();
@@ -304,8 +309,8 @@
* @throws IOException
*/
protected Location doAsyncWrite() throws IOException {
- final ArrayList<MessageAck> cpRemovedMessageLocations;
- final ArrayList<Location> cpActiveJournalLocations;
+ final List<MessageAck> cpRemovedMessageLocations;
+ final List<Location> cpActiveJournalLocations;
final int maxCheckpointMessageAddSize = peristenceAdapter.getMaxCheckpointMessageAddSize();
final Location lastLocation;
// swap out the message hash maps..
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQTransactionStore.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQTransactionStore.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQTransactionStore.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/amq/AMQTransactionStore.java Fri Aug 10 09:57:01 2007
@@ -36,9 +36,10 @@
*/
public class AMQTransactionStore implements TransactionStore {
- private final AMQPersistenceAdapter peristenceAdapter;
Map<TransactionId, AMQTx> inflightTransactions = new LinkedHashMap<TransactionId, AMQTx>();
Map<TransactionId, AMQTx> preparedTransactions = new LinkedHashMap<TransactionId, AMQTx>();
+
+ private final AMQPersistenceAdapter peristenceAdapter;
private boolean doingRecover;
public AMQTransactionStore(AMQPersistenceAdapter adapter) {
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/JDBCMessageStore.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/JDBCMessageStore.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/JDBCMessageStore.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/JDBCMessageStore.java Fri Aug 10 09:57:01 2007
@@ -95,8 +95,9 @@
TransactionContext c = persistenceAdapter.getTransactionContext();
try {
byte data[] = adapter.doGetMessage(c, id);
- if (data == null)
+ if (data == null) {
return null;
+ }
Message answer = (Message)wireFormat.unmarshal(new ByteSequence(data));
return answer;
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/TransactionContext.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/TransactionContext.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/TransactionContext.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/TransactionContext.java Fri Aug 10 09:57:01 2007
@@ -88,8 +88,9 @@
}
private void executeBatch(PreparedStatement p, String message) throws SQLException {
- if (p == null)
+ if (p == null) {
return;
+ }
try {
int[] rc = p.executeBatch();
@@ -143,19 +144,22 @@
}
public void begin() throws IOException {
- if (inTx)
+ if (inTx) {
throw new IOException("Already started.");
+ }
inTx = true;
connection = getConnection();
}
public void commit() throws IOException {
- if (!inTx)
+ if (!inTx) {
throw new IOException("Not started.");
+ }
try {
executeBatch();
- if (!connection.getAutoCommit())
+ if (!connection.getAutoCommit()) {
connection.commit();
+ }
} catch (SQLException e) {
JDBCPersistenceAdapter.log("Commit failed: ", e);
throw IOExceptionSupport.create(e);
@@ -166,8 +170,9 @@
}
public void rollback() throws IOException {
- if (!inTx)
+ if (!inTx) {
throw new IOException("Not started.");
+ }
try {
if (addMessageStatement != null) {
addMessageStatement.close();
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/AxionJDBCAdapter.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/AxionJDBCAdapter.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/AxionJDBCAdapter.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/AxionJDBCAdapter.java Fri Aug 10 09:57:01 2007
@@ -52,7 +52,7 @@
+ ", SELECTOR " + statements.getStringIdDataType()
+ ", LAST_ACKED_ID " + statements.getSequenceDataType()
+ ", PRIMARY KEY ( CONTAINER, CLIENT_ID, SUB_NAME))"
- };
+ };
statements.setCreateSchemaStatements(createStatements);
statements.setDeleteOldMessagesStatement("DELETE FROM " + statements.getFullMessageTableName() + " WHERE ( EXPIRATION<>0 AND EXPIRATION<?)");
statements.setLongDataType("LONG");
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/BlobJDBCAdapter.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/BlobJDBCAdapter.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/BlobJDBCAdapter.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/jdbc/adapter/BlobJDBCAdapter.java Fri Aug 10 09:57:01 2007
@@ -62,16 +62,18 @@
s.setString(3, messageID);
s.setString(4, " ");
- if (s.executeUpdate() != 1)
+ if (s.executeUpdate() != 1) {
throw new JMSException("Failed to broker message: " + messageID + " in container.");
+ }
s.close();
// Select the blob record so that we can update it.
s = c.prepareStatement(statements.getFindMessageStatement());
s.setLong(1, seq);
rs = s.executeQuery();
- if (!rs.next())
+ if (!rs.next()) {
throw new JMSException("Failed to broker message: " + messageID + " in container.");
+ }
// Update the blob
Blob blob = rs.getBlob(1);
@@ -90,11 +92,11 @@
} finally {
try {
rs.close();
- } catch (Throwable e) {
+ } catch (Throwable ignore) {
}
try {
s.close();
- } catch (Throwable e) {
+ } catch (Throwable ignore) {
}
}
}
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/journal/JournalMessageStore.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/journal/JournalMessageStore.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/journal/JournalMessageStore.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/journal/JournalMessageStore.java Fri Aug 10 09:57:01 2007
@@ -22,6 +22,9 @@
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
import org.apache.activeio.journal.RecordLocation;
import org.apache.activemq.broker.ConnectionContext;
@@ -47,22 +50,22 @@
*/
public class JournalMessageStore implements MessageStore {
- private static final Log log = LogFactory.getLog(JournalMessageStore.class);
+ private static final Log LOG = LogFactory.getLog(JournalMessageStore.class);
protected final JournalPersistenceAdapter peristenceAdapter;
protected final JournalTransactionStore transactionStore;
protected final MessageStore longTermStore;
protected final ActiveMQDestination destination;
protected final TransactionTemplate transactionTemplate;
+ protected RecordLocation lastLocation;
+ protected Set<RecordLocation> inFlightTxLocations = new HashSet<RecordLocation>();
- private LinkedHashMap messages = new LinkedHashMap();
- private ArrayList messageAcks = new ArrayList();
+ private Map<MessageId, Message> messages = new LinkedHashMap<MessageId, Message>();
+ private List<MessageAck> messageAcks = new ArrayList<MessageAck>();
/** A MessageStore that we can use to retrieve messages quickly. */
- private LinkedHashMap cpAddedMessageIds;
+ private Map<MessageId, Message> cpAddedMessageIds;
- protected RecordLocation lastLocation;
- protected HashSet inFlightTxLocations = new HashSet();
private UsageManager usageManager;
@@ -87,18 +90,18 @@
final MessageId id = message.getMessageId();
- final boolean debug = log.isDebugEnabled();
+ final boolean debug = LOG.isDebugEnabled();
message.incrementReferenceCount();
final RecordLocation location = peristenceAdapter.writeCommand(message, message.isResponseRequired());
if (!context.isInTransaction()) {
if (debug) {
- log.debug("Journalled message add for: " + id + ", at: " + location);
+ LOG.debug("Journalled message add for: " + id + ", at: " + location);
}
addMessage(message, location);
} else {
if (debug) {
- log.debug("Journalled transacted message add for: " + id + ", at: " + location);
+ LOG.debug("Journalled transacted message add for: " + id + ", at: " + location);
}
synchronized (this) {
inFlightTxLocations.add(location);
@@ -107,7 +110,7 @@
context.getTransaction().addSynchronization(new Synchronization() {
public void afterCommit() throws Exception {
if (debug) {
- log.debug("Transacted message add commit for: " + id + ", at: " + location);
+ LOG.debug("Transacted message add commit for: " + id + ", at: " + location);
}
synchronized (JournalMessageStore.this) {
inFlightTxLocations.remove(location);
@@ -117,7 +120,7 @@
public void afterRollback() throws Exception {
if (debug) {
- log.debug("Transacted message add rollback for: " + id + ", at: " + location);
+ LOG.debug("Transacted message add rollback for: " + id + ", at: " + location);
}
synchronized (JournalMessageStore.this) {
inFlightTxLocations.remove(location);
@@ -144,14 +147,14 @@
longTermStore.addMessage(context, message);
}
} catch (Throwable e) {
- log.warn("Could not replay add for message '" + message.getMessageId() + "'. Message may have already been added. reason: " + e);
+ LOG.warn("Could not replay add for message '" + message.getMessageId() + "'. Message may have already been added. reason: " + e);
}
}
/**
*/
public void removeMessage(ConnectionContext context, final MessageAck ack) throws IOException {
- final boolean debug = log.isDebugEnabled();
+ final boolean debug = LOG.isDebugEnabled();
JournalQueueAck remove = new JournalQueueAck();
remove.setDestination(destination);
remove.setMessageAck(ack);
@@ -159,13 +162,12 @@
final RecordLocation location = peristenceAdapter.writeCommand(remove, ack.isResponseRequired());
if (!context.isInTransaction()) {
if (debug) {
- log.debug("Journalled message remove for: " + ack.getLastMessageId() + ", at: " + location);
+ LOG.debug("Journalled message remove for: " + ack.getLastMessageId() + ", at: " + location);
}
removeMessage(ack, location);
} else {
if (debug) {
- log.debug("Journalled transacted message remove for: " + ack.getLastMessageId() + ", at: "
- + location);
+ LOG.debug("Journalled transacted message remove for: " + ack.getLastMessageId() + ", at: " + location);
}
synchronized (this) {
inFlightTxLocations.add(location);
@@ -174,8 +176,7 @@
context.getTransaction().addSynchronization(new Synchronization() {
public void afterCommit() throws Exception {
if (debug) {
- log.debug("Transacted message remove commit for: " + ack.getLastMessageId()
- + ", at: " + location);
+ LOG.debug("Transacted message remove commit for: " + ack.getLastMessageId() + ", at: " + location);
}
synchronized (JournalMessageStore.this) {
inFlightTxLocations.remove(location);
@@ -184,9 +185,9 @@
}
public void afterRollback() throws Exception {
- if (debug)
- log.debug("Transacted message remove rollback for: " + ack.getLastMessageId()
- + ", at: " + location);
+ if (debug) {
+ LOG.debug("Transacted message remove rollback for: " + ack.getLastMessageId() + ", at: " + location);
+ }
synchronized (JournalMessageStore.this) {
inFlightTxLocations.remove(location);
}
@@ -200,7 +201,7 @@
synchronized (this) {
lastLocation = location;
MessageId id = ack.getLastMessageId();
- Message message = (Message)messages.remove(id);
+ Message message = messages.remove(id);
if (message == null) {
messageAcks.add(ack);
} else {
@@ -217,7 +218,7 @@
longTermStore.removeMessage(context, messageAck);
}
} catch (Throwable e) {
- log.warn("Could not replay acknowledge for message '" + messageAck.getLastMessageId() + "'. Message may have already been acknowledged. reason: " + e);
+ LOG.warn("Could not replay acknowledge for message '" + messageAck.getLastMessageId() + "'. Message may have already been acknowledged. reason: " + e);
}
}
@@ -235,9 +236,8 @@
*/
public RecordLocation checkpoint(final Callback postCheckpointTest) throws IOException {
- RecordLocation rc;
- final ArrayList cpRemovedMessageLocations;
- final ArrayList cpActiveJournalLocations;
+ final List<MessageAck> cpRemovedMessageLocations;
+ final List<RecordLocation> cpActiveJournalLocations;
final int maxCheckpointMessageAddSize = peristenceAdapter.getMaxCheckpointMessageAddSize();
// swap out the message hash maps..
@@ -245,10 +245,10 @@
cpAddedMessageIds = this.messages;
cpRemovedMessageLocations = this.messageAcks;
- cpActiveJournalLocations = new ArrayList(inFlightTxLocations);
+ cpActiveJournalLocations = new ArrayList<RecordLocation>(inFlightTxLocations);
- this.messages = new LinkedHashMap();
- this.messageAcks = new ArrayList();
+ this.messages = new LinkedHashMap<MessageId, Message>();
+ this.messageAcks = new ArrayList<MessageAck>();
}
transactionTemplate.run(new Callback() {
@@ -261,13 +261,13 @@
// Checkpoint the added messages.
synchronized (JournalMessageStore.this) {
- Iterator iterator = cpAddedMessageIds.values().iterator();
+ Iterator<Message> iterator = cpAddedMessageIds.values().iterator();
while (iterator.hasNext()) {
- Message message = (Message)iterator.next();
+ Message message = iterator.next();
try {
longTermStore.addMessage(context, message);
} catch (Throwable e) {
- log.warn("Message could not be added to long term store: " + e.getMessage(), e);
+ LOG.warn("Message could not be added to long term store: " + e.getMessage(), e);
}
size += message.getSize();
message.decrementReferenceCount();
@@ -284,13 +284,13 @@
persitanceAdapter.beginTransaction(context);
// Checkpoint the removed messages.
- Iterator iterator = cpRemovedMessageLocations.iterator();
+ Iterator<MessageAck> iterator = cpRemovedMessageLocations.iterator();
while (iterator.hasNext()) {
try {
- MessageAck ack = (MessageAck)iterator.next();
+ MessageAck ack = iterator.next();
longTermStore.removeMessage(transactionTemplate.getContext(), ack);
} catch (Throwable e) {
- log.debug("Message could not be removed from long term store: " + e.getMessage(), e);
+ LOG.debug("Message could not be removed from long term store: " + e.getMessage(), e);
}
}
@@ -307,7 +307,7 @@
if (cpActiveJournalLocations.size() > 0) {
Collections.sort(cpActiveJournalLocations);
- return (RecordLocation)cpActiveJournalLocations.get(0);
+ return cpActiveJournalLocations.get(0);
}
synchronized (this) {
return lastLocation;
@@ -322,9 +322,10 @@
synchronized (this) {
// Do we have a still have it in the journal?
- answer = (Message)messages.get(identity);
- if (answer == null && cpAddedMessageIds != null)
- answer = (Message)cpAddedMessageIds.get(identity);
+ answer = messages.get(identity);
+ if (answer == null && cpAddedMessageIds != null) {
+ answer = cpAddedMessageIds.get(identity);
+ }
}
if (answer != null) {
@@ -349,15 +350,17 @@
}
public void start() throws Exception {
- if (this.usageManager != null)
+ if (this.usageManager != null) {
this.usageManager.addUsageListener(peristenceAdapter);
+ }
longTermStore.start();
}
public void stop() throws Exception {
longTermStore.stop();
- if (this.usageManager != null)
+ if (this.usageManager != null) {
this.usageManager.removeUsageListener(peristenceAdapter);
+ }
}
/**
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaPersistenceAdapter.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaPersistenceAdapter.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaPersistenceAdapter.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/kahadaptor/KahaPersistenceAdapter.java Fri Aug 10 09:57:01 2007
@@ -212,9 +212,9 @@
return new CommandMarshaller(wireFormat);
}
- protected ListContainer getListContainer(Object id, String containerName) throws IOException {
+ protected ListContainer<TopicSubAck> getListContainer(Object id, String containerName) throws IOException {
Store store = getStore();
- ListContainer container = store.getListContainer(id, containerName);
+ ListContainer<TopicSubAck> container = store.getListContainer(id, containerName);
container.setMarshaller(createMessageMarshaller());
container.load();
return container;
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/TransportFactory.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/TransportFactory.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/TransportFactory.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/TransportFactory.java Fri Aug 10 09:57:01 2007
@@ -35,6 +35,10 @@
public abstract class TransportFactory {
+ private static final FactoryFinder TRANSPORT_FACTORY_FINDER = new FactoryFinder("META-INF/services/org/apache/activemq/transport/");
+ private static final FactoryFinder WIREFORMAT_FACTORY_FINDER = new FactoryFinder("META-INF/services/org/apache/activemq/wireformat/");
+ private static final ConcurrentHashMap TRANSPORT_FACTORYS = new ConcurrentHashMap();
+
public abstract TransportServer doBind(String brokerId, URI location) throws IOException;
public Transport doConnect(URI location, Executor ex) throws Exception {
@@ -44,11 +48,6 @@
public Transport doCompositeConnect(URI location, Executor ex) throws Exception {
return doCompositeConnect(location);
}
-
- private static final FactoryFinder TRANSPORT_FACTORY_FINDER = new FactoryFinder("META-INF/services/org/apache/activemq/transport/");
- private static final FactoryFinder WIREFORMAT_FACTORY_FINDER = new FactoryFinder("META-INF/services/org/apache/activemq/wireformat/");
-
- private static final ConcurrentHashMap TRANSPORT_FACTORYS = new ConcurrentHashMap();
/**
* Creates a normal transport.
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/discovery/rendezvous/JmDNSFactory.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/discovery/rendezvous/JmDNSFactory.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/discovery/rendezvous/JmDNSFactory.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/discovery/rendezvous/JmDNSFactory.java Fri Aug 10 09:57:01 2007
@@ -26,7 +26,7 @@
public class JmDNSFactory {
- static Map registry = new HashMap();
+ static Map<InetAddress, UsageTracker> registry = new HashMap<InetAddress, UsageTracker>();
static class UsageTracker {
AtomicInteger count = new AtomicInteger(0);
@@ -34,7 +34,7 @@
}
static synchronized JmDNS create(final InetAddress address) throws IOException {
- UsageTracker tracker = (UsageTracker)registry.get(address);
+ UsageTracker tracker = registry.get(address);
if (tracker == null) {
tracker = new UsageTracker();
tracker.jmDNS = new JmDNS(address) {
@@ -51,7 +51,7 @@
}
static synchronized boolean onClose(InetAddress address) {
- UsageTracker tracker = (UsageTracker)registry.get(address);
+ UsageTracker tracker = registry.get(address);
if (tracker != null) {
if (tracker.count.decrementAndGet() == 0) {
registry.remove(address);
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/FrameTranslator.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/FrameTranslator.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/FrameTranslator.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/FrameTranslator.java Fri Aug 10 09:57:01 2007
@@ -51,7 +51,7 @@
}
public static void copyStandardHeadersFromMessageToFrame(ActiveMQMessage message, StompFrame command, FrameTranslator ft) throws IOException {
- final Map headers = command.getHeaders();
+ final Map<String, String> headers = command.getHeaders();
headers.put(Stomp.Headers.Message.DESTINATION, ft.convertDestination(message.getDestination()));
headers.put(Stomp.Headers.Message.MESSAGE_ID, message.getJMSMessageID());
@@ -82,12 +82,12 @@
}
public static void copyStandardHeadersFromFrameToMessage(StompFrame command, ActiveMQMessage msg, FrameTranslator ft) throws ProtocolException, JMSException {
- final Map headers = new HashMap(command.getHeaders());
- final String destination = (String)headers.remove(Stomp.Headers.Send.DESTINATION);
+ final Map<String, String> headers = new HashMap<String, String>(command.getHeaders());
+ final String destination = headers.remove(Stomp.Headers.Send.DESTINATION);
msg.setDestination(ft.convertDestination(destination));
// the standard JMS headers
- msg.setJMSCorrelationID((String)headers.remove(Stomp.Headers.Send.CORRELATION_ID));
+ msg.setJMSCorrelationID(headers.remove(Stomp.Headers.Send.CORRELATION_ID));
Object o = headers.remove(Stomp.Headers.Send.EXPIRATION_TIME);
if (o != null) {
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/ProtocolConverter.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/ProtocolConverter.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/ProtocolConverter.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/ProtocolConverter.java Fri Aug 10 09:57:01 2007
@@ -66,9 +66,9 @@
private final LongSequenceGenerator messageIdGenerator = new LongSequenceGenerator();
private final LongSequenceGenerator transactionIdGenerator = new LongSequenceGenerator();
- private final ConcurrentHashMap resposeHandlers = new ConcurrentHashMap();
- private final ConcurrentHashMap subscriptionsByConsumerId = new ConcurrentHashMap();
- private final Map transactions = new ConcurrentHashMap();
+ private final ConcurrentHashMap<Integer, ResponseHandler> resposeHandlers = new ConcurrentHashMap<Integer, ResponseHandler>();
+ private final ConcurrentHashMap<ConsumerId, StompSubscription> subscriptionsByConsumerId = new ConcurrentHashMap<ConsumerId, StompSubscription>();
+ private final Map<String, LocalTransactionId> transactions = new ConcurrentHashMap<String, LocalTransactionId>();
private final StompTransportFilter transportFilter;
private final Object commnadIdMutex = new Object();
@@ -88,14 +88,14 @@
}
protected ResponseHandler createResponseHandler(StompFrame command) {
- final String receiptId = (String)command.getHeaders().get(Stomp.Headers.RECEIPT_REQUESTED);
+ final String receiptId = command.getHeaders().get(Stomp.Headers.RECEIPT_REQUESTED);
// A response may not be needed.
if (receiptId != null) {
return new ResponseHandler() {
public void onResponse(ProtocolConverter converter, Response response) throws IOException {
StompFrame sc = new StompFrame();
sc.setAction(Stomp.Responses.RECEIPT);
- sc.setHeaders(new HashMap(1));
+ sc.setHeaders(new HashMap<String, String>(1));
sc.getHeaders().put(Stomp.Headers.Response.RECEIPT_ID, receiptId);
transportFilter.sendToStomp(sc);
}
@@ -160,10 +160,10 @@
e.printStackTrace(stream);
stream.close();
- HashMap headers = new HashMap();
+ HashMap<String, String> headers = new HashMap<String, String>();
headers.put(Stomp.Headers.Error.MESSAGE, e.getMessage());
- final String receiptId = (String)command.getHeaders().get(Stomp.Headers.RECEIPT_REQUESTED);
+ final String receiptId = command.getHeaders().get(Stomp.Headers.RECEIPT_REQUESTED);
if (receiptId != null) {
headers.put(Stomp.Headers.Response.RECEIPT_ID, receiptId);
}
@@ -180,8 +180,8 @@
protected void onStompSend(StompFrame command) throws IOException, JMSException {
checkConnected();
- Map headers = command.getHeaders();
- String stompTx = (String)headers.get(Stomp.Headers.TRANSACTION);
+ Map<String, String> headers = command.getHeaders();
+ String stompTx = headers.get(Stomp.Headers.TRANSACTION);
ActiveMQMessage message = convertMessage(command);
@@ -191,7 +191,7 @@
message.setJMSTimestamp(System.currentTimeMillis());
if (stompTx != null) {
- TransactionId activemqTx = (TransactionId)transactions.get(stompTx);
+ TransactionId activemqTx = transactions.get(stompTx);
if (activemqTx == null) {
throw new ProtocolException("Invalid transaction id: " + stompTx);
}
@@ -212,24 +212,24 @@
// on the same stomp connection. For example, when 2 subs are created on
// the same topic.
- Map headers = command.getHeaders();
- String messageId = (String)headers.get(Stomp.Headers.Ack.MESSAGE_ID);
+ Map<String, String> headers = command.getHeaders();
+ String messageId = headers.get(Stomp.Headers.Ack.MESSAGE_ID);
if (messageId == null) {
throw new ProtocolException("ACK received without a message-id to acknowledge!");
}
TransactionId activemqTx = null;
- String stompTx = (String)headers.get(Stomp.Headers.TRANSACTION);
+ String stompTx = headers.get(Stomp.Headers.TRANSACTION);
if (stompTx != null) {
- activemqTx = (TransactionId)transactions.get(stompTx);
+ activemqTx = transactions.get(stompTx);
if (activemqTx == null) {
throw new ProtocolException("Invalid transaction id: " + stompTx);
}
}
boolean acked = false;
- for (Iterator iter = subscriptionsByConsumerId.values().iterator(); iter.hasNext();) {
- StompSubscription sub = (StompSubscription)iter.next();
+ for (Iterator<StompSubscription> iter = subscriptionsByConsumerId.values().iterator(); iter.hasNext();) {
+ StompSubscription sub = iter.next();
MessageAck ack = sub.onStompMessageAck(messageId);
if (ack != null) {
ack.setTransactionId(activemqTx);
@@ -248,9 +248,9 @@
protected void onStompBegin(StompFrame command) throws ProtocolException {
checkConnected();
- Map headers = command.getHeaders();
+ Map<String, String> headers = command.getHeaders();
- String stompTx = (String)headers.get(Stomp.Headers.TRANSACTION);
+ String stompTx = headers.get(Stomp.Headers.TRANSACTION);
if (!headers.containsKey(Stomp.Headers.TRANSACTION)) {
throw new ProtocolException("Must specify the transaction you are beginning");
@@ -275,14 +275,14 @@
protected void onStompCommit(StompFrame command) throws ProtocolException {
checkConnected();
- Map headers = command.getHeaders();
+ Map<String, String> headers = command.getHeaders();
- String stompTx = (String)headers.get(Stomp.Headers.TRANSACTION);
+ String stompTx = headers.get(Stomp.Headers.TRANSACTION);
if (stompTx == null) {
throw new ProtocolException("Must specify the transaction you are committing");
}
- TransactionId activemqTx = (TransactionId)transactions.remove(stompTx);
+ TransactionId activemqTx = transactions.remove(stompTx);
if (activemqTx == null) {
throw new ProtocolException("Invalid transaction id: " + stompTx);
}
@@ -297,14 +297,14 @@
protected void onStompAbort(StompFrame command) throws ProtocolException {
checkConnected();
- Map headers = command.getHeaders();
+ Map<String, String> headers = command.getHeaders();
- String stompTx = (String)headers.get(Stomp.Headers.TRANSACTION);
+ String stompTx = headers.get(Stomp.Headers.TRANSACTION);
if (stompTx == null) {
throw new ProtocolException("Must specify the transaction you are committing");
}
- TransactionId activemqTx = (TransactionId)transactions.remove(stompTx);
+ TransactionId activemqTx = transactions.remove(stompTx);
if (activemqTx == null) {
throw new ProtocolException("Invalid transaction id: " + stompTx);
}
@@ -320,10 +320,10 @@
protected void onStompSubscribe(StompFrame command) throws ProtocolException {
checkConnected();
- Map headers = command.getHeaders();
+ Map<String, String> headers = command.getHeaders();
- String subscriptionId = (String)headers.get(Stomp.Headers.Subscribe.ID);
- String destination = (String)headers.get(Stomp.Headers.Subscribe.DESTINATION);
+ String subscriptionId = headers.get(Stomp.Headers.Subscribe.ID);
+ String destination = headers.get(Stomp.Headers.Subscribe.DESTINATION);
ActiveMQDestination actualDest = frameTranslator.convertDestination(destination);
ConsumerId id = new ConsumerId(sessionId, consumerIdGenerator.getNextSequenceId());
@@ -331,7 +331,7 @@
consumerInfo.setPrefetchSize(1000);
consumerInfo.setDispatchAsync(true);
- String selector = (String)headers.remove(Stomp.Headers.Subscribe.SELECTOR);
+ String selector = headers.remove(Stomp.Headers.Subscribe.SELECTOR);
consumerInfo.setSelector(selector);
IntrospectionSupport.setProperties(consumerInfo, headers, "activemq.");
@@ -341,7 +341,7 @@
StompSubscription stompSubscription = new StompSubscription(this, subscriptionId, consumerInfo);
stompSubscription.setDestination(actualDest);
- String ackMode = (String)headers.get(Stomp.Headers.Subscribe.ACK_MODE);
+ String ackMode = headers.get(Stomp.Headers.Subscribe.ACK_MODE);
if (Stomp.Headers.Subscribe.AckModeValues.CLIENT.equals(ackMode)) {
stompSubscription.setAckMode(StompSubscription.CLIENT_ACK);
} else {
@@ -355,7 +355,7 @@
protected void onStompUnsubscribe(StompFrame command) throws ProtocolException {
checkConnected();
- Map headers = command.getHeaders();
+ Map<String, String> headers = command.getHeaders();
ActiveMQDestination destination = null;
Object o = headers.get(Stomp.Headers.Unsubscribe.DESTINATION);
@@ -363,7 +363,7 @@
destination = frameTranslator.convertDestination((String)o);
}
- String subscriptionId = (String)headers.get(Stomp.Headers.Unsubscribe.ID);
+ String subscriptionId = headers.get(Stomp.Headers.Unsubscribe.ID);
if (subscriptionId == null && destination == null) {
throw new ProtocolException("Must specify the subscriptionId or the destination you are unsubscribing from");
@@ -374,8 +374,8 @@
// are created with the same destination. Perhaps this should be
// removed.
//
- for (Iterator iter = subscriptionsByConsumerId.values().iterator(); iter.hasNext();) {
- StompSubscription sub = (StompSubscription)iter.next();
+ for (Iterator<StompSubscription> iter = subscriptionsByConsumerId.values().iterator(); iter.hasNext();) {
+ StompSubscription sub = iter.next();
if ((subscriptionId != null && subscriptionId.equals(sub.getSubscriptionId())) || (destination != null && destination.equals(sub.getDestination()))) {
sendToActiveMQ(sub.getConsumerInfo().createRemoveCommand(), createResponseHandler(command));
iter.remove();
@@ -392,12 +392,12 @@
throw new ProtocolException("Allready connected.");
}
- final Map headers = command.getHeaders();
+ final Map<String, String> headers = command.getHeaders();
// allow anyone to login for now
- String login = (String)headers.get(Stomp.Headers.Connect.LOGIN);
- String passcode = (String)headers.get(Stomp.Headers.Connect.PASSCODE);
- String clientId = (String)headers.get(Stomp.Headers.Connect.CLIENT_ID);
+ String login = headers.get(Stomp.Headers.Connect.LOGIN);
+ String passcode = headers.get(Stomp.Headers.Connect.PASSCODE);
+ String clientId = headers.get(Stomp.Headers.Connect.CLIENT_ID);
final ConnectionInfo connectionInfo = new ConnectionInfo();
@@ -425,13 +425,13 @@
public void onResponse(ProtocolConverter converter, Response response) throws IOException {
connected.set(true);
- HashMap responseHeaders = new HashMap();
+ HashMap<String, String> responseHeaders = new HashMap<String, String>();
responseHeaders.put(Stomp.Headers.Connected.SESSION, connectionInfo.getClientId());
- String requestId = (String)headers.get(Stomp.Headers.Connect.REQUEST_ID);
+ String requestId = headers.get(Stomp.Headers.Connect.REQUEST_ID);
if (requestId == null) {
// TODO legacy
- requestId = (String)headers.get(Stomp.Headers.RECEIPT_REQUESTED);
+ requestId = headers.get(Stomp.Headers.RECEIPT_REQUESTED);
}
if (requestId != null) {
// TODO legacy
@@ -473,7 +473,7 @@
if (command.isResponse()) {
Response response = (Response)command;
- ResponseHandler rh = (ResponseHandler)resposeHandlers.remove(Integer.valueOf(response.getCorrelationId()));
+ ResponseHandler rh = resposeHandlers.remove(Integer.valueOf(response.getCorrelationId()));
if (rh != null) {
rh.onResponse(this, response);
}
@@ -481,7 +481,7 @@
} else if (command.isMessageDispatch()) {
MessageDispatch md = (MessageDispatch)command;
- StompSubscription sub = (StompSubscription)subscriptionsByConsumerId.get(md.getConsumerId());
+ StompSubscription sub = subscriptionsByConsumerId.get(md.getConsumerId());
if (sub != null) {
sub.onMessageDispatch(md);
}
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompFrame.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompFrame.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompFrame.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/stomp/StompFrame.java Fri Aug 10 09:57:01 2007
@@ -37,10 +37,10 @@
private static final byte[] NO_DATA = new byte[] {};
private String action;
- private Map headers = Collections.EMPTY_MAP;
+ private Map<String,String> headers = new HashMap<String,String>();
private byte[] content = NO_DATA;
- public StompFrame(String command, HashMap headers, byte[] data) {
+ public StompFrame(String command, Map<String,String> headers, byte[] data) {
this.action = command;
this.headers = headers;
this.content = data;
@@ -65,11 +65,11 @@
this.content = data;
}
- public Map getHeaders() {
+ public Map<String,String> getHeaders() {
return headers;
}
- public void setHeaders(Map headers) {
+ public void setHeaders(Map<String,String> headers) {
this.headers = headers;
}
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/tcp/SslTransportFactory.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/tcp/SslTransportFactory.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/tcp/SslTransportFactory.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/tcp/SslTransportFactory.java Fri Aug 10 09:57:01 2007
@@ -57,11 +57,12 @@
* @version $Revision: $
*/
public class SslTransportFactory extends TcpTransportFactory {
- // The context used to creat ssl sockets.
- private SSLContext sslContext = null;
-
// The log this uses.,
private static final Log LOG = LogFactory.getLog(SslTransportFactory.class);
+
+ // The context used to creat ssl sockets.
+ private SSLContext sslContext;
+
/**
* Constructor. Nothing special.
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/tcp/TcpBufferedInputStream.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/tcp/TcpBufferedInputStream.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/tcp/TcpBufferedInputStream.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/tcp/TcpBufferedInputStream.java Fri Aug 10 09:57:01 2007
@@ -45,7 +45,8 @@
private void fill() throws IOException {
byte[] buffer = internalBuffer;
- count = position = 0;
+ count = 0;
+ position = 0;
int n = in.read(buffer, position, buffer.length - position);
if (n > 0) {
count = n + position;
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/ByteArrayInputStream.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/ByteArrayInputStream.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/ByteArrayInputStream.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/ByteArrayInputStream.java Fri Aug 10 09:57:01 2007
@@ -40,7 +40,8 @@
public ByteArrayInputStream(byte data[], int offset, int size) {
this.buffer = data;
- this.mark = this.pos = offset;
+ this.mark = offset;
+ this.pos = offset;
this.limit = offset + size;
}
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/ByteSequenceData.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/ByteSequenceData.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/ByteSequenceData.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/ByteSequenceData.java Fri Aug 10 09:57:01 2007
@@ -88,12 +88,12 @@
public static int readUnsignedShortBig(ByteSequence packet) throws IOException {
spaceNeeded(packet, 2);
- return ((read(packet) << 8) + (read(packet) << 0));
+ return (read(packet) << 8) + (read(packet) << 0);
}
public static int readUnsignedShortLittle(ByteSequence packet) throws IOException {
spaceNeeded(packet, 2);
- return ((read(packet) << 0) + (read(packet) << 8));
+ return (read(packet) << 0) + (read(packet) << 8);
}
public static char readCharBig(ByteSequence packet) throws IOException {
@@ -108,24 +108,24 @@
public static int readIntBig(ByteSequence packet) throws IOException {
spaceNeeded(packet, 4);
- return ((read(packet) << 24) + (read(packet) << 16) + (read(packet) << 8) + (read(packet) << 0));
+ return (read(packet) << 24) + (read(packet) << 16) + (read(packet) << 8) + (read(packet) << 0);
}
public static int readIntLittle(ByteSequence packet) throws IOException {
spaceNeeded(packet, 4);
- return ((read(packet) << 0) + (read(packet) << 8) + (read(packet) << 16) + (read(packet) << 24));
+ return (read(packet) << 0) + (read(packet) << 8) + (read(packet) << 16) + (read(packet) << 24);
}
public static long readLongBig(ByteSequence packet) throws IOException {
spaceNeeded(packet, 8);
- return (((long)read(packet) << 56) + ((long)read(packet) << 48) + ((long)read(packet) << 40) + ((long)read(packet) << 32) + ((long)read(packet) << 24)
- + ((read(packet)) << 16) + ((read(packet)) << 8) + ((read(packet)) << 0));
+ return ((long)read(packet) << 56) + ((long)read(packet) << 48) + ((long)read(packet) << 40) + ((long)read(packet) << 32) + ((long)read(packet) << 24)
+ + ((read(packet)) << 16) + ((read(packet)) << 8) + ((read(packet)) << 0);
}
public static long readLongLittle(ByteSequence packet) throws IOException {
spaceNeeded(packet, 8);
- return ((read(packet) << 0) + (read(packet) << 8) + (read(packet) << 16) + ((long)read(packet) << 24) + ((long)read(packet) << 32) + ((long)read(packet) << 40)
- + ((long)read(packet) << 48) + ((long)read(packet) << 56));
+ return (read(packet) << 0) + (read(packet) << 8) + (read(packet) << 16) + ((long)read(packet) << 24) + ((long)read(packet) << 32) + ((long)read(packet) << 40)
+ + ((long)read(packet) << 48) + ((long)read(packet) << 56);
}
public static double readDoubleBig(ByteSequence packet) throws IOException {
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/ClassLoading.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/ClassLoading.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/ClassLoading.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/ClassLoading.java Fri Aug 10 09:57:01 2007
@@ -25,9 +25,29 @@
*
* @version $Rev: 109957 $ $Date$
*/
-public class ClassLoading {
+public final class ClassLoading {
/**
+ * Primitive type name -> class map.
+ */
+ private static final Map<String, Class> PRIMITIVES = new HashMap<String, Class>();
+ /**
+ * VM primitive type primitive type -> name
+ */
+ private static final Map<Class, String> VM_PRIMITIVES_REVERSE = new HashMap<Class, String>();
+ /**
+ * VM primitive type name -> primitive type
+ */
+ private static final Map<String, Class> VM_PRIMITIVES = new HashMap<String, Class>();
+ /**
+ * Map of primitive types to their wrapper classes
+ */
+ private static final Map<Class, Class> PRIMITIVE_WRAPPERS = new HashMap<Class, Class>();
+
+ private ClassLoading() {
+ }
+
+ /**
* Load a class for the given name. <p/>
* <p>
* Handles loading primitive types as well as VM class and array syntax.
@@ -130,11 +150,6 @@
return rc.toString();
}
- /**
- * Primitive type name -> class map.
- */
- private static final Map PRIMITIVES = new HashMap();
-
/** Setup the primitives map. */
static {
PRIMITIVES.put("boolean", Boolean.TYPE);
@@ -156,13 +171,9 @@
* @return Primitive type or null.
*/
private static Class getPrimitiveType(final String name) {
- return (Class) PRIMITIVES.get(name);
+ return PRIMITIVES.get(name);
}
- /**
- * VM primitive type name -> primitive type
- */
- private static final HashMap VM_PRIMITIVES = new HashMap();
/** Setup the vm primitives map. */
static {
@@ -177,10 +188,6 @@
VM_PRIMITIVES.put("V", void.class);
}
- /**
- * VM primitive type primitive type -> name
- */
- private static final HashMap VM_PRIMITIVES_REVERSE = new HashMap();
/** Setup the vm primitives reverse map. */
static {
@@ -219,13 +226,8 @@
* @return Primitive type or null.
*/
private static Class getVMPrimitiveType(final String name) {
- return (Class) VM_PRIMITIVES.get(name);
+ return VM_PRIMITIVES.get(name);
}
-
- /**
- * Map of primitive types to their wrapper classes
- */
- private static final Map PRIMITIVE_WRAPPERS = new HashMap();
/** Setup the wrapper map. */
static {
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/DataByteArrayInputStream.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/DataByteArrayInputStream.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/DataByteArrayInputStream.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/DataByteArrayInputStream.java Fri Aug 10 09:57:01 2007
@@ -49,7 +49,8 @@
*/
public DataByteArrayInputStream(ByteSequence sequence) {
this.buf = sequence.getData();
- this.offset = this.pos = sequence.getOffset();
+ this.offset = sequence.getOffset();
+ this.pos = this.offset;
}
/**
@@ -200,7 +201,7 @@
public int readUnsignedShort() {
int ch1 = read();
int ch2 = read();
- return ((ch1 << 8) + (ch2 << 0));
+ return (ch1 << 8) + (ch2 << 0);
}
public char readChar() {
@@ -214,7 +215,7 @@
int ch2 = read();
int ch3 = read();
int ch4 = read();
- return ((ch1 << 24) + (ch2 << 16) + (ch3 << 8) + (ch4 << 0));
+ return (ch1 << 24) + (ch2 << 16) + (ch3 << 8) + (ch4 << 0);
}
public long readLong() {
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/IntrospectionSupport.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/IntrospectionSupport.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/IntrospectionSupport.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/IntrospectionSupport.java Fri Aug 10 09:57:01 2007
@@ -83,7 +83,7 @@
return rc;
}
- public static boolean setProperties(Object target, Map props, String optionPrefix) {
+ public static boolean setProperties(Object target, Map<String, ?> props, String optionPrefix) {
boolean rc = false;
if (target == null) {
throw new IllegalArgumentException("target was null.");
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/MarshallingSupport.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/MarshallingSupport.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/MarshallingSupport.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/MarshallingSupport.java Fri Aug 10 09:57:01 2007
@@ -35,7 +35,7 @@
*
* @version $Revision$
*/
-public class MarshallingSupport {
+public final class MarshallingSupport {
public static final byte NULL = 0;
public static final byte BOOLEAN_TYPE = 1;
@@ -52,6 +52,9 @@
public static final byte LIST_TYPE = 12;
public static final byte BIG_STRING_TYPE = 13;
+ private MarshallingSupport() {
+ }
+
public static void marshalPrimitiveMap(Map map, DataOutputStream out) throws IOException {
if (map == null) {
out.writeInt(-1);
@@ -66,7 +69,7 @@
}
}
- public static Map unmarshalPrimitiveMap(DataInputStream in) throws IOException {
+ public static Map<String, Object> unmarshalPrimitiveMap(DataInputStream in) throws IOException {
return unmarshalPrimitiveMap(in, Integer.MAX_VALUE);
}
@@ -76,7 +79,7 @@
* @throws IOException
* @throws IOException
*/
- public static Map unmarshalPrimitiveMap(DataInputStream in, int maxPropertySize) throws IOException {
+ public static Map<String, Object> unmarshalPrimitiveMap(DataInputStream in, int maxPropertySize) throws IOException {
int size = in.readInt();
if (size > maxPropertySize) {
throw new IOException("Primitive map is larger than the allowed size: " + size);
@@ -84,7 +87,7 @@
if (size < 0) {
return null;
} else {
- HashMap rc = new HashMap(size);
+ Map<String, Object> rc = new HashMap<String, Object>(size);
for (int i = 0; i < size; i++) {
String name = in.readUTF();
rc.put(name, unmarshalPrimitive(in));
@@ -102,9 +105,9 @@
}
}
- public static List unmarshalPrimitiveList(DataInputStream in) throws IOException {
+ public static List<Object> unmarshalPrimitiveList(DataInputStream in) throws IOException {
int size = in.readInt();
- List answer = new ArrayList(size);
+ List<Object> answer = new ArrayList<Object>(size);
while (size-- > 0) {
answer.add(unmarshalPrimitive(in));
}
@@ -150,7 +153,7 @@
byte type = in.readByte();
switch (type) {
case BYTE_TYPE:
- value = Byte.valueOf(type);
+ value = Byte.valueOf(in.readByte());
break;
case BOOLEAN_TYPE:
value = in.readBoolean() ? Boolean.TRUE : Boolean.FALSE;
@@ -265,7 +268,8 @@
int strlen = text.length();
int utflen = 0;
char[] charr = new char[strlen];
- int c, count = 0;
+ int c = 0;
+ int count = 0;
text.getChars(0, strlen, charr, 0);
@@ -312,7 +316,9 @@
if (utflen > -1) {
StringBuffer str = new StringBuffer(utflen);
byte bytearr[] = new byte[utflen];
- int c, char2, char3;
+ int c;
+ int char2;
+ int char3;
int count = 0;
dataIn.readFully(bytearr, 0, utflen);
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/MemoryIntPropertyEditor.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/MemoryIntPropertyEditor.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/MemoryIntPropertyEditor.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/MemoryIntPropertyEditor.java Fri Aug 10 09:57:01 2007
@@ -60,7 +60,7 @@
public String getAsText() {
Integer value = (Integer)getValue();
- return (value != null ? value.toString() : "");
+ return value != null ? value.toString() : "";
}
}
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/MemoryPropertyEditor.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/MemoryPropertyEditor.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/MemoryPropertyEditor.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/MemoryPropertyEditor.java Fri Aug 10 09:57:01 2007
@@ -60,7 +60,7 @@
public String getAsText() {
Long value = (Long)getValue();
- return (value != null ? value.toString() : "");
+ return value != null ? value.toString() : "";
}
}
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/URISupport.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/URISupport.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/URISupport.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/util/URISupport.java Fri Aug 10 09:57:01 2007
@@ -25,6 +25,7 @@
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
+import java.util.List;
import java.util.Map;
/**
@@ -33,12 +34,12 @@
public class URISupport {
public static class CompositeData {
- String scheme;
- String path;
- URI components[];
- Map parameters;
- String fragment;
- public String host;
+ private String host;
+ private String scheme;
+ private String path;
+ private URI components[];
+ private Map<String, String> parameters;
+ private String fragment;
public URI[] getComponents() {
return components;
@@ -48,7 +49,7 @@
return fragment;
}
- public Map getParameters() {
+ public Map<String, String> getParameters() {
return parameters;
}
@@ -100,9 +101,9 @@
}
}
- public static Map parseQuery(String uri) throws URISyntaxException {
+ public static Map<String, String> parseQuery(String uri) throws URISyntaxException {
try {
- Map rc = new HashMap();
+ Map<String, String> rc = new HashMap<String, String>();
if (uri != null) {
String[] parameters = uri.split("&");
for (int i = 0; i < parameters.length; i++) {
@@ -122,7 +123,7 @@
}
}
- public static Map parseParamters(URI uri) throws URISyntaxException {
+ public static Map<String,String> parseParamters(URI uri) throws URISyntaxException {
return uri.getQuery() == null ? Collections.EMPTY_MAP : parseQuery(stripPrefix(uri.getQuery(), "?"));
}
@@ -210,7 +211,7 @@
* @return
*/
private static String[] splitComponents(String str) {
- ArrayList l = new ArrayList();
+ List<String> l = new ArrayList<String>();
int last = 0;
int depth = 0;
Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/AutoFailTestSupport.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/AutoFailTestSupport.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/AutoFailTestSupport.java (original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/AutoFailTestSupport.java Fri Aug 10 09:57:01 2007
@@ -30,10 +30,9 @@
*/
public abstract class AutoFailTestSupport extends TestCase {
- private static final Log LOG = LogFactory.getLog(AutoFailTestSupport.class);
-
public static final int EXIT_SUCCESS = 0;
public static final int EXIT_ERROR = 1;
+ private static final Log LOG = LogFactory.getLog(AutoFailTestSupport.class);
private long maxTestTime = 5 * 60 * 1000; // 5 mins by default
private Thread autoFailThread;
Modified: activemq/trunk/activemq-core/src/test/java/org/apache/activemq/JMSConsumerTest.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/JMSConsumerTest.java?view=diff&rev=564679&r1=564678&r2=564679
==============================================================================
--- activemq/trunk/activemq-core/src/test/java/org/apache/activemq/JMSConsumerTest.java (original)
+++ activemq/trunk/activemq-core/src/test/java/org/apache/activemq/JMSConsumerTest.java Fri Aug 10 09:57:01 2007
@@ -41,8 +41,14 @@
*/
public class JMSConsumerTest extends JmsTestSupport {
- private static final org.apache.commons.logging.Log LOG = org.apache.commons.logging.LogFactory
- .getLog(JMSConsumerTest.class);
+ private static final org.apache.commons.logging.Log LOG = org.apache.commons.logging.LogFactory.getLog(JMSConsumerTest.class);
+
+ public ActiveMQDestination destination;
+ public int deliveryMode;
+ public int prefetch;
+ public int ackMode;
+ public byte destinationType;
+ public boolean durableConsumer;
public static Test suite() {
return suite(JMSConsumerTest.class);
@@ -52,21 +58,10 @@
junit.textui.TestRunner.run(suite());
}
- public ActiveMQDestination destination;
- public int deliveryMode;
- public int prefetch;
- public int ackMode;
- public byte destinationType;
- public boolean durableConsumer;
-
public void initCombosForTestMessageListenerWithConsumerCanBeStopped() {
- 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)});
+ 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 testMessageListenerWithConsumerCanBeStopped() throws Exception {
@@ -113,16 +108,11 @@
}
public void initCombosForTestMutiReceiveWithPrefetch1() {
- 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),
+ 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),
- Byte.valueOf(ActiveMQDestination.TOPIC_TYPE),
- Byte.valueOf(ActiveMQDestination.TEMP_QUEUE_TYPE),
- Byte.valueOf(ActiveMQDestination.TEMP_TOPIC_TYPE)});
+ 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 testMutiReceiveWithPrefetch1() throws Exception {
@@ -150,8 +140,7 @@
}
public void initCombosForTestDurableConsumerSelectorChange() {
- addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT),
- Integer.valueOf(DeliveryMode.PERSISTENT)});
+ addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
addCombinationValues("destinationType", new Object[] {Byte.valueOf(ActiveMQDestination.TOPIC_TYPE)});
}
@@ -164,8 +153,7 @@
destination = createDestination(session, destinationType);
MessageProducer producer = session.createProducer(destination);
producer.setDeliveryMode(deliveryMode);
- MessageConsumer consumer = session.createDurableSubscriber((Topic)destination, "test", "color='red'",
- false);
+ MessageConsumer consumer = session.createDurableSubscriber((Topic)destination, "test", "color='red'", false);
// Send the messages
TextMessage message = session.createTextMessage("1st");
@@ -196,13 +184,9 @@
}
public void initCombosForTestSendReceiveBytesMessage() {
- 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)});
+ 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 {
@@ -229,13 +213,9 @@
}
public void initCombosForTestSetMessageListenerAfterStart() {
- 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)});
+ 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 testSetMessageListenerAfterStart() throws Exception {
@@ -270,10 +250,8 @@
}
public void initCombosForTestMessageListenerUnackedWithPrefetch1StayInQueue() {
- 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),
+ 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)});
}
@@ -359,13 +337,9 @@
}
public void initCombosForTestMessageListenerWithConsumerWithPrefetch1() {
- 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)});
+ 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 testMessageListenerWithConsumerWithPrefetch1() throws Exception {
@@ -400,13 +374,9 @@
}
public void initCombosForTestMessageListenerWithConsumer() {
- 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)});
+ 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 testMessageListenerWithConsumer() throws Exception {
@@ -439,10 +409,8 @@
}
public void initCombosForTestUnackedWithPrefetch1StayInQueue() {
- 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),
+ 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)});
}
@@ -490,8 +458,7 @@
}
public void initCombosForTestPrefetch1MessageNotDispatched() {
- addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT),
- Integer.valueOf(DeliveryMode.PERSISTENT)});
+ addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT), Integer.valueOf(DeliveryMode.PERSISTENT)});
}
public void testPrefetch1MessageNotDispatched() throws Exception {
@@ -535,8 +502,7 @@
public void initCombosForTestDontStart() {
addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT)});
- addCombinationValues("destinationType", new Object[] {Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
- Byte.valueOf(ActiveMQDestination.TOPIC_TYPE)});
+ addCombinationValues("destinationType", new Object[] {Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TOPIC_TYPE)});
}
public void testDontStart() throws Exception {
@@ -554,8 +520,7 @@
public void initCombosForTestStartAfterSend() {
addCombinationValues("deliveryMode", new Object[] {Integer.valueOf(DeliveryMode.NON_PERSISTENT)});
- addCombinationValues("destinationType", new Object[] {Byte.valueOf(ActiveMQDestination.QUEUE_TYPE),
- Byte.valueOf(ActiveMQDestination.TOPIC_TYPE)});
+ addCombinationValues("destinationType", new Object[] {Byte.valueOf(ActiveMQDestination.QUEUE_TYPE), Byte.valueOf(ActiveMQDestination.TOPIC_TYPE)});
}
public void testStartAfterSend() throws Exception {
@@ -576,13 +541,9 @@
}
public void initCombosForTestReceiveMessageWithConsumer() {
- 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)});
+ 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 {