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

More checkstyle fixes 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@564057 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Hiram R. Chirino 2007-08-09 00:34:07 +00:00
parent f812e34179
commit 74a7a8bbfc
724 changed files with 3525 additions and 4208 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

30
activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
View File

@ -41,7 +41,6 @@ import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.ExceptionListener;
import javax.jms.IllegalStateException;
import javax.jms.InvalidDestinationException;
import javax.jms.JMSException;
import javax.jms.Queue;
import javax.jms.QueueConnection;
@ -290,8 +289,8 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon
checkClosedOrFailed();
ensureConnectionInfoSent();
boolean doSessionAsync = alwaysSessionAsync || sessions.size() > 0 || transacted || acknowledgeMode == Session.CLIENT_ACKNOWLEDGE;
return new ActiveMQSession(this, getNextSessionId(), (transacted ? Session.SESSION_TRANSACTED : (acknowledgeMode == Session.SESSION_TRANSACTED
? Session.AUTO_ACKNOWLEDGE : acknowledgeMode)), dispatchAsync, alwaysSessionAsync);
return new ActiveMQSession(this, getNextSessionId(), transacted ? Session.SESSION_TRANSACTED : (acknowledgeMode == Session.SESSION_TRANSACTED
? Session.AUTO_ACKNOWLEDGE : acknowledgeMode), dispatchAsync, alwaysSessionAsync);
}
/**
@ -1168,10 +1167,11 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon
Response response = (Response)this.transport.request(command);
if (response.isException()) {
ExceptionResponse er = (ExceptionResponse)response;
if (er.getException() instanceof JMSException)
if (er.getException() instanceof JMSException) {
throw (JMSException)er.getException();
else
} else {
throw JMSExceptionSupport.create(er.getException());
}
}
return response;
} catch (IOException e) {
@ -1196,10 +1196,11 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon
Response response = (Response)this.transport.request(command, timeout);
if (response != null && response.isException()) {
ExceptionResponse er = (ExceptionResponse)response;
if (er.getException() instanceof JMSException)
if (er.getException() instanceof JMSException) {
throw (JMSException)er.getException();
else
} else {
throw JMSExceptionSupport.create(er.getException());
}
}
return response;
} catch (IOException e) {
@ -1361,9 +1362,9 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon
* @throws IllegalStateException if the connection is in used.
*/
public void changeUserInfo(String userName, String password) throws JMSException {
if (isConnectionInfoSentToBroker)
if (isConnectionInfoSentToBroker) {
throw new IllegalStateException("changeUserInfo used Connection is not allowed");
}
this.info.setUserName(userName);
this.info.setPassword(password);
}
@ -1374,8 +1375,9 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon
*/
public String getResourceManagerId() throws JMSException {
waitForBrokerInfo();
if (brokerInfo == null)
if (brokerInfo == null) {
throw new JMSException("Connection failed before Broker info was received.");
}
return brokerInfo.getBrokerId().getValue();
}
@ -1580,8 +1582,6 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon
onAsyncException(error.getException());
}
});
new Thread("Async error worker") {
}.start();
return null;
}
@ -1633,8 +1633,9 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon
if (!closed.get() && !closing.get()) {
if (this.exceptionListener != null) {
if (!(error instanceof JMSException))
if (!(error instanceof JMSException)) {
error = JMSExceptionSupport.create(error);
}
final JMSException e = (JMSException)error;
asyncConnectionThread.execute(new Runnable() {
@ -1744,8 +1745,9 @@ public class ActiveMQConnection implements Connection, TopicConnection, QueueCon
// If we are not watching the advisories.. then
// we will assume that the temp destination does exist.
if (advisoryConsumer == null)
if (advisoryConsumer == null) {
return false;
}
return !activeTempDestinations.contains(dest);
}

33
activemq-core/src/main/java/org/apache/activemq/ActiveMQConnectionFactory.java
View File

@ -16,6 +16,23 @@
*/
package org.apache.activemq;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.naming.Context;
import org.apache.activemq.blob.BlobTransferPolicy;
import org.apache.activemq.jndi.JNDIBaseStorable;
import org.apache.activemq.management.JMSStatsImpl;
@ -29,22 +46,6 @@ import org.apache.activemq.util.JMSExceptionSupport;
import org.apache.activemq.util.URISupport;
import org.apache.activemq.util.URISupport.CompositeData;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.JMSException;
import javax.jms.QueueConnection;
import javax.jms.QueueConnectionFactory;
import javax.jms.TopicConnection;
import javax.jms.TopicConnectionFactory;
import javax.naming.Context;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.Executor;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.ThreadFactory;
/**
* A ConnectionFactory is an an Administered object, and is used for creating
* Connections. <p/> This class also implements QueueConnectionFactory and

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

@ -28,7 +28,7 @@ import javax.jms.ConnectionMetaData;
* the <CODE>Connection</CODE> object.
*/
public class ActiveMQConnectionMetaData implements ConnectionMetaData {
public final class ActiveMQConnectionMetaData implements ConnectionMetaData {
public static final String PROVIDER_VERSION;
public static final int PROVIDER_MAJOR_VERSION;

40
activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageConsumer.java
View File

@ -16,7 +16,29 @@
*/
package org.apache.activemq;
import org.apache.activemq.command.*;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.jms.IllegalStateException;
import javax.jms.InvalidDestinationException;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQMessage;
import org.apache.activemq.command.ActiveMQTempDestination;
import org.apache.activemq.command.ConsumerId;
import org.apache.activemq.command.ConsumerInfo;
import org.apache.activemq.command.MessageAck;
import org.apache.activemq.command.MessageDispatch;
import org.apache.activemq.command.MessagePull;
import org.apache.activemq.management.JMSConsumerStatsImpl;
import org.apache.activemq.management.StatsCapable;
import org.apache.activemq.management.StatsImpl;
@ -29,18 +51,6 @@ import org.apache.activemq.util.JMSExceptionSupport;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import javax.jms.IllegalStateException;
import javax.jms.*;
import javax.jms.Message;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* A client uses a <CODE>MessageConsumer</CODE> object to receive messages
* from a destination. A <CODE> MessageConsumer</CODE> object is created by
@ -612,7 +622,7 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
Thread.currentThread().interrupt();
}
}
if ((session.isTransacted() || session.isDupsOkAcknowledge())) {
if (session.isTransacted() || session.isDupsOkAcknowledge()) {
acknowledge();
}
if (session.isClientAcknowledge()) {
@ -885,7 +895,7 @@ public class ActiveMQMessageConsumer implements MessageAvailableConsumer, StatsC
}
}
if (!unconsumedMessages.isClosed()) {
if (this.info.isBrowser() || session.connection.isDuplicate(this, md.getMessage()) == false) {
if (this.info.isBrowser() || !session.connection.isDuplicate(this, md.getMessage())) {
if (listener != null && unconsumedMessages.isRunning()) {
ActiveMQMessage message = createActiveMQMessage(md);
beforeMessageIsConsumed(md);

21
activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageProducer.java
View File

@ -16,6 +16,15 @@
*/
package org.apache.activemq;
import java.util.HashMap;
import java.util.concurrent.atomic.AtomicLong;
import javax.jms.Destination;
import javax.jms.IllegalStateException;
import javax.jms.InvalidDestinationException;
import javax.jms.JMSException;
import javax.jms.Message;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ProducerAck;
import org.apache.activemq.command.ProducerId;
@ -26,14 +35,6 @@ import org.apache.activemq.management.StatsImpl;
import org.apache.activemq.memory.UsageManager;
import org.apache.activemq.util.IntrospectionSupport;
import javax.jms.Destination;
import javax.jms.IllegalStateException;
import javax.jms.InvalidDestinationException;
import javax.jms.JMSException;
import javax.jms.Message;
import java.util.HashMap;
import java.util.concurrent.atomic.AtomicLong;
/**
* A client uses a <CODE>MessageProducer</CODE> object to send messages to a
* destination. A <CODE>MessageProducer</CODE> object is created by passing a
@ -139,14 +140,14 @@ public class ActiveMQMessageProducer extends ActiveMQMessageProducerSupport impl
* to some internal error.
*/
public void close() throws JMSException {
if (closed == false) {
if (!closed) {
dispose();
this.session.asyncSendPacket(info.createRemoveCommand());
}
}
public void dispose() {
if (closed == false) {
if (!closed) {
this.session.removeProducer(this);
closed = true;
}

7
activemq-core/src/main/java/org/apache/activemq/ActiveMQMessageProducerSupport.java
View File

@ -16,9 +16,12 @@
*/
package org.apache.activemq;
import javax.jms.*;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.IllegalStateException;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageProducer;
/**
* A useful base class for implementing a {@link MessageProducer}
@ -194,7 +197,7 @@ public abstract class ActiveMQMessageProducerSupport implements MessageProducer,
* @see javax.jms.Message#DEFAULT_TIME_TO_LIVE
*/
public void setTimeToLive(long timeToLive) throws JMSException {
if (timeToLive < 0l) {
if (timeToLive < 0L) {
throw new IllegalStateException("cannot set a negative timeToLive");
}
checkClosed();

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

@ -20,10 +20,10 @@ import java.util.Enumeration;
import javax.jms.BytesMessage;
import javax.jms.Destination;
import javax.jms.MessageEOFException;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
import javax.jms.MessageEOFException;
import javax.jms.ObjectMessage;
import javax.jms.Queue;
import javax.jms.StreamMessage;

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

@ -72,7 +72,7 @@ public class ActiveMQOutputStream extends OutputStream implements Disposable {
}
public void close() throws IOException {
if (closed == false) {
if (!closed) {
flushBuffer();
try {
// Send an EOS style empty message to signal EOS.
@ -86,7 +86,7 @@ public class ActiveMQOutputStream extends OutputStream implements Disposable {
}
public void dispose() {
if (closed == false) {
if (!closed) {
this.connection.removeOutputStream(this);
closed = true;
}

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

@ -17,6 +17,7 @@
package org.apache.activemq;
import java.util.Enumeration;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.jms.IllegalStateException;
import javax.jms.JMSException;
@ -28,8 +29,6 @@ import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ConsumerId;
import org.apache.activemq.command.MessageDispatch;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* A client uses a <CODE>QueueBrowser</CODE> object to look at messages on a
* queue without removing them. <p/>

27
activemq-core/src/main/java/org/apache/activemq/ActiveMQSession.java
View File

@ -29,8 +29,6 @@ import java.util.concurrent.atomic.AtomicBoolean;
import javax.jms.BytesMessage;
import javax.jms.Destination;
import javax.jms.IllegalStateException;
import javax.jms.InvalidDestinationException;
import javax.jms.InvalidSelectorException;
import javax.jms.JMSException;
import javax.jms.MapMessage;
import javax.jms.Message;
@ -52,11 +50,32 @@ import javax.jms.Topic;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;
import javax.jms.TopicSubscriber;
import javax.jms.TransactionRolledBackException;
import org.apache.activemq.blob.BlobTransferPolicy;
import org.apache.activemq.blob.BlobUploader;
import org.apache.activemq.command.*;
import org.apache.activemq.command.ActiveMQBlobMessage;
import org.apache.activemq.command.ActiveMQBytesMessage;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQMapMessage;
import org.apache.activemq.command.ActiveMQMessage;
import org.apache.activemq.command.ActiveMQObjectMessage;
import org.apache.activemq.command.ActiveMQQueue;
import org.apache.activemq.command.ActiveMQStreamMessage;
import org.apache.activemq.command.ActiveMQTempDestination;
import org.apache.activemq.command.ActiveMQTempQueue;
import org.apache.activemq.command.ActiveMQTempTopic;
import org.apache.activemq.command.ActiveMQTextMessage;
import org.apache.activemq.command.ActiveMQTopic;
import org.apache.activemq.command.Command;
import org.apache.activemq.command.ConsumerId;
import org.apache.activemq.command.MessageAck;
import org.apache.activemq.command.MessageDispatch;
import org.apache.activemq.command.MessageId;
import org.apache.activemq.command.ProducerId;
import org.apache.activemq.command.Response;
import org.apache.activemq.command.SessionId;
import org.apache.activemq.command.SessionInfo;
import org.apache.activemq.command.TransactionId;
import org.apache.activemq.management.JMSSessionStatsImpl;
import org.apache.activemq.management.StatsCapable;
import org.apache.activemq.management.StatsImpl;

9
activemq-core/src/main/java/org/apache/activemq/ActiveMQTopicPublisher.java
View File

@ -17,17 +17,12 @@
package org.apache.activemq;
import org.apache.activemq.command.ActiveMQDestination;
import javax.jms.Destination;
import javax.jms.InvalidDestinationException;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageFormatException;
import javax.jms.Session;
import javax.jms.Topic;
import javax.jms.TopicPublisher;
import javax.jms.TopicSession;
import org.apache.activemq.command.ActiveMQDestination;
/**
* A client uses a <CODE>TopicPublisher</CODE> object to publish messages on

6
activemq-core/src/main/java/org/apache/activemq/ActiveMQTopicSubscriber.java
View File

@ -17,13 +17,13 @@
package org.apache.activemq;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ConsumerId;
import javax.jms.JMSException;
import javax.jms.Topic;
import javax.jms.TopicSubscriber;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ConsumerId;
/**
* A client uses a <CODE>TopicSubscriber</CODE> object to receive messages
* that have been published to a topic. A <CODE>TopicSubscriber</CODE> object

6
activemq-core/src/main/java/org/apache/activemq/ActiveMQXAConnectionFactory.java
View File

@ -16,8 +16,7 @@
*/
package org.apache.activemq;
import org.apache.activemq.management.JMSStatsImpl;
import org.apache.activemq.transport.Transport;
import java.net.URI;
import javax.jms.JMSException;
import javax.jms.XAConnection;
@ -27,7 +26,8 @@ import javax.jms.XAQueueConnectionFactory;
import javax.jms.XATopicConnection;
import javax.jms.XATopicConnectionFactory;
import java.net.URI;
import org.apache.activemq.management.JMSStatsImpl;
import org.apache.activemq.transport.Transport;
/**
* A factory of {@link XAConnection} instances

9
activemq-core/src/main/java/org/apache/activemq/BlobMessage.java
View File

@ -16,11 +16,12 @@
*/
package org.apache.activemq;
import javax.jms.JMSException;
import java.net.URL;
import java.net.MalformedURLException;
import java.io.InputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.MalformedURLException;
import java.net.URL;
import javax.jms.JMSException;
/**
* Represents a message which has a typically out of band Binary Large Object

6
activemq-core/src/main/java/org/apache/activemq/CustomDestination.java
View File

@ -17,13 +17,13 @@
package org.apache.activemq;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.TopicSubscriber;
import javax.jms.QueueReceiver;
import javax.jms.TopicPublisher;
import javax.jms.QueueSender;
import javax.jms.JMSException;
import javax.jms.TopicPublisher;
import javax.jms.TopicSubscriber;
/**
* Represents a hook to allow the support of custom destinations

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

@ -16,13 +16,11 @@
*/
package org.apache.activemq;
import org.apache.activemq.command.ActiveMQMessage;
import javax.jms.Message;
import javax.jms.Session;
import javax.jms.MessageProducer;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
import javax.jms.Session;
/**
* A plugin strategy for transforming a message before it is sent by the JMS client or before it is

5
activemq-core/src/main/java/org/apache/activemq/TransactionContext.java
View File

@ -19,6 +19,7 @@ package org.apache.activemq;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;
import javax.jms.JMSException;
import javax.jms.TransactionInProgressException;
@ -29,20 +30,18 @@ import javax.transaction.xa.Xid;
import org.apache.activemq.command.ConnectionId;
import org.apache.activemq.command.DataArrayResponse;
import org.apache.activemq.command.DataStructure;
import org.apache.activemq.command.IntegerResponse;
import org.apache.activemq.command.LocalTransactionId;
import org.apache.activemq.command.TransactionId;
import org.apache.activemq.command.TransactionInfo;
import org.apache.activemq.command.XATransactionId;
import org.apache.activemq.command.DataStructure;
import org.apache.activemq.transaction.Synchronization;
import org.apache.activemq.util.JMSExceptionSupport;
import org.apache.activemq.util.LongSequenceGenerator;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.util.concurrent.ConcurrentHashMap;
/**
* A TransactionContext provides the means to control a JMS transaction. It
* provides a local transaction interface and also an XAResource interface. <p/>

4
activemq-core/src/main/java/org/apache/activemq/advisory/AdvisoryBroker.java
View File

@ -16,8 +16,8 @@
*/
package org.apache.activemq.advisory;
import java.io.IOException;
import java.util.Iterator;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.BrokerFilter;
@ -42,8 +42,6 @@ import org.apache.activemq.util.LongSequenceGenerator;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.util.concurrent.ConcurrentHashMap;
/**
* This broker filter handles tracking the state of the broker for purposes of
* publishing advisory messages to advisory consumers.

7
activemq-core/src/main/java/org/apache/activemq/advisory/AdvisorySupport.java
View File

@ -16,11 +16,11 @@
*/
package org.apache.activemq.advisory;
import javax.jms.Destination;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQTopic;
import javax.jms.Destination;
public class AdvisorySupport {
public static final String ADVISORY_TOPIC_PREFIX = "ActiveMQ.Advisory.";
@ -102,8 +102,9 @@ public class AdvisorySupport {
return TEMP_QUEUE_ADVISORY_TOPIC;
case ActiveMQDestination.TEMP_TOPIC_TYPE:
return TEMP_TOPIC_ADVISORY_TOPIC;
default:
throw new RuntimeException("Unknown destination type: " + destination.getDestinationType());
}
throw new RuntimeException("Unknown destination type: " + destination.getDestinationType());
}
public static boolean isDestinationAdvisoryTopic(ActiveMQDestination destination) {

4
activemq-core/src/main/java/org/apache/activemq/advisory/ConsumerEvent.java
View File

@ -16,11 +16,11 @@
*/
package org.apache.activemq.advisory;
import org.apache.activemq.command.ConsumerId;
import java.util.EventObject;
import javax.jms.Destination;
import java.util.EventObject;
import org.apache.activemq.command.ConsumerId;
/**
* An event when the number of consumers on a given destination changes.

38
activemq-core/src/main/java/org/apache/activemq/advisory/ConsumerEventSource.java
View File

@ -19,6 +19,14 @@ package org.apache.activemq.advisory;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import org.apache.activemq.Service;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQMessage;
@ -29,14 +37,6 @@ import org.apache.activemq.command.RemoveInfo;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
/**
* An object which can be used to listen to the number of active consumers
* available on a given destination.
@ -82,27 +82,24 @@ public class ConsumerEventSource implements Service, MessageListener {
public void onMessage(Message message) {
if (message instanceof ActiveMQMessage) {
ActiveMQMessage activeMessage = (ActiveMQMessage) message;
ActiveMQMessage activeMessage = (ActiveMQMessage)message;
Object command = activeMessage.getDataStructure();
int count = 0;
if (command instanceof ConsumerInfo) {
count = consumerCount.incrementAndGet();
count = extractConsumerCountFromMessage(message, count);
fireConsumerEvent(new ConsumerStartedEvent(this, destination, (ConsumerInfo) command, count));
}
else if (command instanceof RemoveInfo) {
RemoveInfo removeInfo = (RemoveInfo) command;
fireConsumerEvent(new ConsumerStartedEvent(this, destination, (ConsumerInfo)command, count));
} else if (command instanceof RemoveInfo) {
RemoveInfo removeInfo = (RemoveInfo)command;
if (removeInfo.isConsumerRemove()) {
count = consumerCount.decrementAndGet();
count = extractConsumerCountFromMessage(message, count);
fireConsumerEvent(new ConsumerStoppedEvent(this, destination, (ConsumerId) removeInfo.getObjectId(), count));
fireConsumerEvent(new ConsumerStoppedEvent(this, destination, (ConsumerId)removeInfo.getObjectId(), count));
}
}
else {
} else {
log.warn("Unknown command: " + command);
}
}
else {
} else {
log.warn("Unknown message type: " + message + ". Message ignored");
}
}
@ -116,12 +113,11 @@ public class ConsumerEventSource implements Service, MessageListener {
try {
Object value = message.getObjectProperty("consumerCount");
if (value instanceof Number) {
Number n = (Number) value;
Number n = (Number)value;
return n.intValue();
}
log.warn("No consumerCount header available on the message: " + message);
}
catch (Exception e) {
} catch (Exception e) {
log.warn("Failed to extract consumerCount from message: " + message + ".Reason: " + e, e);
}
return count;

4
activemq-core/src/main/java/org/apache/activemq/advisory/ProducerEvent.java
View File

@ -16,11 +16,11 @@
*/
package org.apache.activemq.advisory;
import org.apache.activemq.command.ProducerId;
import java.util.EventObject;
import javax.jms.Destination;
import java.util.EventObject;
import org.apache.activemq.command.ProducerId;
/**
* An event when the number of producers on a given destination changes.

28
activemq-core/src/main/java/org/apache/activemq/advisory/ProducerEventSource.java
View File

@ -16,6 +16,9 @@
*/
package org.apache.activemq.advisory;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import javax.jms.Connection;
import javax.jms.Destination;
import javax.jms.JMSException;
@ -23,6 +26,7 @@ import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import javax.jms.Session;
import org.apache.activemq.Service;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQMessage;
@ -32,8 +36,6 @@ import org.apache.activemq.command.ProducerInfo;
import org.apache.activemq.command.RemoveInfo;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
/**
* An object which can be used to listen to the number of active consumers
@ -80,27 +82,24 @@ public class ProducerEventSource implements Service, MessageListener {
public void onMessage(Message message) {
if (message instanceof ActiveMQMessage) {
ActiveMQMessage activeMessage = (ActiveMQMessage) message;
ActiveMQMessage activeMessage = (ActiveMQMessage)message;
Object command = activeMessage.getDataStructure();
int count = 0;
if (command instanceof ProducerInfo) {
count = producerCount.incrementAndGet();
count = extractProducerCountFromMessage(message, count);
fireProducerEvent(new ProducerStartedEvent(this, destination, (ProducerInfo) command, count));
}
else if (command instanceof RemoveInfo) {
RemoveInfo removeInfo = (RemoveInfo) command;
fireProducerEvent(new ProducerStartedEvent(this, destination, (ProducerInfo)command, count));
} else if (command instanceof RemoveInfo) {
RemoveInfo removeInfo = (RemoveInfo)command;
if (removeInfo.isProducerRemove()) {
count = producerCount.decrementAndGet();
count = extractProducerCountFromMessage(message, count);
fireProducerEvent(new ProducerStoppedEvent(this, destination, (ProducerId) removeInfo.getObjectId(), count));
fireProducerEvent(new ProducerStoppedEvent(this, destination, (ProducerId)removeInfo.getObjectId(), count));
}
}
else {
} else {
log.warn("Unknown command: " + command);
}
}
else {
} else {
log.warn("Unknown message type: " + message + ". Message ignored");
}
}
@ -109,12 +108,11 @@ public class ProducerEventSource implements Service, MessageListener {
try {
Object value = message.getObjectProperty("producerCount");
if (value instanceof Number) {
Number n = (Number) value;
Number n = (Number)value;
return n.intValue();
}
log.warn("No producerCount header available on the message: " + message);
}
catch (Exception e) {
} catch (Exception e) {
log.warn("Failed to extract producerCount from message: " + message + ".Reason: " + e, e);
}
return count;

7
activemq-core/src/main/java/org/apache/activemq/blob/BlobUploadStrategy.java
View File

@ -16,14 +16,15 @@
*/
package org.apache.activemq.blob;
import org.apache.activemq.command.ActiveMQBlobMessage;
import javax.jms.JMSException;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.net.URL;
import javax.jms.JMSException;
import org.apache.activemq.command.ActiveMQBlobMessage;
/**
* Represents a strategy of uploading a file/stream to some remote
*

14
activemq-core/src/main/java/org/apache/activemq/blob/BlobUploader.java
View File

@ -16,17 +16,18 @@
*/
package org.apache.activemq.blob;
import org.apache.activemq.command.ActiveMQBlobMessage;
import javax.jms.JMSException;
import java.io.File;
import java.io.IOException;
import java.io.InputStream;
import java.net.URL;
import javax.jms.JMSException;
import org.apache.activemq.command.ActiveMQBlobMessage;
/**
* A helper class to represent a required upload of a BLOB to some remote URL
*
*
* @version $Revision: $
*/
public class BlobUploader {
@ -35,7 +36,6 @@ public class BlobUploader {
private File file;
private InputStream in;
public BlobUploader(BlobTransferPolicy blobTransferPolicy, InputStream in) {
this.blobTransferPolicy = blobTransferPolicy;
this.in = in;
@ -49,13 +49,11 @@ public class BlobUploader {
public URL upload(ActiveMQBlobMessage message) throws JMSException, IOException {
if (file != null) {
return getStrategy().uploadFile(message, file);
}
else {
} else {
return getStrategy().uploadStream(message, in);
}
}
public BlobTransferPolicy getBlobTransferPolicy() {
return blobTransferPolicy;
}

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

@ -30,9 +30,11 @@ import java.util.Set;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.management.MBeanServer;
import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;
import org.apache.activemq.ActiveMQConnectionMetaData;
import org.apache.activemq.Service;
import org.apache.activemq.advisory.AdvisoryBroker;
@ -77,10 +79,10 @@ import org.apache.activemq.transport.TransportFactory;
import org.apache.activemq.transport.TransportServer;
import org.apache.activemq.transport.vm.VMTransportFactory;
import org.apache.activemq.util.IOExceptionSupport;
import org.apache.activemq.util.IOHelper;
import org.apache.activemq.util.JMXSupport;
import org.apache.activemq.util.ServiceStopper;
import org.apache.activemq.util.URISupport;
import org.apache.activemq.util.IOHelper;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

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

@ -16,9 +16,9 @@
*/
package org.apache.activemq.broker;
import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.command.ConnectionId;
@ -29,8 +29,6 @@ import org.apache.activemq.security.MessageAuthorizationPolicy;
import org.apache.activemq.security.SecurityContext;
import org.apache.activemq.transaction.Transaction;
import java.io.IOException;
/**
* Used to hold context information needed to process requests sent to a broker.
*

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

@ -16,10 +16,10 @@
*/
package org.apache.activemq.broker;
import org.apache.activemq.command.ActiveMQDestination;
import javax.jms.JMSException;
import org.apache.activemq.command.ActiveMQDestination;
/**
* An exception thrown if a destination is attempted to be created when it already exists.
*

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

@ -16,18 +16,18 @@
*/
package org.apache.activemq.broker;
import org.apache.activemq.util.IntrospectionSupport;
import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.MalformedURLException;
import java.net.URI;
import java.net.URL;
import java.net.MalformedURLException;
import java.util.Map;
import java.util.Properties;
import org.apache.activemq.util.IntrospectionSupport;
/**
* A {@link BrokerFactoryHandler} which uses a properties file to configure the
* broker's various policies.

8
activemq-core/src/main/java/org/apache/activemq/broker/SslBrokerService.java
View File

@ -17,10 +17,6 @@
package org.apache.activemq.broker;
import org.apache.activemq.transport.TransportFactory;
import org.apache.activemq.transport.TransportServer;
import org.apache.activemq.transport.tcp.SslTransportFactory;
import java.io.IOException;
import java.net.URI;
import java.security.KeyManagementException;
@ -29,6 +25,10 @@ import java.security.SecureRandom;
import javax.net.ssl.KeyManager;
import javax.net.ssl.TrustManager;
import org.apache.activemq.transport.TransportFactory;
import org.apache.activemq.transport.TransportServer;
import org.apache.activemq.transport.tcp.SslTransportFactory;
/**
* A BrokerService that allows access to the key and trust managers used by SSL
* connections. There is no reason to use this class unless SSL is being used

16
activemq-core/src/main/java/org/apache/activemq/broker/TransactionBroker.java
View File

@ -16,8 +16,15 @@
*/
package org.apache.activemq.broker;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import javax.jms.JMSException;
import javax.transaction.xa.XAException;
import org.apache.activemq.ActiveMQMessageAudit;
import org.apache.activemq.command.ConnectionInfo;
import org.apache.activemq.command.LocalTransactionId;
@ -36,15 +43,6 @@ import org.apache.activemq.util.WrappedException;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import javax.jms.JMSException;
import javax.transaction.xa.XAException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.Map;
/**
* This broker filter handles the transaction related operations in the Broker
* interface.

69
activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java
View File

@ -30,6 +30,7 @@ import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import org.apache.activemq.Service;
import org.apache.activemq.broker.ft.MasterBroker;
import org.apache.activemq.broker.region.ConnectionStatistics;
@ -97,8 +98,7 @@ import org.apache.commons.logging.LogFactory;
public class TransportConnection implements Service, Connection, Task, CommandVisitor {
private static final Log LOG = LogFactory.getLog(TransportConnection.class);
private static final Log TRANSPORTLOG = LogFactory.getLog(TransportConnection.class.getName()
+ ".Transport");
private static final Log TRANSPORTLOG = LogFactory.getLog(TransportConnection.class.getName() + ".Transport");
private static final Log SERVICELOG = LogFactory.getLog(TransportConnection.class.getName() + ".Service");
// Keeps track of the broker and connector that created this connection.
protected final Broker broker;
@ -191,8 +191,7 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
* @param taskRunnerFactory - can be null if you want direct dispatch to the
* transport else commands are sent async.
*/
public TransportConnection(TransportConnector connector, final Transport transport, Broker broker,
TaskRunnerFactory taskRunnerFactory) {
public TransportConnection(TransportConnector connector, final Transport transport, Broker broker, TaskRunnerFactory taskRunnerFactory) {
this.connector = connector;
this.broker = broker;
RegionBroker rb = (RegionBroker)broker.getAdaptor(RegionBroker.class);
@ -272,8 +271,7 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
else if (e.getClass() == BrokerStoppedException.class) {
if (!disposed.get()) {
if (SERVICELOG.isDebugEnabled())
SERVICELOG
.debug("Broker has been stopped. Notifying client and closing his connection.");
SERVICELOG.debug("Broker has been stopped. Notifying client and closing his connection.");
ConnectionError ce = new ConnectionError();
ce.setException(e);
dispatchSync(ce);
@ -403,8 +401,7 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
}
TransactionState transactionState = cs.getTransactionState(info.getTransactionId());
if (transactionState == null)
throw new IllegalStateException("Cannot prepare a transaction that had not been started: "
+ info.getTransactionId());
throw new IllegalStateException("Cannot prepare a transaction that had not been started: " + info.getTransactionId());
// Avoid dups.
if (!transactionState.isPrepared()) {
transactionState.setPrepared(true);
@ -473,8 +470,7 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
return broker.messagePull(lookupConnectionState(pull.getConsumerId()).getContext(), pull);
}
public Response processMessageDispatchNotification(MessageDispatchNotification notification)
throws Exception {
public Response processMessageDispatchNotification(MessageDispatchNotification notification) throws Exception {
broker.processDispatchNotification(notification);
return null;
}
@ -503,9 +499,7 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
TransportConnectionState cs = lookupConnectionState(connectionId);
SessionState ss = cs.getSessionState(sessionId);
if (ss == null)
throw new IllegalStateException(
"Cannot add a producer to a session that had not been registered: "
+ sessionId);
throw new IllegalStateException("Cannot add a producer to a session that had not been registered: " + sessionId);
// Avoid replaying dup commands
if (!ss.getProducerIds().contains(info.getProducerId())) {
broker.addProducer(cs.getContext(), info);
@ -524,9 +518,7 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
TransportConnectionState cs = lookupConnectionState(connectionId);
SessionState ss = cs.getSessionState(sessionId);
if (ss == null)
throw new IllegalStateException(
"Cannot remove a producer from a session that had not been registered: "
+ sessionId);
throw new IllegalStateException("Cannot remove a producer from a session that had not been registered: " + sessionId);
ProducerState ps = ss.removeProducer(id);
if (ps == null)
throw new IllegalStateException("Cannot remove a producer that had not been registered: " + id);
@ -541,9 +533,7 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
TransportConnectionState cs = lookupConnectionState(connectionId);
SessionState ss = cs.getSessionState(sessionId);
if (ss == null)
throw new IllegalStateException(
"Cannot add a consumer to a session that had not been registered: "
+ sessionId);
throw new IllegalStateException("Cannot add a consumer to a session that had not been registered: " + sessionId);
// Avoid replaying dup commands
if (!ss.getConsumerIds().contains(info.getConsumerId())) {
broker.addConsumer(cs.getContext(), info);
@ -562,9 +552,7 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
TransportConnectionState cs = lookupConnectionState(connectionId);
SessionState ss = cs.getSessionState(sessionId);
if (ss == null)
throw new IllegalStateException(
"Cannot remove a consumer from a session that had not been registered: "
+ sessionId);
throw new IllegalStateException("Cannot remove a consumer from a session that had not been registered: " + sessionId);
ConsumerState consumerState = ss.removeConsumer(id);
if (consumerState == null)
throw new IllegalStateException("Cannot remove a consumer that had not been registered: " + id);
@ -641,8 +629,7 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
if (state.getConnection() != this) {
LOG.debug("Killing previous stale connection: " + state.getConnection().getRemoteAddress());
state.getConnection().stop();
LOG.debug("Connection " + getRemoteAddress() + " taking over previous connection: "
+ state.getConnection().getRemoteAddress());
LOG.debug("Connection " + getRemoteAddress() + " taking over previous connection: " + state.getConnection().getRemoteAddress());
state.setConnection(this);
state.reset(info);
}
@ -765,8 +752,7 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
}
protected void processDispatch(Command command) throws IOException {
final MessageDispatch messageDispatch = (MessageDispatch)(command.isMessageDispatch()
? command : null);
final MessageDispatch messageDispatch = (MessageDispatch)(command.isMessageDispatch() ? command : null);
try {
if (!disposed.get()) {
if (messageDispatch != null) {
@ -846,8 +832,7 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
transport.start();
if (taskRunnerFactory != null) {
taskRunner = taskRunnerFactory.createTaskRunner(this, "ActiveMQ Connection Dispatcher: "
+ getRemoteAddress());
taskRunner = taskRunnerFactory.createTaskRunner(this, "ActiveMQ Connection Dispatcher: " + getRemoteAddress());
} else {
taskRunner = null;
}
@ -1114,8 +1099,7 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
uri = URISupport.createURIWithQuery(uri, URISupport.createQueryString(map));
Transport localTransport = TransportFactory.connect(uri);
Transport remoteBridgeTransport = new ResponseCorrelator(transport);
duplexBridge = NetworkBridgeFactory.createBridge(config, localTransport,
remoteBridgeTransport);
duplexBridge = NetworkBridgeFactory.createBridge(config, localTransport, remoteBridgeTransport);
// now turn duplex off this side
info.setDuplexConnection(false);
duplexBridge.setCreatedByDuplex(true);
@ -1180,8 +1164,7 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
ProducerState producerState = ss.getProducerState(id);
if (producerState != null && producerState.getInfo() != null) {
ProducerInfo info = producerState.getInfo();
result.setMutable(info.getDestination() == null
|| info.getDestination().isComposite());
result.setMutable(info.getDestination() == null || info.getDestination().isComposite());
}
}
producerExchanges.put(id, result);
@ -1285,8 +1268,7 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
//
// /////////////////////////////////////////////////////////////////
protected TransportConnectionState registerConnectionState(ConnectionId connectionId,
TransportConnectionState state) {
protected TransportConnectionState registerConnectionState(ConnectionId connectionId, TransportConnectionState state) {
TransportConnectionState rc = connectionState;
connectionState = state;
return rc;
@ -1309,44 +1291,35 @@ public class TransportConnection implements Service, Connection, Task, CommandVi
protected TransportConnectionState lookupConnectionState(String connectionId) {
TransportConnectionState cs = connectionState;
if (cs == null)
throw new IllegalStateException(
"Cannot lookup a connectionId for a connection that had not been registered: "
+ connectionId);
throw new IllegalStateException("Cannot lookup a connectionId for a connection that had not been registered: " + connectionId);
return cs;
}
protected TransportConnectionState lookupConnectionState(ConsumerId id) {
TransportConnectionState cs = connectionState;
if (cs == null)
throw new IllegalStateException(
"Cannot lookup a consumer from a connection that had not been registered: "
+ id.getParentId().getParentId());
throw new IllegalStateException("Cannot lookup a consumer from a connection that had not been registered: " + id.getParentId().getParentId());
return cs;
}
protected TransportConnectionState lookupConnectionState(ProducerId id) {
TransportConnectionState cs = connectionState;
if (cs == null)
throw new IllegalStateException(
"Cannot lookup a producer from a connection that had not been registered: "
+ id.getParentId().getParentId());
throw new IllegalStateException("Cannot lookup a producer from a connection that had not been registered: " + id.getParentId().getParentId());
return cs;
}
protected TransportConnectionState lookupConnectionState(SessionId id) {
TransportConnectionState cs = connectionState;
if (cs == null)
throw new IllegalStateException(
"Cannot lookup a session from a connection that had not been registered: "
+ id.getParentId());
throw new IllegalStateException("Cannot lookup a session from a connection that had not been registered: " + id.getParentId());
return cs;
}
protected TransportConnectionState lookupConnectionState(ConnectionId connectionId) {
TransportConnectionState cs = connectionState;
if (cs == null)
throw new IllegalStateException("Cannot lookup a connection that had not been registered: "
+ connectionId);
throw new IllegalStateException("Cannot lookup a connection that had not been registered: " + connectionId);
return cs;
}

3
activemq-core/src/main/java/org/apache/activemq/broker/TransportConnector.java
View File

@ -20,6 +20,7 @@ import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Iterator;
import java.util.concurrent.CopyOnWriteArrayList;
import javax.management.MBeanServer;
import javax.management.ObjectName;
@ -40,8 +41,6 @@ import org.apache.activemq.util.ServiceSupport;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.util.concurrent.CopyOnWriteArrayList;
/**
* @org.apache.xbean.XBean
* @version $Revision: 1.6 $

5
activemq-core/src/main/java/org/apache/activemq/broker/TransportStatusDetector.java
View File

@ -16,6 +16,8 @@
*/
package org.apache.activemq.broker;
import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.CopyOnWriteArraySet;
import java.util.concurrent.atomic.AtomicBoolean;
@ -24,9 +26,6 @@ import org.apache.activemq.ThreadPriorities;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.util.Iterator;
import java.util.Set;
/**
* Used to provide information on the status of the Connection
*

2
activemq-core/src/main/java/org/apache/activemq/broker/ft/MasterConnector.java
View File

@ -20,6 +20,7 @@ import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.activemq.Service;
import org.apache.activemq.broker.BrokerService;
@ -43,7 +44,6 @@ import org.apache.activemq.util.ServiceStopper;
import org.apache.activemq.util.ServiceSupport;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* Connects a Slave Broker to a Master when using <a

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

@ -35,7 +35,6 @@ import javax.management.openmbean.TabularDataSupport;
import javax.management.openmbean.TabularType;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.jmx.OpenTypeSupport.OpenTypeFactory;
import org.apache.activemq.broker.region.Destination;
import org.apache.activemq.command.ActiveMQDestination;

6
activemq-core/src/main/java/org/apache/activemq/broker/jmx/DestinationViewMBean.java
View File

@ -16,14 +16,14 @@
*/
package org.apache.activemq.broker.jmx;
import java.util.List;
import java.util.Map;
import javax.jms.InvalidSelectorException;
import javax.management.openmbean.CompositeData;
import javax.management.openmbean.OpenDataException;
import javax.management.openmbean.TabularData;
import java.util.List;
import java.util.Map;
public interface DestinationViewMBean {

9
activemq-core/src/main/java/org/apache/activemq/broker/jmx/DurableSubscriptionViewMBean.java
View File

@ -16,16 +16,14 @@
*/
package org.apache.activemq.broker.jmx;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.command.RemoveSubscriptionInfo;
import javax.management.openmbean.CompositeData;
import javax.management.openmbean.OpenDataException;
import javax.management.openmbean.TabularData;
/**
* @version $Revision: 1.5 $
*/
public interface DurableSubscriptionViewMBean extends SubscriptionViewMBean{
public interface DurableSubscriptionViewMBean extends SubscriptionViewMBean {
/**
* @return name of the durable subscription name
*/
@ -48,7 +46,8 @@ public interface DurableSubscriptionViewMBean extends SubscriptionViewMBean{
public TabularData browseAsTable() throws OpenDataException;
/**
* Destroys the durable subscription so that messages will no longer be stored for this subscription
* Destroys the durable subscription so that messages will no longer be
* stored for this subscription
*/
public void destroy() throws Exception;
}

7
activemq-core/src/main/java/org/apache/activemq/broker/jmx/ManagedRegionBroker.java
View File

@ -23,8 +23,10 @@ import java.util.Hashtable;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import javax.management.InstanceNotFoundException;
import javax.management.MBeanServer;
@ -70,9 +72,6 @@ import org.apache.activemq.util.SubscriptionKey;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
public class ManagedRegionBroker extends RegionBroker {
private static final Log log = LogFactory.getLog(ManagedRegionBroker.class);
private final MBeanServer mbeanServer;

12
activemq-core/src/main/java/org/apache/activemq/broker/jmx/ManagedTransportConnection.java
View File

@ -16,6 +16,12 @@
*/
package org.apache.activemq.broker.jmx;
import java.io.IOException;
import java.util.Hashtable;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.TransportConnection;
import org.apache.activemq.broker.TransportConnector;
@ -28,12 +34,6 @@ import org.apache.activemq.util.JMXSupport;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import java.io.IOException;
import java.util.Hashtable;
/**
* A managed transport connection
*

12
activemq-core/src/main/java/org/apache/activemq/broker/jmx/ManagedTransportConnector.java
View File

@ -16,18 +16,18 @@
*/
package org.apache.activemq.broker.jmx;
import java.io.IOException;
import java.net.URISyntaxException;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.Connection;
import org.apache.activemq.broker.TransportConnector;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.transport.TransportServer;
import javax.management.MBeanServer;
import javax.management.ObjectName;
import java.io.IOException;
import java.net.URISyntaxException;
/**
* A managed transport connector which can create multiple managed connections
* as clients connect.

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

@ -17,11 +17,12 @@
package org.apache.activemq.broker.jmx;
import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.net.MalformedURLException;
import java.rmi.registry.LocateRegistry;
import java.util.List;
import java.util.concurrent.atomic.AtomicBoolean;
import javax.management.Attribute;
import javax.management.JMException;
import javax.management.MBeanServer;
@ -31,11 +32,10 @@ import javax.management.ObjectName;
import javax.management.remote.JMXConnectorServer;
import javax.management.remote.JMXConnectorServerFactory;
import javax.management.remote.JMXServiceURL;
import org.apache.activemq.Service;
import org.apache.activemq.util.ClassLoading;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.util.concurrent.atomic.AtomicBoolean;
/**
* A Flow provides different dispatch policies within the NMR

4
activemq-core/src/main/java/org/apache/activemq/broker/jmx/SubscriptionView.java
View File

@ -16,12 +16,12 @@
*/
package org.apache.activemq.broker.jmx;
import javax.jms.InvalidSelectorException;
import org.apache.activemq.broker.region.Subscription;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ConsumerInfo;
import javax.jms.InvalidSelectorException;
/**
* @version $Revision: 1.5 $
*/

4
activemq-core/src/main/java/org/apache/activemq/broker/jmx/TopicView.java
View File

@ -16,11 +16,7 @@
*/
package org.apache.activemq.broker.jmx;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.Topic;
import org.apache.activemq.command.ConsumerId;
import org.apache.activemq.command.ConsumerInfo;
import org.apache.activemq.command.RemoveSubscriptionInfo;
public class TopicView extends DestinationView implements TopicViewMBean {

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

@ -16,12 +16,13 @@
*/
package org.apache.activemq.broker.region;
import java.io.IOException;
import java.util.concurrent.CopyOnWriteArrayList;
import javax.jms.InvalidSelectorException;
import javax.jms.JMSException;
import javax.management.ObjectName;
import java.io.IOException;
import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.command.ActiveMQDestination;
@ -36,8 +37,6 @@ import org.apache.activemq.selector.SelectorParser;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.util.concurrent.CopyOnWriteArrayList;
abstract public class AbstractSubscription implements Subscription {
static private final Log log = LogFactory.getLog(AbstractSubscription.class);

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

@ -17,7 +17,6 @@
package org.apache.activemq.broker.region;
import org.apache.activemq.management.CountStatisticImpl;
import org.apache.activemq.management.StatsImpl;
@ -53,7 +52,7 @@ public class ConnectionStatistics extends StatsImpl {
enqueues.reset();
dequeues.reset();
}
public void setEnabled(boolean enabled) {
super.setEnabled(enabled);
enqueues.setEnabled(enabled);
@ -64,12 +63,10 @@ public class ConnectionStatistics extends StatsImpl {
if (parent != null) {
enqueues.setParent(parent.getEnqueues());
dequeues.setParent(parent.getDequeues());
}
else {
} else {
enqueues.setParent(null);
dequeues.setParent(null);
}
}
}

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

@ -17,7 +17,6 @@
package org.apache.activemq.broker.region;
import org.apache.activemq.command.Message;
import org.apache.activemq.management.CountStatisticImpl;
import org.apache.activemq.management.PollCountStatisticImpl;
import org.apache.activemq.management.StatsImpl;

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

@ -16,10 +16,10 @@
*/
package org.apache.activemq.broker.region;
import org.apache.activemq.broker.ConnectionContext;
import javax.jms.JMSException;
import org.apache.activemq.broker.ConnectionContext;
/**
* Represents a filter on message references
*

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

@ -17,14 +17,14 @@ package org.apache.activemq.broker.region;
import java.io.IOException;
import java.util.Iterator;
import java.util.LinkedList;
import javax.jms.InvalidSelectorException;
import javax.jms.JMSException;
import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.cursors.PendingMessageCursor;
import org.apache.activemq.broker.region.cursors.VMPendingMessageCursor;
import org.apache.activemq.broker.region.policy.DeadLetterStrategy;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ConsumerControl;
import org.apache.activemq.command.ConsumerInfo;
import org.apache.activemq.command.Message;
@ -36,7 +36,6 @@ import org.apache.activemq.command.MessagePull;
import org.apache.activemq.command.Response;
import org.apache.activemq.thread.Scheduler;
import org.apache.activemq.transaction.Synchronization;
import org.apache.activemq.util.BrokerSupport;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;

10
activemq-core/src/main/java/org/apache/activemq/broker/region/QueueRegion.java
View File

@ -16,17 +16,17 @@
*/
package org.apache.activemq.broker.region;
import java.util.Iterator;
import java.util.Set;
import javax.jms.InvalidSelectorException;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ConsumerInfo;
import org.apache.activemq.memory.UsageManager;
import org.apache.activemq.thread.TaskRunnerFactory;
import javax.jms.InvalidSelectorException;
import java.util.Iterator;
import java.util.Set;
/**
*
* @version $Revision: 1.9 $

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

@ -16,6 +16,9 @@
*/
package org.apache.activemq.broker.region;
import java.util.Map;
import java.util.Set;
import org.apache.activemq.Service;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.ConsumerBrokerExchange;
@ -29,9 +32,6 @@ import org.apache.activemq.command.MessagePull;
import org.apache.activemq.command.RemoveSubscriptionInfo;
import org.apache.activemq.command.Response;
import java.util.Map;
import java.util.Set;
/**
* A Region is used to implement the different QOS options available to
* a broker. A Broker is composed of multiple message processing Regions that

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

@ -18,6 +18,9 @@ package org.apache.activemq.broker.region;
import java.io.IOException;
import javax.jms.InvalidSelectorException;
import javax.management.ObjectName;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ConsumerInfo;
@ -27,9 +30,6 @@ import org.apache.activemq.command.MessagePull;
import org.apache.activemq.command.Response;
import org.apache.activemq.filter.MessageEvaluationContext;
import javax.jms.InvalidSelectorException;
import javax.management.ObjectName;
/**
* @version $Revision: 1.5 $
*/

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

@ -18,7 +18,7 @@ package org.apache.activemq.broker.region;
import javax.jms.InvalidSelectorException;
import javax.jms.JMSException;
import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQTempDestination;

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

@ -15,9 +15,8 @@
package org.apache.activemq.broker.region;
import javax.jms.JMSException;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.broker.region.cursors.VMPendingMessageCursor;
import org.apache.activemq.broker.region.policy.PolicyEntry;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ConsumerInfo;
@ -31,7 +30,7 @@ import org.apache.commons.logging.LogFactory;
*/
public class TempTopicRegion extends AbstractRegion {
private static final Log log = LogFactory.getLog(TempTopicRegion.class);
private static final Log LOG = LogFactory.getLog(TempTopicRegion.class);
public TempTopicRegion(RegionBroker broker, DestinationStatistics destinationStatistics, UsageManager memoryManager, TaskRunnerFactory taskRunnerFactory,
DestinationFactory destinationFactory) {
@ -58,7 +57,7 @@ public class TempTopicRegion extends AbstractRegion {
answer.init();
return answer;
} catch (Exception e) {
log.error("Failed to create TopicSubscription ", e);
LOG.error("Failed to create TopicSubscription ", e);
JMSException jmsEx = new JMSException("Couldn't create TopicSubscription");
jmsEx.setLinkedException(e);
throw jmsEx;

3
activemq-core/src/main/java/org/apache/activemq/broker/region/TopicSubscription.java
View File

@ -17,8 +17,9 @@ package org.apache.activemq.broker.region;
import java.io.IOException;
import java.util.LinkedList;
import java.util.concurrent.atomic.AtomicLong;
import javax.jms.InvalidSelectorException;
import javax.jms.JMSException;
import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.cursors.FilePendingMessageCursor;

84
activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/PendingMessageCursor.java
View File

@ -23,15 +23,16 @@ import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.memory.UsageManager;
/**
* Interface to pending message (messages awaiting disptach to a consumer) cursor
* Interface to pending message (messages awaiting disptach to a consumer)
* cursor
*
* @version $Revision$
*/
public interface PendingMessageCursor extends Service{
public interface PendingMessageCursor extends Service {
/**
* Add a destination
*
* @param context
* @param destination
* @throws Exception
@ -40,53 +41,57 @@ public interface PendingMessageCursor extends Service{
/**
* remove a destination
*
* @param context
* @param destination
* @throws Exception
*/
public void remove(ConnectionContext context, Destination destination) throws Exception;
/**
* @return true if there are no pending messages
*/
public boolean isEmpty();
/**
* check if a Destination is Empty for this cursor
*
* @param destination
* @return true id the Destination is empty
*/
public boolean isEmpty(Destination destination);
/**
* reset the cursor
*
*/
public void reset();
/**
* hint to the cursor to release any locks it might have
* grabbed after a reset
*
* hint to the cursor to release any locks it might have grabbed after a
* reset
*/
public void release();
/**
* add message to await dispatch
*
* @param node
* @throws IOException
* @throws Exception
* @throws IOException
* @throws Exception
*/
public void addMessageLast(MessageReference node) throws Exception;
public void addMessageLast(MessageReference node) throws Exception;
/**
* add message to await dispatch
*
* @param node
* @throws Exception
* @throws Exception
*/
public void addMessageFirst(MessageReference node) throws Exception;
/**
* Add a message recovered from a retroactive policy
*
* @param node
* @throws Exception
*/
@ -104,7 +109,6 @@ public interface PendingMessageCursor extends Service{
/**
* remove the message at the cursor position
*
*/
public void remove();
@ -115,17 +119,17 @@ public interface PendingMessageCursor extends Service{
/**
* clear all pending messages
*
*/
public void clear();
/**
* Informs the Broker if the subscription needs to intervention to recover it's state
* e.g. DurableTopicSubscriber may do
* Informs the Broker if the subscription needs to intervention to recover
* it's state e.g. DurableTopicSubscriber may do
*
* @return true if recovery required
*/
public boolean isRecoveryRequired();
/**
* @return the maximum batch size
*/
@ -133,75 +137,75 @@ public interface PendingMessageCursor extends Service{
/**
* Set the max batch size
*
* @param maxBatchSize
*/
public void setMaxBatchSize(int maxBatchSize);
/**
* Give the cursor a hint that we are about to remove
* messages from memory only
* Give the cursor a hint that we are about to remove messages from memory
* only
*/
public void resetForGC();
/**
* remove a node
*
* @param node
*/
public void remove(MessageReference node);
/**
* free up any internal buffers
*
*/
public void gc();
/**
* Set the UsageManager
*
* @param usageManager
* @see org.apache.activemq.memory.UsageManager
*/
public void setUsageManager(UsageManager usageManager);
/**
* @return the usageManager
*/
public UsageManager getUsageManager();
/**
* @return the memoryUsageHighWaterMark
*/
public int getMemoryUsageHighWaterMark();
/**
* @param memoryUsageHighWaterMark the memoryUsageHighWaterMark to set
*/
public void setMemoryUsageHighWaterMark(int memoryUsageHighWaterMark);
/**
* @return true if the cursor is full
*/
public boolean isFull();
/**
* @return true if the cursor has buffered messages ready to deliver
*/
public boolean hasMessagesBufferedToDeliver();
/**
* destroy the cursor
* @throws Exception
*
* @throws Exception
*/
public void destroy() throws Exception;
/**
* Page in a restricted number of messages
*
* @param maxItems
* @return a list of paged in messages
*/
public LinkedList pageInList(int maxItems);
}

1
activemq-core/src/main/java/org/apache/activemq/broker/region/cursors/StoreDurableSubscriberCursor.java
View File

@ -19,6 +19,7 @@ import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.Map;
import org.apache.activemq.advisory.AdvisorySupport;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.Destination;

5
activemq-core/src/main/java/org/apache/activemq/broker/region/group/MessageGroupHashBucket.java
View File

@ -80,8 +80,7 @@ public class MessageGroupHashBucket implements MessageGroupMap {
final MessageGroupSet answer = createMessageGroupSet(bucketNumber);
if (parent == null) {
return answer;
}
else {
} else {
// union the two sets together
return new MessageGroupSet() {
public boolean contains(String groupID) {
@ -103,7 +102,7 @@ public class MessageGroupHashBucket implements MessageGroupMap {
protected int getBucketNumber(String groupId) {
int bucket = groupId.hashCode() % bucketCount;
// bucket could be negative
if( bucket < 0 )
if (bucket < 0)
bucket *= -1;
return bucket;
}

5
activemq-core/src/main/java/org/apache/activemq/broker/region/group/SimpleMessageGroupMap.java
View File

@ -16,13 +16,12 @@
*/
package org.apache.activemq.broker.region.group;
import java.util.Iterator;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import org.apache.activemq.command.ConsumerId;
import java.util.Iterator;
import java.util.Map;
/**
* A simple implementation which tracks every individual GroupID value but
* which can become a memory leak if clients die before they complete a message

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

@ -32,10 +32,10 @@ public abstract class AbstractDeadLetterStrategy implements DeadLetterStrategy {
boolean result = false;
if (message != null) {
result = true;
if (message.isPersistent() == false && processNonPersistent == false) {
if (!message.isPersistent() && !processNonPersistent) {
result = false;
}
if (message.isExpired() && processExpired == false) {
if (message.isExpired() && !processExpired) {
result = false;
}
}

1
activemq-core/src/main/java/org/apache/activemq/broker/region/policy/DispatchPolicy.java
View File

@ -18,7 +18,6 @@ package org.apache.activemq.broker.region.policy;
import java.util.List;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.filter.MessageEvaluationContext;

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

@ -18,15 +18,14 @@ package org.apache.activemq.broker.region.policy;
import java.util.ArrayList;
import java.util.List;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.broker.region.Subscription;
import org.apache.activemq.broker.region.SubscriptionRecovery;
import org.apache.activemq.broker.region.Topic;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.Message;
import org.apache.activemq.filter.DestinationFilter;
import org.apache.activemq.filter.MessageEvaluationContext;
/**
* This implementation of {@link SubscriptionRecoveryPolicy} will keep a fixed

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

@ -25,7 +25,6 @@ import org.apache.activemq.command.ActiveMQTopic;
* DLQ using the subject naming hierarchy.
*
* @org.apache.xbean.XBean
*
* @version $Revision$
*/
public class IndividualDeadLetterStrategy extends AbstractDeadLetterStrategy {
@ -38,8 +37,7 @@ public class IndividualDeadLetterStrategy extends AbstractDeadLetterStrategy {
public ActiveMQDestination getDeadLetterQueueFor(ActiveMQDestination originalDestination) {
if (originalDestination.isQueue()) {
return createDestination(originalDestination, queuePrefix, useQueueForQueueMessages);
}
else {
} else {
return createDestination(originalDestination, topicPrefix, useQueueForTopicMessages);
}
}
@ -99,8 +97,7 @@ public class IndividualDeadLetterStrategy extends AbstractDeadLetterStrategy {
String name = prefix + originalDestination.getPhysicalName();
if (useQueue) {
return new ActiveMQQueue(name);
}
else {
} else {
return new ActiveMQTopic(name);
}
}

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

@ -16,11 +16,11 @@
*/
package org.apache.activemq.broker.region.policy;
import org.apache.activemq.broker.region.MessageReference;
import java.io.IOException;
import java.util.LinkedList;
import org.apache.activemq.broker.region.MessageReference;
/**
* A strategy for evicting messages from slow consumers.
*

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

@ -16,11 +16,11 @@
*/
package org.apache.activemq.broker.region.policy;
import javax.jms.MessageListener;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.Message;
import javax.jms.MessageListener;
/**
* Represents some kind of query which will load initial messages from some source for a new topic subscriber.
*

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

@ -18,7 +18,6 @@ package org.apache.activemq.broker.region.policy;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.broker.region.Subscription;
import org.apache.activemq.broker.region.SubscriptionRecovery;
import org.apache.activemq.broker.region.Topic;
import org.apache.activemq.command.ActiveMQDestination;
@ -28,12 +27,10 @@ import org.apache.activemq.command.Message;
* This SubscriptionRecoveryPolicy disable recovery of messages.
*
* @org.apache.xbean.XBean
*
* @version $Revision$
*/
public class NoSubscriptionRecoveryPolicy implements SubscriptionRecoveryPolicy {
public SubscriptionRecoveryPolicy copy() {
// This object is immutable
return this;
@ -52,7 +49,7 @@ public class NoSubscriptionRecoveryPolicy implements SubscriptionRecoveryPolicy
public void stop() throws Exception {
}
public Message[] browse(ActiveMQDestination dest) throws Exception{
public Message[] browse(ActiveMQDestination dest) throws Exception {
return new Message[0];
}

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

@ -16,10 +16,10 @@
*/
package org.apache.activemq.broker.region.policy;
import org.apache.activemq.broker.region.MessageReference;
import java.util.LinkedList;
import org.apache.activemq.broker.region.MessageReference;
/**
* An eviction strategy which evicts the oldest message first (which is the
* default).

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

@ -16,12 +16,12 @@
*/
package org.apache.activemq.broker.region.policy;
import org.apache.activemq.broker.region.MessageReference;
import java.io.IOException;
import java.util.Iterator;
import java.util.LinkedList;
import org.apache.activemq.broker.region.MessageReference;
/**
* An eviction strategy which evicts the oldest message with the lowest priority first.
*

10
activemq-core/src/main/java/org/apache/activemq/broker/region/policy/PendingQueueMessageStoragePolicy.java
View File

@ -18,20 +18,20 @@ import org.apache.activemq.broker.region.Queue;
import org.apache.activemq.broker.region.cursors.PendingMessageCursor;
import org.apache.activemq.kaha.Store;
/**
* Abstraction to allow different policies for holding messages awaiting dispatch on a Queue
* Abstraction to allow different policies for holding messages awaiting
* dispatch on a Queue
*
* @version $Revision$
*/
public interface PendingQueueMessageStoragePolicy{
public interface PendingQueueMessageStoragePolicy {
/**
* Retrieve the configured pending message storage cursor;
* @param queue
*
* @param queue
* @param tmpStore
* @return the cursor
*
*/
public PendingMessageCursor getQueuePendingMessageCursor(Queue queue, Store tmpStore);
}

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

@ -16,11 +16,11 @@
*/
package org.apache.activemq.broker.region.policy;
import java.util.List;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.filter.DestinationMap;
import java.util.List;
/**
* Represents a destination based configuration of policies so that individual
* destinations or wildcard hierarchies of destinations can be configured using

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

@ -16,22 +16,28 @@
*/
package org.apache.activemq.broker.region.policy;
import java.util.concurrent.atomic.AtomicLong;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import org.apache.activemq.ActiveMQMessageTransformation;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.Destination;
import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.broker.region.SubscriptionRecovery;
import org.apache.activemq.broker.region.Topic;
import org.apache.activemq.command.*;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQMessage;
import org.apache.activemq.command.ConnectionId;
import org.apache.activemq.command.MessageId;
import org.apache.activemq.command.ProducerId;
import org.apache.activemq.command.SessionId;
import org.apache.activemq.util.IdGenerator;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageListener;
import java.util.concurrent.atomic.AtomicLong;
/**
* This implementation of {@link SubscriptionRecoveryPolicy} will perform a user
* specific query mechanism to load any messages they may have missed.

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

@ -20,7 +20,6 @@ package org.apache.activemq.broker.region.policy;
import org.apache.activemq.Service;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.region.MessageReference;
import org.apache.activemq.broker.region.Subscription;
import org.apache.activemq.broker.region.SubscriptionRecovery;
import org.apache.activemq.broker.region.Topic;
import org.apache.activemq.command.ActiveMQDestination;

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

@ -16,10 +16,10 @@
*/
package org.apache.activemq.broker.region.virtual;
import org.apache.activemq.broker.region.Destination;
import java.util.Collection;
import org.apache.activemq.broker.region.Destination;
/**
*
* @version $Revision$

9
activemq-core/src/main/java/org/apache/activemq/broker/region/virtual/CompositeDestinationInterceptor.java
View File

@ -29,7 +29,7 @@ import org.apache.activemq.filter.MessageEvaluationContext;
/**
* Represents a composite {@link Destination} where send()s are replicated to
* each Destination instance.
*
*
* @version $Revision$
*/
public class CompositeDestinationInterceptor extends DestinationFilter {
@ -53,7 +53,7 @@ public class CompositeDestinationInterceptor extends DestinationFilter {
Object value = iter.next();
if (value instanceof FilteredDestination) {
FilteredDestination filteredDestination = (FilteredDestination) value;
FilteredDestination filteredDestination = (FilteredDestination)value;
if (messageContext == null) {
messageContext = new MessageEvaluationContext();
messageContext.setMessageReference(message);
@ -62,9 +62,8 @@ public class CompositeDestinationInterceptor extends DestinationFilter {
if (filteredDestination.matches(messageContext)) {
destination = filteredDestination.getDestination();
}
}
else if (value instanceof ActiveMQDestination) {
destination = (ActiveMQDestination) value;
} else if (value instanceof ActiveMQDestination) {
destination = (ActiveMQDestination)value;
}
if (destination == null) {
continue;

6
activemq-core/src/main/java/org/apache/activemq/broker/region/virtual/FilteredDestination.java
View File

@ -16,14 +16,14 @@
*/
package org.apache.activemq.broker.region.virtual;
import javax.jms.InvalidSelectorException;
import javax.jms.JMSException;
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.filter.BooleanExpression;
import org.apache.activemq.filter.MessageEvaluationContext;
import org.apache.activemq.selector.SelectorParser;
import javax.jms.InvalidSelectorException;
import javax.jms.JMSException;
/**
* Represents a destination which is filtered using some predicate such as a selector
* so that messages are only dispatched to the destination if they match the filter.

13
activemq-core/src/main/java/org/apache/activemq/broker/region/virtual/VirtualDestinationInterceptor.java
View File

@ -30,11 +30,9 @@ import org.apache.activemq.filter.DestinationMap;
/**
* Implements <a
* href="http://activemq.apache.org/virtual-destinations.html">Virtual
* Topics</a>.
* href="http://activemq.apache.org/virtual-destinations.html">Virtual Topics</a>.
*
* @org.apache.xbean.XBean
*
* @version $Revision$
*/
public class VirtualDestinationInterceptor implements DestinationInterceptor {
@ -46,15 +44,14 @@ public class VirtualDestinationInterceptor implements DestinationInterceptor {
Set virtualDestinations = destinationMap.get(destination.getActiveMQDestination());
List destinations = new ArrayList();
for (Iterator iter = virtualDestinations.iterator(); iter.hasNext();) {
VirtualDestination virtualDestination = (VirtualDestination) iter.next();
VirtualDestination virtualDestination = (VirtualDestination)iter.next();
Destination newNestination = virtualDestination.intercept(destination);
destinations.add(newNestination);
}
if (!destinations.isEmpty()) {
if (destinations.size() == 1) {
return (Destination) destinations.get(0);
}
else {
return (Destination)destinations.get(0);
} else {
// should rarely be used but here just in case
return createCompositeDestination(destination, destinations);
}
@ -79,7 +76,7 @@ public class VirtualDestinationInterceptor implements DestinationInterceptor {
return new DestinationFilter(destination) {
public void send(ProducerBrokerExchange context, Message messageSend) throws Exception {
for (Iterator iter = destinations.iterator(); iter.hasNext();) {
Destination destination = (Destination) iter.next();
Destination destination = (Destination)iter.next();
destination.send(context, messageSend);
}
}

39
activemq-core/src/main/java/org/apache/activemq/broker/util/CommandAgent.java
View File

@ -16,16 +16,6 @@
*/
package org.apache.activemq.broker.util;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.Service;
import org.apache.activemq.advisory.AdvisorySupport;
import org.apache.activemq.util.ServiceStopper;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.FactoryBean;
import javax.jms.Connection;
import javax.jms.ConnectionFactory;
import javax.jms.Destination;
@ -33,9 +23,19 @@ import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.Session;
import org.apache.activemq.ActiveMQConnectionFactory;
import org.apache.activemq.Service;
import org.apache.activemq.advisory.AdvisorySupport;
import org.apache.activemq.util.ServiceStopper;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.FactoryBean;
import org.springframework.beans.factory.InitializingBean;
/**
* An agent which listens to commands on a JMS destination
*
*
* @version $Revision$
* @org.apache.xbean.XBean
*/
@ -50,7 +50,6 @@ public class CommandAgent implements Service, InitializingBean, DisposableBean,
private Session session;
private MessageConsumer consumer;
public void start() throws Exception {
session = getConnection().createSession(false, Session.AUTO_ACKNOWLEDGE);
listener = new CommandMessageListener(session);
@ -68,8 +67,7 @@ public class CommandAgent implements Service, InitializingBean, DisposableBean,
try {
consumer.close();
consumer = null;
}
catch (JMSException e) {
} catch (JMSException e) {
stopper.onException(this, e);
}
}
@ -77,8 +75,7 @@ public class CommandAgent implements Service, InitializingBean, DisposableBean,
try {
session.close();
session = null;
}
catch (JMSException e) {
} catch (JMSException e) {
stopper.onException(this, e);
}
}
@ -86,15 +83,15 @@ public class CommandAgent implements Service, InitializingBean, DisposableBean,
try {
connection.close();
connection = null;
}
catch (JMSException e) {
} catch (JMSException e) {
stopper.onException(this, e);
}
}
stopper.throwFirstException();
}
// the following methods ensure that we are created on startup and the lifecycles respected
// the following methods ensure that we are created on startup and the
// lifecycles respected
// TODO there must be a simpler way?
public void afterPropertiesSet() throws Exception {
start();
@ -116,10 +113,8 @@ public class CommandAgent implements Service, InitializingBean, DisposableBean,
return true;
}
// Properties
//-------------------------------------------------------------------------
// -------------------------------------------------------------------------
public String getBrokerUrl() {
return brokerUrl;
}

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

@ -17,7 +17,6 @@
package org.apache.activemq.broker.util;
import javax.jms.TextMessage;
import javax.jms.JMSException;
/**
* Represents a processor of text based commands

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

@ -17,7 +17,6 @@
package org.apache.activemq.broker.util;
import org.apache.activemq.broker.BrokerPluginSupport;
import org.apache.activemq.broker.ConnectionContext;
import org.apache.activemq.broker.ConsumerBrokerExchange;
import org.apache.activemq.broker.ProducerBrokerExchange;
import org.apache.activemq.command.Message;

9
activemq-core/src/main/java/org/apache/activemq/broker/view/DotFileInterceptorSupport.java
View File

@ -16,14 +16,14 @@
*/
package org.apache.activemq.broker.view;
import java.io.FileWriter;
import java.io.PrintWriter;
import org.apache.activemq.broker.Broker;
import org.apache.activemq.broker.BrokerFilter;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import java.io.FileWriter;
import java.io.PrintWriter;
/**
* Useful base class
*
@ -46,8 +46,7 @@ public abstract class DotFileInterceptorSupport extends BrokerFilter {
PrintWriter writer = new PrintWriter(new FileWriter(file));
try {
generateFile(writer);
}
finally {
} finally {
writer.close();
}
}

16
activemq-core/src/main/java/org/apache/activemq/camel/CamelDestination.java
View File

@ -16,14 +16,6 @@
*/
package org.apache.activemq.camel;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQSession;
import org.apache.activemq.CustomDestination;
import org.apache.camel.CamelContext;
import org.apache.camel.CamelContextAware;
import org.apache.camel.Endpoint;
import org.apache.camel.component.jms.JmsBinding;
import javax.jms.JMSException;
import javax.jms.MessageConsumer;
import javax.jms.MessageProducer;
@ -32,6 +24,14 @@ import javax.jms.QueueSender;
import javax.jms.TopicPublisher;
import javax.jms.TopicSubscriber;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQSession;
import org.apache.activemq.CustomDestination;
import org.apache.camel.CamelContext;
import org.apache.camel.CamelContextAware;
import org.apache.camel.Endpoint;
import org.apache.camel.component.jms.JmsBinding;
/**
* @version $Revision: $
*/

40
activemq-core/src/main/java/org/apache/activemq/camel/CamelMessageConsumer.java
View File

@ -16,6 +16,12 @@
*/
package org.apache.activemq.camel;
import javax.jms.IllegalStateException;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import org.apache.activemq.ActiveMQSession;
import org.apache.activemq.util.JMSExceptionSupport;
import org.apache.camel.Consumer;
@ -24,13 +30,10 @@ import org.apache.camel.Exchange;
import org.apache.camel.PollingConsumer;
import org.apache.camel.Processor;
import javax.jms.*;
import javax.jms.IllegalStateException;
/**
* A JMS {@link javax.jms.MessageConsumer} which consumes message exchanges from a
* Camel {@link Endpoint}
*
* A JMS {@link javax.jms.MessageConsumer} which consumes message exchanges from
* a Camel {@link Endpoint}
*
* @version $Revision: $
*/
public class CamelMessageConsumer implements MessageConsumer {
@ -62,11 +65,9 @@ public class CamelMessageConsumer implements MessageConsumer {
if (pollingConsumer != null) {
pollingConsumer.stop();
}
}
catch (JMSException e) {
} catch (JMSException e) {
throw e;
}
catch (Exception e) {
} catch (Exception e) {
throw JMSExceptionSupport.create(e);
}
}
@ -99,7 +100,7 @@ public class CamelMessageConsumer implements MessageConsumer {
}
// Properties
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
public CamelDestination getDestination() {
return destination;
@ -122,7 +123,7 @@ public class CamelMessageConsumer implements MessageConsumer {
}
// Implementation methods
//-----------------------------------------------------------------------
// -----------------------------------------------------------------------
protected PollingConsumer getPollingConsumer() throws JMSException {
try {
@ -131,11 +132,9 @@ public class CamelMessageConsumer implements MessageConsumer {
pollingConsumer.start();
}
return pollingConsumer;
}
catch (JMSException e) {
} catch (JMSException e) {
throw e;
}
catch (Exception e) {
} catch (Exception e) {
throw JMSExceptionSupport.create(e);
}
}
@ -144,8 +143,7 @@ public class CamelMessageConsumer implements MessageConsumer {
if (exchange != null) {
Message message = destination.getBinding().makeJmsMessage(exchange, session);
return message;
}
else {
} else {
return null;
}
}
@ -160,11 +158,9 @@ public class CamelMessageConsumer implements MessageConsumer {
});
answer.start();
return answer;
}
catch (JMSException e) {
} catch (JMSException e) {
throw e;
}
catch (Exception e) {
} catch (Exception e) {
throw JMSExceptionSupport.create(e);
}
}

35
activemq-core/src/main/java/org/apache/activemq/camel/CamelMessageProducer.java
View File

@ -16,6 +16,11 @@
*/
package org.apache.activemq.camel;
import javax.jms.Destination;
import javax.jms.IllegalStateException;
import javax.jms.JMSException;
import javax.jms.Message;
import org.apache.activemq.ActiveMQMessageProducerSupport;
import org.apache.activemq.ActiveMQSession;
import org.apache.activemq.util.JMSExceptionSupport;
@ -24,11 +29,6 @@ import org.apache.camel.Producer;
import org.apache.camel.component.jms.JmsExchange;
import org.apache.camel.util.ObjectHelper;
import javax.jms.Destination;
import javax.jms.IllegalStateException;
import javax.jms.JMSException;
import javax.jms.Message;
/**
* A JMS {@link javax.jms.MessageProducer} which sends message exchanges to a
* Camel {@link Endpoint}
@ -47,11 +47,9 @@ public class CamelMessageProducer extends ActiveMQMessageProducerSupport {
this.endpoint = endpoint;
try {
this.producer = endpoint.createProducer();
}
catch (JMSException e) {
} catch (JMSException e) {
throw e;
}
catch (Exception e) {
} catch (Exception e) {
throw JMSExceptionSupport.create(e);
}
}
@ -69,33 +67,28 @@ public class CamelMessageProducer extends ActiveMQMessageProducerSupport {
closed = true;
try {
producer.stop();
}
catch (JMSException e) {
} catch (JMSException e) {
throw e;
}
catch (Exception e) {
} catch (Exception e) {
throw JMSExceptionSupport.create(e);
}
}
}
public void send(Destination destination, Message message, int deliveryMode, int priority, long timeToLive) throws JMSException {
CamelDestination camelDestination = null;
CamelDestination camelDestination = null;
if (ObjectHelper.equals(destination, this.destination)) {
camelDestination = this.destination;
}
else {
} else {
// TODO support any CamelDestination?
throw new IllegalArgumentException("Invalid destination setting: " + destination + " when expected: " + this.destination);
}
try {
JmsExchange exchange = new JmsExchange(endpoint.getContext(), camelDestination.getBinding(), message);
producer.process(exchange);
}
catch (JMSException e) {
} catch (JMSException e) {
throw e;
}
catch (Exception e) {
} catch (Exception e) {
throw JMSExceptionSupport.create(e);
}
}
@ -105,4 +98,4 @@ public class CamelMessageProducer extends ActiveMQMessageProducerSupport {
throw new IllegalStateException("The producer is closed");
}
}
}
}

6
activemq-core/src/main/java/org/apache/activemq/camel/CamelQueue.java
View File

@ -16,12 +16,12 @@
*/
package org.apache.activemq.camel;
import org.apache.activemq.ActiveMQSession;
import javax.jms.JMSException;
import javax.jms.Queue;
import javax.jms.QueueSender;
import javax.jms.QueueReceiver;
import javax.jms.QueueSender;
import org.apache.activemq.ActiveMQSession;
/**
* A JMS {@link Queue} object which refers to a Camel endpoint

6
activemq-core/src/main/java/org/apache/activemq/camel/CamelQueueReceiver.java
View File

@ -16,13 +16,13 @@
*/
package org.apache.activemq.camel;
import org.apache.activemq.ActiveMQSession;
import org.apache.camel.Endpoint;
import javax.jms.JMSException;
import javax.jms.Queue;
import javax.jms.QueueReceiver;
import org.apache.activemq.ActiveMQSession;
import org.apache.camel.Endpoint;
/**
* A JMS {@link javax.jms.QueueReceiver} which consumes message exchanges from a
* Camel {@link org.apache.camel.Endpoint}

4
activemq-core/src/main/java/org/apache/activemq/camel/CamelTopic.java
View File

@ -16,13 +16,13 @@
*/
package org.apache.activemq.camel;
import org.apache.activemq.ActiveMQSession;
import javax.jms.JMSException;
import javax.jms.Topic;
import javax.jms.TopicPublisher;
import javax.jms.TopicSubscriber;
import org.apache.activemq.ActiveMQSession;
/**
* A JMS {@link javax.jms.Topic} object which refers to a Camel endpoint
*

10
activemq-core/src/main/java/org/apache/activemq/camel/CamelTopicPublisher.java
View File

@ -16,14 +16,14 @@
*/
package org.apache.activemq.camel;
import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.Topic;
import javax.jms.TopicPublisher;
import org.apache.activemq.ActiveMQSession;
import org.apache.camel.Endpoint;
import javax.jms.JMSException;
import javax.jms.TopicPublisher;
import javax.jms.Topic;
import javax.jms.Message;
/**
* A JMS {@link javax.jms.TopicPublisher} which sends message exchanges to a
* Camel {@link Endpoint}

6
activemq-core/src/main/java/org/apache/activemq/camel/CamelTopicSubscriber.java
View File

@ -16,13 +16,13 @@
*/
package org.apache.activemq.camel;
import org.apache.activemq.ActiveMQSession;
import org.apache.camel.Endpoint;
import javax.jms.JMSException;
import javax.jms.Topic;
import javax.jms.TopicSubscriber;
import org.apache.activemq.ActiveMQSession;
import org.apache.camel.Endpoint;
/**
* A JMS {@link javax.jms.TopicSubscriber} which consumes message exchanges from a
* Camel {@link Endpoint}

34
activemq-core/src/main/java/org/apache/activemq/command/ActiveMQBlobMessage.java
View File

@ -16,19 +16,20 @@
*/
package org.apache.activemq.command;
import org.apache.activemq.BlobMessage;
import org.apache.activemq.blob.BlobUploader;
import org.apache.activemq.util.JMSExceptionSupport;
import javax.jms.JMSException;
import java.io.IOException;
import java.io.InputStream;
import java.net.MalformedURLException;
import java.net.URL;
import javax.jms.JMSException;
import org.apache.activemq.BlobMessage;
import org.apache.activemq.blob.BlobUploader;
import org.apache.activemq.util.JMSExceptionSupport;
/**
* An implementation of {@link BlobMessage} for out of band BLOB transfer
*
*
* @version $Revision: $
* @openwire:marshaller code="29"
*/
@ -45,7 +46,6 @@ public class ActiveMQBlobMessage extends ActiveMQMessage implements BlobMessage
private transient BlobUploader blobUploader;
private transient URL url;
public Message copy() {
ActiveMQBlobMessage copy = new ActiveMQBlobMessage();
copy(copy);
@ -76,8 +76,9 @@ public class ActiveMQBlobMessage extends ActiveMQMessage implements BlobMessage
}
/**
* The MIME type of the BLOB which can be used to apply different content types to messages.
*
* The MIME type of the BLOB which can be used to apply different content
* types to messages.
*
* @openwire:property version=3 cache=true
*/
public String getMimeType() {
@ -96,8 +97,9 @@ public class ActiveMQBlobMessage extends ActiveMQMessage implements BlobMessage
}
/**
* The name of the attachment which can be useful information if transmitting files over ActiveMQ
*
* The name of the attachment which can be useful information if
* transmitting files over ActiveMQ
*
* @openwire:property version=3 cache=false
*/
public void setName(String name) {
@ -131,8 +133,7 @@ public class ActiveMQBlobMessage extends ActiveMQMessage implements BlobMessage
if (url == null && remoteBlobUrl != null) {
try {
url = new URL(remoteBlobUrl);
}
catch (MalformedURLException e) {
} catch (MalformedURLException e) {
throw JMSExceptionSupport.create(e);
}
}
@ -144,7 +145,6 @@ public class ActiveMQBlobMessage extends ActiveMQMessage implements BlobMessage
remoteBlobUrl = url != null ? url.toExternalForm() : null;
}
public BlobUploader getBlobUploader() {
return blobUploader;
}
@ -156,13 +156,13 @@ public class ActiveMQBlobMessage extends ActiveMQMessage implements BlobMessage
public void onSend() throws JMSException {
super.onSend();
// lets ensure we upload the BLOB first out of band before we send the message
// lets ensure we upload the BLOB first out of band before we send the
// message
if (blobUploader != null) {
try {
URL value = blobUploader.upload(this);
setURL(value);
}
catch (IOException e) {
} catch (IOException e) {
throw JMSExceptionSupport.create(e);
}
}

26
activemq-core/src/main/java/org/apache/activemq/command/ActiveMQBytesMessage.java
View File

@ -16,17 +16,29 @@
*/
package org.apache.activemq.command;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.util.ByteArrayInputStream;
import org.apache.activemq.util.ByteArrayOutputStream;
import org.apache.activemq.util.*;
import javax.jms.*;
import java.io.*;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.EOFException;
import java.io.FilterOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.zip.Deflater;
import java.util.zip.DeflaterOutputStream;
import java.util.zip.InflaterInputStream;
import javax.jms.BytesMessage;
import javax.jms.JMSException;
import javax.jms.MessageFormatException;
import javax.jms.MessageNotReadableException;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.util.ByteArrayInputStream;
import org.apache.activemq.util.ByteArrayOutputStream;
import org.apache.activemq.util.ByteSequence;
import org.apache.activemq.util.ByteSequenceData;
import org.apache.activemq.util.JMSExceptionSupport;
/**
* A <CODE>BytesMessage</CODE> object is used to send a message containing a
* stream of uninterpreted bytes. It inherits from the <CODE>Message</CODE>

26
activemq-core/src/main/java/org/apache/activemq/command/ActiveMQDestination.java
View File

@ -16,9 +16,15 @@
*/
package org.apache.activemq.command;
import org.apache.activemq.jndi.JNDIBaseStorable;
import org.apache.activemq.util.IntrospectionSupport;
import org.apache.activemq.util.URISupport;
import java.io.Externalizable;
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Map;
import java.util.Properties;
import java.util.StringTokenizer;
import javax.jms.Destination;
import javax.jms.JMSException;
@ -26,16 +32,10 @@ import javax.jms.Queue;
import javax.jms.TemporaryQueue;
import javax.jms.TemporaryTopic;
import javax.jms.Topic;
import java.io.Externalizable;
import java.io.IOException;
import java.io.ObjectInput;
import java.io.ObjectOutput;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.Map;
import java.util.Properties;
import java.util.StringTokenizer;
import org.apache.activemq.jndi.JNDIBaseStorable;
import org.apache.activemq.util.IntrospectionSupport;
import org.apache.activemq.util.URISupport;
/**
* @openwire:marshaller

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

@ -142,7 +142,6 @@ public class ActiveMQMapMessage extends ActiveMQMessage implements MapMessage {
* Builds the message body from data
*
* @throws JMSException
*
* @throws IOException
*/
private void loadContent() throws JMSException {
@ -503,7 +502,7 @@ public class ActiveMQMapMessage extends ActiveMQMessage implements MapMessage {
*/
public void setBoolean(String name, boolean value) throws JMSException {
initializeWriting();
put(name, (value) ? Boolean.TRUE : Boolean.FALSE);
put(name, value ? Boolean.TRUE : Boolean.FALSE);
}
/**

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

@ -415,8 +415,11 @@ public class ActiveMQMessage extends Message implements org.apache.activemq.Mess
}
protected void checkValidObject(Object value) throws MessageFormatException {
if (!(value instanceof Boolean || value instanceof Byte || value instanceof Short || value instanceof Integer || value instanceof Long || value instanceof Float
|| value instanceof Double || value instanceof Character || value instanceof String || value == null)) {
boolean valid = value instanceof Boolean || value instanceof Byte || value instanceof Short || value instanceof Integer || value instanceof Long ;
valid = valid || value instanceof Float || value instanceof Double || value instanceof Character || value instanceof String || value == null;
if (!valid) {
ActiveMQConnection conn = getConnection();
// conn is null if we are in the broker rather than a JMS client

19
activemq-core/src/main/java/org/apache/activemq/command/ActiveMQObjectMessage.java
View File

@ -17,15 +17,6 @@
package org.apache.activemq.command;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.util.ByteArrayInputStream;
import org.apache.activemq.util.ByteArrayOutputStream;
import org.apache.activemq.util.ByteSequence;
import org.apache.activemq.util.ClassLoadingAwareObjectInputStream;
import org.apache.activemq.util.JMSExceptionSupport;
import javax.jms.JMSException;
import javax.jms.ObjectMessage;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
@ -36,6 +27,16 @@ import java.io.Serializable;
import java.util.zip.DeflaterOutputStream;
import java.util.zip.InflaterInputStream;
import javax.jms.JMSException;
import javax.jms.ObjectMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.util.ByteArrayInputStream;
import org.apache.activemq.util.ByteArrayOutputStream;
import org.apache.activemq.util.ByteSequence;
import org.apache.activemq.util.ClassLoadingAwareObjectInputStream;
import org.apache.activemq.util.JMSExceptionSupport;
/**
* An <CODE>ObjectMessage</CODE> object is used to send a message that
* contains a serializable object in the Java programming language ("Java

23
activemq-core/src/main/java/org/apache/activemq/command/ActiveMQTextMessage.java
View File

@ -16,17 +16,6 @@
*/
package org.apache.activemq.command;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.util.ByteArrayInputStream;
import org.apache.activemq.util.ByteArrayOutputStream;
import org.apache.activemq.util.ByteSequence;
import org.apache.activemq.util.JMSExceptionSupport;
import org.apache.activemq.util.MarshallingSupport;
import org.apache.activemq.wireformat.WireFormat;
import javax.jms.JMSException;
import javax.jms.MessageNotWriteableException;
import javax.jms.TextMessage;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
@ -35,6 +24,18 @@ import java.io.OutputStream;
import java.util.zip.DeflaterOutputStream;
import java.util.zip.InflaterInputStream;
import javax.jms.JMSException;
import javax.jms.MessageNotWriteableException;
import javax.jms.TextMessage;
import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.util.ByteArrayInputStream;
import org.apache.activemq.util.ByteArrayOutputStream;
import org.apache.activemq.util.ByteSequence;
import org.apache.activemq.util.JMSExceptionSupport;
import org.apache.activemq.util.MarshallingSupport;
import org.apache.activemq.wireformat.WireFormat;
/**
* @openwire:marshaller code="28"
* @version $Revision$

9
activemq-core/src/main/java/org/apache/activemq/command/RemoveSubscriptionInfo.java
View File

@ -19,19 +19,17 @@ package org.apache.activemq.command;
import org.apache.activemq.state.CommandVisitor;
/**
*
* @openwire:marshaller code="9"
* @version $Revision: 1.7 $
*/
public class RemoveSubscriptionInfo extends BaseCommand {
public static final byte DATA_STRUCTURE_TYPE=CommandTypes.REMOVE_SUBSCRIPTION_INFO;
public static final byte DATA_STRUCTURE_TYPE = CommandTypes.REMOVE_SUBSCRIPTION_INFO;
protected ConnectionId connectionId;
protected String clientId;
protected String subscriptionName;
public byte getDataStructureType() {
return DATA_STRUCTURE_TYPE;
}
@ -42,6 +40,7 @@ public class RemoveSubscriptionInfo extends BaseCommand {
public ConnectionId getConnectionId() {
return connectionId;
}
public void setConnectionId(ConnectionId connectionId) {
this.connectionId = connectionId;
}
@ -60,7 +59,7 @@ public class RemoveSubscriptionInfo extends BaseCommand {
public void setSubcriptionName(String subscriptionName) {
this.subscriptionName = subscriptionName;
}
public String getSubscriptionName() {
return subscriptionName;
}
@ -81,7 +80,7 @@ public class RemoveSubscriptionInfo extends BaseCommand {
}
public Response visit(CommandVisitor visitor) throws Exception {
return visitor.processRemoveSubscription( this );
return visitor.processRemoveSubscription(this);
}
}

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

@ -23,10 +23,10 @@ import org.apache.activemq.state.CommandVisitor;
* @version $Revision: 1.6 $
*/
public class Response extends BaseCommand {
public static final byte DATA_STRUCTURE_TYPE=CommandTypes.RESPONSE;
public static final byte DATA_STRUCTURE_TYPE = CommandTypes.RESPONSE;
int correlationId;
public byte getDataStructureType() {
return DATA_STRUCTURE_TYPE;
}
@ -37,11 +37,11 @@ public class Response extends BaseCommand {
public int getCorrelationId() {
return correlationId;
}
public void setCorrelationId(int responseId) {
this.correlationId = responseId;
}
public boolean isResponse() {
return true;
}

6
activemq-core/src/main/java/org/apache/activemq/filter/CompositeDestinationFilter.java
View File

@ -20,11 +20,11 @@ import org.apache.activemq.command.ActiveMQDestination;
/**
* A {@link DestinationFilter} used for composite destinations
*
*
* @version $Revision: 1.3 $
*/
public class CompositeDestinationFilter extends DestinationFilter {
private DestinationFilter filters[];
public CompositeDestinationFilter(ActiveMQDestination destination) {
@ -32,7 +32,7 @@ public class CompositeDestinationFilter extends DestinationFilter {
filters = new DestinationFilter[destinations.length];
for (int i = 0; i < destinations.length; i++) {
ActiveMQDestination childDestination = destinations[i];
filters[i]= DestinationFilter.parseFilter(childDestination);
filters[i] = DestinationFilter.parseFilter(childDestination);
}
}

28
activemq-core/src/main/java/org/apache/activemq/filter/DestinationMap.java
View File

@ -51,8 +51,7 @@ public class DestinationMap {
* or composite destinations this will typically be a List of matching
* values.
*
* @param key
* the destination to lookup
* @param key the destination to lookup
* @return a List of matching values or an empty list if there are no
* matching values.
*/
@ -64,9 +63,8 @@ public class DestinationMap {
ActiveMQDestination childDestination = destinations[i];
Object value = get(childDestination);
if (value instanceof Set) {
answer.addAll((Set) value);
}
else if (value != null) {
answer.addAll((Set)value);
} else if (value != null) {
answer.add(value);
}
}
@ -108,7 +106,7 @@ public class DestinationMap {
public int getTopicRootChildCount() {
return topicRootNode.getChildCount();
}
public int getQueueRootChildCount() {
return queueRootNode.getChildCount();
}
@ -121,7 +119,6 @@ public class DestinationMap {
return topicRootNode;
}
// Implementation methods
// -------------------------------------------------------------------------
@ -131,13 +128,12 @@ public class DestinationMap {
*/
protected void setEntries(List entries) {
for (Iterator iter = entries.iterator(); iter.hasNext();) {
Object element = (Object) iter.next();
Object element = (Object)iter.next();
Class type = getEntryClass();
if (type.isInstance(element)) {
DestinationMapEntry entry = (DestinationMapEntry) element;
DestinationMapEntry entry = (DestinationMapEntry)element;
put(entry.getDestination(), entry.getValue());
}
else {
} else {
throw new IllegalArgumentException("Each entry must be an instance of type: " + type.getName() + " but was: " + element);
}
}
@ -162,14 +158,14 @@ public class DestinationMap {
/**
* @param key
* @return
* @return
*/
public Set removeAll(ActiveMQDestination key) {
Set rc = new HashSet();
if (key.isComposite()) {
ActiveMQDestination[] destinations = key.getCompositeDestinations();
for (int i = 0; i < destinations.length; i++) {
rc.add( removeAll(destinations[i]) );
rc.add(removeAll(destinations[i]));
}
return rc;
}
@ -183,8 +179,7 @@ public class DestinationMap {
* no matching value. If there are multiple values, the results are sorted
* and the last item (the biggest) is returned.
*
* @param destination
* the destination to find the value for
* @param destination the destination to find the value for
* @return the largest matching value or null if no value matches
*/
public Object chooseValue(ActiveMQDestination destination) {
@ -202,8 +197,7 @@ public class DestinationMap {
protected DestinationMapNode getRootNode(ActiveMQDestination key) {
if (key.isQueue()) {
return queueRootNode;
}
else {
} else {
return topicRootNode;
}
}

9
activemq-core/src/main/java/org/apache/activemq/filter/DestinationMapEntry.java
View File

@ -31,16 +31,13 @@ public abstract class DestinationMapEntry implements InitializingBean, Comparabl
private ActiveMQDestination destination;
public int compareTo(Object that) {
if (that instanceof DestinationMapEntry) {
DestinationMapEntry thatEntry = (DestinationMapEntry) that;
DestinationMapEntry thatEntry = (DestinationMapEntry)that;
return ActiveMQDestination.compare(destination, thatEntry.destination);
}
else if (that == null) {
} else if (that == null) {
return 1;
}
else {
} else {
return getClass().getName().compareTo(that.getClass().getName());
}
}

Some files were not shown because too many files have changed in this diff Show More