You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ch...@apache.org on 2007/08/08 20:58:13 UTC
svn commit: r563982 [15/32] - in /activemq/trunk/activemq-core/src:
main/java/org/apache/activemq/ main/java/org/apache/activemq/advisory/
main/java/org/apache/activemq/blob/ main/java/org/apache/activemq/broker/
main/java/org/apache/activemq/broker/jm...
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/ConnectionStateTracker.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/ConnectionStateTracker.java?view=diff&rev=563982&r1=563981&r2=563982
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/ConnectionStateTracker.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/ConnectionStateTracker.java Wed Aug 8 11:56:59 2007
@@ -38,34 +38,36 @@
import org.apache.activemq.util.IOExceptionSupport;
/**
- * Tracks the state of a connection so a newly established transport can
- * be re-initialized to the state that was tracked.
+ * Tracks the state of a connection so a newly established transport can be
+ * re-initialized to the state that was tracked.
*
* @version $Revision$
*/
public class ConnectionStateTracker extends CommandVisitorAdapter {
- private final static Tracked TRACKED_RESPONSE_MARKER = new Tracked(null);
-
- private boolean trackTransactions = false;
-
- private boolean restoreSessions=true;
- private boolean restoreConsumers=true;
- private boolean restoreProducers=true;
- private boolean restoreTransaction=true;
-
+ private final static Tracked TRACKED_RESPONSE_MARKER = new Tracked(null);
+
+ private boolean trackTransactions = false;
+
+ private boolean restoreSessions = true;
+ private boolean restoreConsumers = true;
+ private boolean restoreProducers = true;
+ private boolean restoreTransaction = true;
+
protected final ConcurrentHashMap connectionStates = new ConcurrentHashMap();
-
+
private class RemoveTransactionAction implements Runnable {
- private final TransactionInfo info;
- public RemoveTransactionAction(TransactionInfo info) {
- this.info = info;
- }
- public void run() {
- ConnectionId connectionId = info.getConnectionId();
- ConnectionState cs = (ConnectionState) connectionStates.get(connectionId);
- cs.removeTransactionState(info.getTransactionId());
- }
+ private final TransactionInfo info;
+
+ public RemoveTransactionAction(TransactionInfo info) {
+ this.info = info;
+ }
+
+ public void run() {
+ ConnectionId connectionId = info.getConnectionId();
+ ConnectionState cs = (ConnectionState)connectionStates.get(connectionId);
+ cs.removeTransactionState(info.getTransactionId());
+ }
}
/**
@@ -77,40 +79,40 @@
*/
public Tracked track(Command command) throws IOException {
try {
- return (Tracked) command.visit(this);
+ return (Tracked)command.visit(this);
} catch (IOException e) {
throw e;
} catch (Throwable e) {
throw IOExceptionSupport.create(e);
}
- }
-
- public void restore( Transport transport ) throws IOException {
+ }
+
+ public void restore(Transport transport) throws IOException {
// Restore the connections.
for (Iterator iter = connectionStates.values().iterator(); iter.hasNext();) {
- ConnectionState connectionState = (ConnectionState) iter.next();
+ ConnectionState connectionState = (ConnectionState)iter.next();
transport.oneway(connectionState.getInfo());
restoreTempDestinations(transport, connectionState);
-
- if( restoreSessions )
+
+ if (restoreSessions)
restoreSessions(transport, connectionState);
-
- if( restoreTransaction )
- restoreTransactions(transport, connectionState);
+
+ if (restoreTransaction)
+ restoreTransactions(transport, connectionState);
}
}
private void restoreTransactions(Transport transport, ConnectionState connectionState) throws IOException {
- for (Iterator iter = connectionState.getTransactionStates().iterator(); iter.hasNext();) {
- TransactionState transactionState = (TransactionState) iter.next();
- for (Iterator iterator = transactionState.getCommands().iterator(); iterator.hasNext();) {
- Command command = (Command) iterator.next();
- transport.oneway(command);
- }
- }
- }
+ for (Iterator iter = connectionState.getTransactionStates().iterator(); iter.hasNext();) {
+ TransactionState transactionState = (TransactionState)iter.next();
+ for (Iterator iterator = transactionState.getCommands().iterator(); iterator.hasNext();) {
+ Command command = (Command)iterator.next();
+ transport.oneway(command);
+ }
+ }
+ }
- /**
+ /**
* @param transport
* @param connectionState
* @throws IOException
@@ -118,13 +120,13 @@
protected void restoreSessions(Transport transport, ConnectionState connectionState) throws IOException {
// Restore the connection's sessions
for (Iterator iter2 = connectionState.getSessionStates().iterator(); iter2.hasNext();) {
- SessionState sessionState = (SessionState) iter2.next();
+ SessionState sessionState = (SessionState)iter2.next();
transport.oneway(sessionState.getInfo());
- if( restoreProducers )
+ if (restoreProducers)
restoreProducers(transport, sessionState);
-
- if( restoreConsumers )
+
+ if (restoreConsumers)
restoreConsumers(transport, sessionState);
}
}
@@ -137,7 +139,7 @@
protected void restoreConsumers(Transport transport, SessionState sessionState) throws IOException {
// Restore the session's consumers
for (Iterator iter3 = sessionState.getConsumerStates().iterator(); iter3.hasNext();) {
- ConsumerState consumerState = (ConsumerState) iter3.next();
+ ConsumerState consumerState = (ConsumerState)iter3.next();
transport.oneway(consumerState.getInfo());
}
}
@@ -150,7 +152,7 @@
protected void restoreProducers(Transport transport, SessionState sessionState) throws IOException {
// Restore the session's producers
for (Iterator iter3 = sessionState.getProducerStates().iterator(); iter3.hasNext();) {
- ProducerState producerState = (ProducerState) iter3.next();
+ ProducerState producerState = (ProducerState)iter3.next();
transport.oneway(producerState.getInfo());
}
}
@@ -160,44 +162,44 @@
* @param connectionState
* @throws IOException
*/
- protected void restoreTempDestinations(Transport transport, ConnectionState connectionState) throws IOException {
+ protected void restoreTempDestinations(Transport transport, ConnectionState connectionState)
+ throws IOException {
// Restore the connection's temp destinations.
for (Iterator iter2 = connectionState.getTempDesinations().iterator(); iter2.hasNext();) {
- transport.oneway((DestinationInfo) iter2.next());
+ transport.oneway((DestinationInfo)iter2.next());
}
}
- public Response processAddDestination(DestinationInfo info){
- if(info!=null){
- ConnectionState cs=(ConnectionState)connectionStates.get(info.getConnectionId());
- if(cs!=null&&info.getDestination().isTemporary()){
+ public Response processAddDestination(DestinationInfo info) {
+ if (info != null) {
+ ConnectionState cs = (ConnectionState)connectionStates.get(info.getConnectionId());
+ if (cs != null && info.getDestination().isTemporary()) {
cs.addTempDestination(info);
}
}
return TRACKED_RESPONSE_MARKER;
}
- public Response processRemoveDestination(DestinationInfo info){
- if(info!=null){
- ConnectionState cs=(ConnectionState)connectionStates.get(info.getConnectionId());
- if(cs!=null&&info.getDestination().isTemporary()){
+ public Response processRemoveDestination(DestinationInfo info) {
+ if (info != null) {
+ ConnectionState cs = (ConnectionState)connectionStates.get(info.getConnectionId());
+ if (cs != null && info.getDestination().isTemporary()) {
cs.removeTempDestination(info.getDestination());
}
}
return TRACKED_RESPONSE_MARKER;
}
-
- public Response processAddProducer(ProducerInfo info){
- if(info!=null&&info.getProducerId()!=null){
- SessionId sessionId=info.getProducerId().getParentId();
- if(sessionId!=null){
- ConnectionId connectionId=sessionId.getParentId();
- if(connectionId!=null){
- ConnectionState cs=(ConnectionState)connectionStates.get(connectionId);
- if(cs!=null){
- SessionState ss=cs.getSessionState(sessionId);
- if(ss!=null){
+ public Response processAddProducer(ProducerInfo info) {
+ if (info != null && info.getProducerId() != null) {
+ SessionId sessionId = info.getProducerId().getParentId();
+ if (sessionId != null) {
+ ConnectionId connectionId = sessionId.getParentId();
+ if (connectionId != null) {
+ ConnectionState cs = (ConnectionState)connectionStates.get(connectionId);
+ if (cs != null) {
+ SessionState ss = cs.getSessionState(sessionId);
+ if (ss != null) {
ss.addProducer(info);
}
}
@@ -206,17 +208,17 @@
}
return TRACKED_RESPONSE_MARKER;
}
-
- public Response processRemoveProducer(ProducerId id){
- if(id!=null){
- SessionId sessionId=id.getParentId();
- if(sessionId!=null){
- ConnectionId connectionId=sessionId.getParentId();
- if(connectionId!=null){
- ConnectionState cs=(ConnectionState)connectionStates.get(connectionId);
- if(cs!=null){
- SessionState ss=cs.getSessionState(sessionId);
- if(ss!=null){
+
+ public Response processRemoveProducer(ProducerId id) {
+ if (id != null) {
+ SessionId sessionId = id.getParentId();
+ if (sessionId != null) {
+ ConnectionId connectionId = sessionId.getParentId();
+ if (connectionId != null) {
+ ConnectionState cs = (ConnectionState)connectionStates.get(connectionId);
+ if (cs != null) {
+ SessionState ss = cs.getSessionState(sessionId);
+ if (ss != null) {
ss.removeProducer(id);
}
}
@@ -226,16 +228,16 @@
return TRACKED_RESPONSE_MARKER;
}
- public Response processAddConsumer(ConsumerInfo info){
- if(info!=null){
- SessionId sessionId=info.getConsumerId().getParentId();
- if(sessionId!=null){
- ConnectionId connectionId=sessionId.getParentId();
- if(connectionId!=null){
- ConnectionState cs=(ConnectionState)connectionStates.get(connectionId);
- if(cs!=null){
- SessionState ss=cs.getSessionState(sessionId);
- if(ss!=null){
+ public Response processAddConsumer(ConsumerInfo info) {
+ if (info != null) {
+ SessionId sessionId = info.getConsumerId().getParentId();
+ if (sessionId != null) {
+ ConnectionId connectionId = sessionId.getParentId();
+ if (connectionId != null) {
+ ConnectionState cs = (ConnectionState)connectionStates.get(connectionId);
+ if (cs != null) {
+ SessionState ss = cs.getSessionState(sessionId);
+ if (ss != null) {
ss.addConsumer(info);
}
}
@@ -244,17 +246,17 @@
}
return TRACKED_RESPONSE_MARKER;
}
-
- public Response processRemoveConsumer(ConsumerId id){
- if(id!=null){
- SessionId sessionId=id.getParentId();
- if(sessionId!=null){
- ConnectionId connectionId=sessionId.getParentId();
- if(connectionId!=null){
- ConnectionState cs=(ConnectionState)connectionStates.get(connectionId);
- if(cs!=null){
- SessionState ss=cs.getSessionState(sessionId);
- if(ss!=null){
+
+ public Response processRemoveConsumer(ConsumerId id) {
+ if (id != null) {
+ SessionId sessionId = id.getParentId();
+ if (sessionId != null) {
+ ConnectionId connectionId = sessionId.getParentId();
+ if (connectionId != null) {
+ ConnectionState cs = (ConnectionState)connectionStates.get(connectionId);
+ if (cs != null) {
+ SessionState ss = cs.getSessionState(sessionId);
+ if (ss != null) {
ss.removeConsumer(id);
}
}
@@ -263,56 +265,55 @@
}
return TRACKED_RESPONSE_MARKER;
}
-
- public Response processAddSession(SessionInfo info){
- if(info!=null){
- ConnectionId connectionId=info.getSessionId().getParentId();
- if(connectionId!=null){
- ConnectionState cs=(ConnectionState)connectionStates.get(connectionId);
- if(cs!=null){
+
+ public Response processAddSession(SessionInfo info) {
+ if (info != null) {
+ ConnectionId connectionId = info.getSessionId().getParentId();
+ if (connectionId != null) {
+ ConnectionState cs = (ConnectionState)connectionStates.get(connectionId);
+ if (cs != null) {
cs.addSession(info);
}
}
}
return TRACKED_RESPONSE_MARKER;
}
-
- public Response processRemoveSession(SessionId id){
- if(id!=null){
- ConnectionId connectionId=id.getParentId();
- if(connectionId!=null){
- ConnectionState cs=(ConnectionState)connectionStates.get(connectionId);
- if(cs!=null){
+
+ public Response processRemoveSession(SessionId id) {
+ if (id != null) {
+ ConnectionId connectionId = id.getParentId();
+ if (connectionId != null) {
+ ConnectionState cs = (ConnectionState)connectionStates.get(connectionId);
+ if (cs != null) {
cs.removeSession(id);
}
}
}
return TRACKED_RESPONSE_MARKER;
}
-
- public Response processAddConnection(ConnectionInfo info){
+
+ public Response processAddConnection(ConnectionInfo info) {
if (info != null) {
- connectionStates.put(info.getConnectionId(), new ConnectionState(info));
+ connectionStates.put(info.getConnectionId(), new ConnectionState(info));
}
return TRACKED_RESPONSE_MARKER;
}
-
+
public Response processRemoveConnection(ConnectionId id) throws Exception {
if (id != null) {
- connectionStates.remove(id);
+ connectionStates.remove(id);
}
return TRACKED_RESPONSE_MARKER;
}
-
- public Response processMessage(Message send) throws Exception{
- if(trackTransactions&&send!=null&&send.getTransactionId()!=null){
- ConnectionId connectionId=send.getProducerId().getParentId().getParentId();
- if(connectionId!=null){
- ConnectionState cs=(ConnectionState)connectionStates.get(connectionId);
- if(cs!=null){
- TransactionState transactionState=cs.getTransactionState(send.getTransactionId());
- if(transactionState!=null){
+ public Response processMessage(Message send) throws Exception {
+ if (trackTransactions && send != null && send.getTransactionId() != null) {
+ ConnectionId connectionId = send.getProducerId().getParentId().getParentId();
+ if (connectionId != null) {
+ ConnectionState cs = (ConnectionState)connectionStates.get(connectionId);
+ if (cs != null) {
+ TransactionState transactionState = cs.getTransactionState(send.getTransactionId());
+ if (transactionState != null) {
transactionState.addCommand(send);
}
}
@@ -320,16 +321,16 @@
return TRACKED_RESPONSE_MARKER;
}
return null;
- }
-
- public Response processMessageAck(MessageAck ack){
- if(trackTransactions&&ack!=null&&ack.getTransactionId()!=null){
- ConnectionId connectionId=ack.getConsumerId().getParentId().getParentId();
- if(connectionId!=null){
- ConnectionState cs=(ConnectionState)connectionStates.get(connectionId);
- if(cs!=null){
- TransactionState transactionState=cs.getTransactionState(ack.getTransactionId());
- if(transactionState!=null){
+ }
+
+ public Response processMessageAck(MessageAck ack) {
+ if (trackTransactions && ack != null && ack.getTransactionId() != null) {
+ ConnectionId connectionId = ack.getConsumerId().getParentId().getParentId();
+ if (connectionId != null) {
+ ConnectionState cs = (ConnectionState)connectionStates.get(connectionId);
+ if (cs != null) {
+ TransactionState transactionState = cs.getTransactionState(ack.getTransactionId());
+ if (transactionState != null) {
transactionState.addCommand(ack);
}
}
@@ -338,29 +339,29 @@
}
return null;
}
-
- public Response processBeginTransaction(TransactionInfo info){
- if(trackTransactions&&info!=null && info.getTransactionId() != null){
- ConnectionId connectionId=info.getConnectionId();
- if(connectionId!=null){
- ConnectionState cs=(ConnectionState)connectionStates.get(connectionId);
- if(cs!=null){
+
+ public Response processBeginTransaction(TransactionInfo info) {
+ if (trackTransactions && info != null && info.getTransactionId() != null) {
+ ConnectionId connectionId = info.getConnectionId();
+ if (connectionId != null) {
+ ConnectionState cs = (ConnectionState)connectionStates.get(connectionId);
+ if (cs != null) {
cs.addTransactionState(info.getTransactionId());
}
}
return TRACKED_RESPONSE_MARKER;
}
return null;
- }
-
- public Response processPrepareTransaction(TransactionInfo info) throws Exception{
- if(trackTransactions&&info!=null){
- ConnectionId connectionId=info.getConnectionId();
- if(connectionId!=null){
- ConnectionState cs=(ConnectionState)connectionStates.get(connectionId);
- if(cs!=null){
- TransactionState transactionState=cs.getTransactionState(info.getTransactionId());
- if(transactionState!=null){
+ }
+
+ public Response processPrepareTransaction(TransactionInfo info) throws Exception {
+ if (trackTransactions && info != null) {
+ ConnectionId connectionId = info.getConnectionId();
+ if (connectionId != null) {
+ ConnectionState cs = (ConnectionState)connectionStates.get(connectionId);
+ if (cs != null) {
+ TransactionState transactionState = cs.getTransactionState(info.getTransactionId());
+ if (transactionState != null) {
transactionState.addCommand(info);
}
}
@@ -369,15 +370,15 @@
}
return null;
}
-
- public Response processCommitTransactionOnePhase(TransactionInfo info) throws Exception{
- if(trackTransactions&&info!=null){
- ConnectionId connectionId=info.getConnectionId();
- if(connectionId!=null){
- ConnectionState cs=(ConnectionState)connectionStates.get(connectionId);
- if(cs!=null){
- TransactionState transactionState=cs.getTransactionState(info.getTransactionId());
- if(transactionState!=null){
+
+ public Response processCommitTransactionOnePhase(TransactionInfo info) throws Exception {
+ if (trackTransactions && info != null) {
+ ConnectionId connectionId = info.getConnectionId();
+ if (connectionId != null) {
+ ConnectionState cs = (ConnectionState)connectionStates.get(connectionId);
+ if (cs != null) {
+ TransactionState transactionState = cs.getTransactionState(info.getTransactionId());
+ if (transactionState != null) {
transactionState.addCommand(info);
return new Tracked(new RemoveTransactionAction(info));
}
@@ -385,16 +386,16 @@
}
}
return null;
- }
-
- public Response processCommitTransactionTwoPhase(TransactionInfo info) throws Exception{
- if(trackTransactions&&info!=null){
- ConnectionId connectionId=info.getConnectionId();
- if(connectionId!=null){
- ConnectionState cs=(ConnectionState)connectionStates.get(connectionId);
- if(cs!=null){
- TransactionState transactionState=cs.getTransactionState(info.getTransactionId());
- if(transactionState!=null){
+ }
+
+ public Response processCommitTransactionTwoPhase(TransactionInfo info) throws Exception {
+ if (trackTransactions && info != null) {
+ ConnectionId connectionId = info.getConnectionId();
+ if (connectionId != null) {
+ ConnectionState cs = (ConnectionState)connectionStates.get(connectionId);
+ if (cs != null) {
+ TransactionState transactionState = cs.getTransactionState(info.getTransactionId());
+ if (transactionState != null) {
transactionState.addCommand(info);
return new Tracked(new RemoveTransactionAction(info));
}
@@ -403,15 +404,15 @@
}
return null;
}
-
- public Response processRollbackTransaction(TransactionInfo info) throws Exception{
- if(trackTransactions&&info!=null){
- ConnectionId connectionId=info.getConnectionId();
- if(connectionId!=null){
- ConnectionState cs=(ConnectionState)connectionStates.get(connectionId);
- if(cs!=null){
- TransactionState transactionState=cs.getTransactionState(info.getTransactionId());
- if(transactionState!=null){
+
+ public Response processRollbackTransaction(TransactionInfo info) throws Exception {
+ if (trackTransactions && info != null) {
+ ConnectionId connectionId = info.getConnectionId();
+ if (connectionId != null) {
+ ConnectionState cs = (ConnectionState)connectionStates.get(connectionId);
+ if (cs != null) {
+ TransactionState transactionState = cs.getTransactionState(info.getTransactionId());
+ if (transactionState != null) {
transactionState.addCommand(info);
return new Tracked(new RemoveTransactionAction(info));
}
@@ -420,15 +421,15 @@
}
return null;
}
-
- public Response processEndTransaction(TransactionInfo info) throws Exception{
- if(trackTransactions&&info!=null){
- ConnectionId connectionId=info.getConnectionId();
- if(connectionId!=null){
- ConnectionState cs=(ConnectionState)connectionStates.get(connectionId);
- if(cs!=null){
- TransactionState transactionState=cs.getTransactionState(info.getTransactionId());
- if(transactionState!=null){
+
+ public Response processEndTransaction(TransactionInfo info) throws Exception {
+ if (trackTransactions && info != null) {
+ ConnectionId connectionId = info.getConnectionId();
+ if (connectionId != null) {
+ ConnectionState cs = (ConnectionState)connectionStates.get(connectionId);
+ if (cs != null) {
+ TransactionState transactionState = cs.getTransactionState(info.getTransactionId());
+ if (transactionState != null) {
transactionState.addCommand(info);
}
}
@@ -437,7 +438,7 @@
}
return null;
}
-
+
public boolean isRestoreConsumers() {
return restoreConsumers;
}
@@ -462,20 +463,20 @@
this.restoreSessions = restoreSessions;
}
- public boolean isTrackTransactions() {
- return trackTransactions;
- }
-
- public void setTrackTransactions(boolean trackTransactions) {
- this.trackTransactions = trackTransactions;
- }
-
- public boolean isRestoreTransaction() {
- return restoreTransaction;
- }
-
- public void setRestoreTransaction(boolean restoreTransaction) {
- this.restoreTransaction = restoreTransaction;
- }
+ public boolean isTrackTransactions() {
+ return trackTransactions;
+ }
+
+ public void setTrackTransactions(boolean trackTransactions) {
+ this.trackTransactions = trackTransactions;
+ }
+
+ public boolean isRestoreTransaction() {
+ return restoreTransaction;
+ }
+
+ public void setRestoreTransaction(boolean restoreTransaction) {
+ this.restoreTransaction = restoreTransaction;
+ }
}
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/ProducerState.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/ProducerState.java?view=diff&rev=563982&r1=563981&r2=563982
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/ProducerState.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/ProducerState.java Wed Aug 8 11:56:59 2007
@@ -19,18 +19,20 @@
import org.apache.activemq.command.ProducerInfo;
-public class ProducerState {
+public class ProducerState {
final ProducerInfo info;
- private long lastSequenceId=-1;
-
+ private long lastSequenceId = -1;
+
public ProducerState(ProducerInfo info) {
this.info = info;
- }
+ }
+
public String toString() {
return info.toString();
}
+
public ProducerInfo getInfo() {
return info;
}
-
+
}
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/SessionState.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/SessionState.java?view=diff&rev=563982&r1=563981&r2=563982
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/SessionState.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/SessionState.java Wed Aug 8 11:56:59 2007
@@ -29,68 +29,74 @@
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
-public class SessionState {
+public class SessionState {
final SessionInfo info;
-
+
public final ConcurrentHashMap producers = new ConcurrentHashMap();
public final ConcurrentHashMap consumers = new ConcurrentHashMap();
private final AtomicBoolean shutdown = new AtomicBoolean(false);
-
+
public SessionState(SessionInfo info) {
this.info = info;
- }
+ }
+
public String toString() {
return info.toString();
}
-
+
public void addProducer(ProducerInfo info) {
- checkShutdown();
- producers.put(info.getProducerId(), new ProducerState(info));
- }
+ checkShutdown();
+ producers.put(info.getProducerId(), new ProducerState(info));
+ }
+
public ProducerState removeProducer(ProducerId id) {
- return (ProducerState) producers.remove(id);
+ return (ProducerState)producers.remove(id);
}
-
+
public void addConsumer(ConsumerInfo info) {
- checkShutdown();
- consumers.put(info.getConsumerId(), new ConsumerState(info));
- }
+ checkShutdown();
+ consumers.put(info.getConsumerId(), new ConsumerState(info));
+ }
+
public ConsumerState removeConsumer(ConsumerId id) {
- return (ConsumerState) consumers.remove(id);
+ return (ConsumerState)consumers.remove(id);
}
-
+
public SessionInfo getInfo() {
return info;
}
-
+
public Set getConsumerIds() {
return consumers.keySet();
- }
+ }
+
public Set getProducerIds() {
return producers.keySet();
- }
+ }
+
public Collection getProducerStates() {
return producers.values();
}
- public ProducerState getProducerState(ProducerId producerId) {
- return (ProducerState) producers.get(producerId);
- }
-
+
+ public ProducerState getProducerState(ProducerId producerId) {
+ return (ProducerState)producers.get(producerId);
+ }
+
public Collection getConsumerStates() {
return consumers.values();
}
-
+
public ConsumerState getConsumerState(ConsumerId consumerId) {
return (ConsumerState)consumers.get(consumerId);
}
-
+
private void checkShutdown() {
- if( shutdown.get() )
- throw new IllegalStateException("Disposed");
- }
-
+ if (shutdown.get())
+ throw new IllegalStateException("Disposed");
+ }
+
public void shutdown() {
- shutdown.set(false);
+ shutdown.set(false);
}
}
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/Tracked.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/Tracked.java?view=diff&rev=563982&r1=563981&r2=563982
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/Tracked.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/Tracked.java Wed Aug 8 11:56:59 2007
@@ -19,22 +19,22 @@
import org.apache.activemq.command.Response;
public class Tracked extends Response {
-
- private Runnable runnable;
-
- public Tracked(Runnable runnable) {
- this.runnable = runnable;
- }
-
- public void onResponses() {
- if( runnable != null ) {
- runnable.run();
- runnable=null;
- }
- }
-
- public boolean isWaitingForResponse() {
- return runnable!=null;
- }
-
+
+ private Runnable runnable;
+
+ public Tracked(Runnable runnable) {
+ this.runnable = runnable;
+ }
+
+ public void onResponses() {
+ if (runnable != null) {
+ runnable.run();
+ runnable = null;
+ }
+ }
+
+ public boolean isWaitingForResponse() {
+ return runnable != null;
+ }
+
}
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/TransactionState.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/TransactionState.java?view=diff&rev=563982&r1=563981&r2=563982
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/TransactionState.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/state/TransactionState.java Wed Aug 8 11:56:59 2007
@@ -18,61 +18,65 @@
import java.util.ArrayList;
import java.util.List;
+import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.activemq.command.Command;
import org.apache.activemq.command.TransactionId;
-import java.util.concurrent.atomic.AtomicBoolean;
-
-public class TransactionState {
+public class TransactionState {
final TransactionId id;
-
+
public final ArrayList commands = new ArrayList();
private final AtomicBoolean shutdown = new AtomicBoolean(false);
- private boolean prepared;
+ private boolean prepared;
+
+ private int preparedResult;
- private int preparedResult;
-
public TransactionState(TransactionId id) {
this.id = id;
- }
+ }
+
public String toString() {
return id.toString();
}
-
+
public void addCommand(Command operation) {
- checkShutdown();
- commands.add(operation);
- }
+ checkShutdown();
+ commands.add(operation);
+ }
public List getCommands() {
- return commands;
- }
-
+ return commands;
+ }
+
private void checkShutdown() {
- if( shutdown.get() )
- throw new IllegalStateException("Disposed");
- }
-
+ if (shutdown.get())
+ throw new IllegalStateException("Disposed");
+ }
+
public void shutdown() {
- shutdown.set(false);
+ shutdown.set(false);
+ }
+
+ public TransactionId getId() {
+ return id;
+ }
+
+ public void setPrepared(boolean prepared) {
+ this.prepared = prepared;
+ }
+
+ public boolean isPrepared() {
+ return prepared;
+ }
+
+ public void setPreparedResult(int preparedResult) {
+ this.preparedResult = preparedResult;
+ }
+
+ public int getPreparedResult() {
+ return preparedResult;
}
- public TransactionId getId() {
- return id;
- }
-
- public void setPrepared(boolean prepared) {
- this.prepared = prepared;
- }
- public boolean isPrepared() {
- return prepared;
- }
- public void setPreparedResult(int preparedResult) {
- this.preparedResult = preparedResult;
- }
- public int getPreparedResult() {
- return preparedResult;
- }
}
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/MessageStore.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/MessageStore.java?view=diff&rev=563982&r1=563981&r2=563982
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/MessageStore.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/MessageStore.java Wed Aug 8 11:56:59 2007
@@ -24,47 +24,48 @@
import org.apache.activemq.memory.UsageManager;
/**
- * Represents a message store which is used by the persistent
- * implementations
+ * Represents a message store which is used by the persistent implementations
*
* @version $Revision: 1.5 $
*/
-public interface MessageStore extends Service{
+public interface MessageStore extends Service {
/**
* Adds a message to the message store
*
* @param context context
- * @param message
- * @throws IOException
+ * @param message
+ * @throws IOException
*/
- public void addMessage(ConnectionContext context,Message message) throws IOException;
+ public void addMessage(ConnectionContext context, Message message) throws IOException;
/**
- * Looks up a message using either the String messageID or the messageNumber. Implementations are encouraged to fill
- * in the missing key if its easy to do so.
+ * Looks up a message using either the String messageID or the
+ * messageNumber. Implementations are encouraged to fill in the missing key
+ * if its easy to do so.
*
* @param identity which contains either the messageID or the messageNumber
* @return the message or null if it does not exist
- * @throws IOException
+ * @throws IOException
*/
public Message getMessage(MessageId identity) throws IOException;
/**
* Removes a message from the message store.
*
- * @param context
- * @param ack the ack request that cause the message to be removed. It conatins the identity which contains the
- * messageID of the message that needs to be removed.
- * @throws IOException
+ * @param context
+ * @param ack the ack request that cause the message to be removed. It
+ * conatins the identity which contains the messageID of the
+ * message that needs to be removed.
+ * @throws IOException
*/
- public void removeMessage(ConnectionContext context,MessageAck ack) throws IOException;
+ public void removeMessage(ConnectionContext context, MessageAck ack) throws IOException;
/**
* Removes all the messages from the message store.
*
- * @param context
- * @throws IOException
+ * @param context
+ * @throws IOException
*/
public void removeAllMessages(ConnectionContext context) throws IOException;
@@ -84,25 +85,24 @@
public ActiveMQDestination getDestination();
/**
- * @param usageManager The UsageManager that is controlling the destination's memory usage.
+ * @param usageManager The UsageManager that is controlling the
+ * destination's memory usage.
*/
public void setUsageManager(UsageManager usageManager);
/**
* @return the number of messages ready to deliver
- * @throws IOException
+ * @throws IOException
*
*/
public int getMessageCount() throws IOException;
/**
* A hint to the Store to reset any batching state for the Destination
- *
+ *
*/
public void resetBatching();
-
- public void recoverNextMessages(int maxReturned,MessageRecoveryListener listener)
- throws Exception;
-
+ public void recoverNextMessages(int maxReturned, MessageRecoveryListener listener) throws Exception;
+
}
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ProxyMessageStore.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ProxyMessageStore.java?view=diff&rev=563982&r1=563981&r2=563982
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ProxyMessageStore.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ProxyMessageStore.java Wed Aug 8 11:56:59 2007
@@ -31,11 +31,11 @@
public class ProxyMessageStore implements MessageStore {
final MessageStore delegate;
-
+
public ProxyMessageStore(MessageStore delegate) {
this.delegate = delegate;
}
-
+
public MessageStore getDelegate() {
return delegate;
}
@@ -43,24 +43,31 @@
public void addMessage(ConnectionContext context, Message message) throws IOException {
delegate.addMessage(context, message);
}
+
public Message getMessage(MessageId identity) throws IOException {
return delegate.getMessage(identity);
}
+
public void recover(MessageRecoveryListener listener) throws Exception {
delegate.recover(listener);
}
+
public void removeAllMessages(ConnectionContext context) throws IOException {
delegate.removeAllMessages(context);
}
+
public void removeMessage(ConnectionContext context, MessageAck ack) throws IOException {
delegate.removeMessage(context, ack);
}
+
public void start() throws Exception {
delegate.start();
- }
+ }
+
public void stop() throws Exception {
delegate.stop();
}
+
public ActiveMQDestination getDestination() {
return delegate.getDestination();
}
@@ -69,19 +76,17 @@
delegate.setUsageManager(usageManager);
}
-
- public int getMessageCount() throws IOException{
+ public int getMessageCount() throws IOException {
return delegate.getMessageCount();
}
+ public void recoverNextMessages(int maxReturned, MessageRecoveryListener listener) throws Exception {
+ delegate.recoverNextMessages(maxReturned, listener);
- public void recoverNextMessages(int maxReturned,MessageRecoveryListener listener) throws Exception{
- delegate.recoverNextMessages(maxReturned,listener);
-
}
- public void resetBatching(){
+ public void resetBatching() {
delegate.resetBatching();
-
+
}
}
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ProxyTopicMessageStore.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ProxyTopicMessageStore.java?view=diff&rev=563982&r1=563981&r2=563982
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ProxyTopicMessageStore.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ProxyTopicMessageStore.java Wed Aug 8 11:56:59 2007
@@ -32,64 +32,74 @@
public class ProxyTopicMessageStore implements TopicMessageStore {
final TopicMessageStore delegate;
-
+
public ProxyTopicMessageStore(TopicMessageStore delegate) {
this.delegate = delegate;
}
-
+
public MessageStore getDelegate() {
return delegate;
}
-
+
public void addMessage(ConnectionContext context, Message message) throws IOException {
delegate.addMessage(context, message);
}
+
public Message getMessage(MessageId identity) throws IOException {
return delegate.getMessage(identity);
}
+
public void recover(MessageRecoveryListener listener) throws Exception {
delegate.recover(listener);
}
+
public void removeAllMessages(ConnectionContext context) throws IOException {
delegate.removeAllMessages(context);
}
+
public void removeMessage(ConnectionContext context, MessageAck ack) throws IOException {
delegate.removeMessage(context, ack);
}
+
public void start() throws Exception {
delegate.start();
}
+
public void stop() throws Exception {
delegate.stop();
}
-
+
public SubscriptionInfo lookupSubscription(String clientId, String subscriptionName) throws IOException {
return delegate.lookupSubscription(clientId, subscriptionName);
}
- public void acknowledge(ConnectionContext context, String clientId, String subscriptionName, MessageId messageId)
- throws IOException {
+
+ public void acknowledge(ConnectionContext context, String clientId, String subscriptionName,
+ MessageId messageId) throws IOException {
delegate.acknowledge(context, clientId, subscriptionName, messageId);
}
+
public void addSubsciption(SubscriptionInfo subscriptionInfo, boolean retroactive) throws IOException {
delegate.addSubsciption(subscriptionInfo, retroactive);
}
+
public void deleteSubscription(String clientId, String subscriptionName) throws IOException {
delegate.deleteSubscription(clientId, subscriptionName);
}
- public void recoverSubscription(String clientId, String subscriptionName, MessageRecoveryListener listener) throws Exception {
+ public void recoverSubscription(String clientId, String subscriptionName, MessageRecoveryListener listener)
+ throws Exception {
delegate.recoverSubscription(clientId, subscriptionName, listener);
}
-
- public void recoverNextMessages(String clientId,String subscriptionName,int maxReturned,MessageRecoveryListener listener) throws Exception{
- delegate.recoverNextMessages(clientId, subscriptionName, maxReturned,listener);
- }
-
- public void resetBatching(String clientId,String subscriptionName) {
- delegate.resetBatching(clientId,subscriptionName);
+
+ public void recoverNextMessages(String clientId, String subscriptionName, int maxReturned,
+ MessageRecoveryListener listener) throws Exception {
+ delegate.recoverNextMessages(clientId, subscriptionName, maxReturned, listener);
+ }
+
+ public void resetBatching(String clientId, String subscriptionName) {
+ delegate.resetBatching(clientId, subscriptionName);
}
-
-
+
public ActiveMQDestination getDestination() {
return delegate.getDestination();
}
@@ -97,27 +107,26 @@
public SubscriptionInfo[] getAllSubscriptions() throws IOException {
return delegate.getAllSubscriptions();
}
-
+
public void setUsageManager(UsageManager usageManager) {
delegate.setUsageManager(usageManager);
}
- public int getMessageCount(String clientId,String subscriberName) throws IOException{
- return delegate.getMessageCount(clientId,subscriberName);
- }
-
-
- public int getMessageCount() throws IOException{
+ public int getMessageCount(String clientId, String subscriberName) throws IOException {
+ return delegate.getMessageCount(clientId, subscriberName);
+ }
+
+ public int getMessageCount() throws IOException {
return delegate.getMessageCount();
}
- public void recoverNextMessages(int maxReturned,MessageRecoveryListener listener) throws Exception{
- delegate.recoverNextMessages(maxReturned,listener);
-
+ public void recoverNextMessages(int maxReturned, MessageRecoveryListener listener) throws Exception {
+ delegate.recoverNextMessages(maxReturned, listener);
+
}
- public void resetBatching(){
+ public void resetBatching() {
delegate.resetBatching();
-
+
}
}
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ReferenceStore.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ReferenceStore.java?view=diff&rev=563982&r1=563981&r2=563982
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ReferenceStore.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ReferenceStore.java Wed Aug 8 11:56:59 2007
@@ -23,59 +23,65 @@
import org.apache.activemq.command.MessageId;
/**
- * Represents a message store which is used by the persistent
- * implementations
+ * Represents a message store which is used by the persistent implementations
*
* @version $Revision: 1.5 $
*/
public interface ReferenceStore extends MessageStore {
- public class ReferenceData {
- long expiration;
- int fileId;
- int offset;
-
- public long getExpiration() {
- return expiration;
- }
- public void setExpiration(long expiration) {
- this.expiration = expiration;
- }
- public int getFileId() {
- return fileId;
- }
- public void setFileId(int file) {
- this.fileId = file;
- }
- public int getOffset() {
- return offset;
- }
- public void setOffset(int offset) {
- this.offset = offset;
- }
-
- @Override
- public String toString() {
- return "ReferenceData fileId="+fileId+", offset="+offset+", expiration="+expiration;
- }
- }
-
+ public class ReferenceData {
+ long expiration;
+ int fileId;
+ int offset;
+
+ public long getExpiration() {
+ return expiration;
+ }
+
+ public void setExpiration(long expiration) {
+ this.expiration = expiration;
+ }
+
+ public int getFileId() {
+ return fileId;
+ }
+
+ public void setFileId(int file) {
+ this.fileId = file;
+ }
+
+ public int getOffset() {
+ return offset;
+ }
+
+ public void setOffset(int offset) {
+ this.offset = offset;
+ }
+
+ @Override
+ public String toString() {
+ return "ReferenceData fileId=" + fileId + ", offset=" + offset + ", expiration=" + expiration;
+ }
+ }
+
/**
* Adds a message reference to the message store
*/
- public void addMessageReference(ConnectionContext context, MessageId messageId, ReferenceData data) throws IOException;
+ public void addMessageReference(ConnectionContext context, MessageId messageId, ReferenceData data)
+ throws IOException;
/**
- * Looks up a message using either the String messageID or the messageNumber. Implementations are encouraged to fill
- * in the missing key if its easy to do so.
+ * Looks up a message using either the String messageID or the
+ * messageNumber. Implementations are encouraged to fill in the missing key
+ * if its easy to do so.
*/
public ReferenceData getMessageReference(MessageId identity) throws IOException;
-
+
/**
* @return true if it supports external batch control
*/
public boolean supportsExternalBatchControl();
-
+
public void setBatch(MessageId startAfter);
-
+
}
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ReferenceStoreAdapter.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ReferenceStoreAdapter.java?view=diff&rev=563982&r1=563981&r2=563982
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ReferenceStoreAdapter.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/ReferenceStoreAdapter.java Wed Aug 8 11:56:59 2007
@@ -27,61 +27,67 @@
/**
* Adapter to the actual persistence mechanism used with ActiveMQ
- *
+ *
* @version $Revision: 1.3 $
*/
public interface ReferenceStoreAdapter extends PersistenceAdapter {
/**
- * Factory method to create a new queue message store with the given destination name
- * @param destination
+ * Factory method to create a new queue message store with the given
+ * destination name
+ *
+ * @param destination
* @return the QueueReferenceStore
- * @throws IOException
+ * @throws IOException
*/
public ReferenceStore createQueueReferenceStore(ActiveMQQueue destination) throws IOException;
/**
- * Factory method to create a new topic message store with the given destination name
+ * Factory method to create a new topic message store with the given
+ * destination name
+ *
* @param destination
- * @return the TopicRefererenceStore
- * @throws IOException
+ * @return the TopicRefererenceStore
+ * @throws IOException
*/
public TopicReferenceStore createTopicReferenceStore(ActiveMQTopic destination) throws IOException;
- /**
- * @return Set of File ids in use
- * @throws IOException
- */
- public Set<Integer> getReferenceFileIdsInUse() throws IOException;
-
+ /**
+ * @return Set of File ids in use
+ * @throws IOException
+ */
+ public Set<Integer> getReferenceFileIdsInUse() throws IOException;
+
/**
* If the store isn't valid, it can be recoverd at start-up
+ *
* @return true if the reference store is in a consistent state
*/
public boolean isStoreValid();
-
+
/**
* called by recover to clear out message references
- * @throws IOException
+ *
+ * @throws IOException
*/
public void clearMessages() throws IOException;
-
+
/**
- * recover any state
- * @throws IOException
- *
+ * recover any state
+ *
+ * @throws IOException
*/
public void recoverState() throws IOException;
-
+
/**
* Save prepared transactions
+ *
* @param map
* @throws IOException
*/
- public void savePreparedState(Map<TransactionId, AMQTx> map)throws IOException;
-
+ public void savePreparedState(Map<TransactionId, AMQTx> map) throws IOException;
+
/**
- *
* @return saved prepared transactions
* @throws IOException
*/
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/TopicMessageStore.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/TopicMessageStore.java?view=diff&rev=563982&r1=563981&r2=563982
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/TopicMessageStore.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/TopicMessageStore.java Wed Aug 8 11:56:59 2007
@@ -19,15 +19,16 @@
import org.apache.activemq.command.Message;
import org.apache.activemq.command.MessageId;
import org.apache.activemq.command.SubscriptionInfo;
+
/**
* A MessageStore for durable topic subscriptions
*
* @version $Revision: 1.4 $
*/
-public interface TopicMessageStore extends MessageStore{
+public interface TopicMessageStore extends MessageStore {
/**
- * Stores the last acknowledged messgeID for the given subscription so that we can recover and commence dispatching
- * messages from the last checkpoint
+ * Stores the last acknowledged messgeID for the given subscription so that
+ * we can recover and commence dispatching messages from the last checkpoint
*
* @param context
* @param clientId
@@ -36,8 +37,8 @@
* @param subscriptionPersistentId
* @throws IOException
*/
- public void acknowledge(ConnectionContext context,String clientId,String subscriptionName,MessageId messageId)
- throws IOException;
+ public void acknowledge(ConnectionContext context, String clientId, String subscriptionName,
+ MessageId messageId) throws IOException;
/**
* @param clientId
@@ -46,12 +47,14 @@
* @throws IOException
* @throws JMSException
*/
- public void deleteSubscription(String clientId,String subscriptionName) throws IOException;
+ public void deleteSubscription(String clientId, String subscriptionName) throws IOException;
/**
- * For the new subscription find the last acknowledged message ID and then find any new messages since then and
- * dispatch them to the subscription. <p/> e.g. if we dispatched some messages to a new durable topic subscriber,
- * then went down before acknowledging any messages, we need to know the correct point from which to recover from.
+ * For the new subscription find the last acknowledged message ID and then
+ * find any new messages since then and dispatch them to the subscription.
+ * <p/> e.g. if we dispatched some messages to a new durable topic
+ * subscriber, then went down before acknowledging any messages, we need to
+ * know the correct point from which to recover from.
*
* @param clientId
* @param subscriptionName
@@ -60,12 +63,12 @@
*
* @throws Exception
*/
- public void recoverSubscription(String clientId,String subscriptionName,MessageRecoveryListener listener)
- throws Exception;
+ public void recoverSubscription(String clientId, String subscriptionName, MessageRecoveryListener listener)
+ throws Exception;
/**
- * For an active subscription - retrieve messages from the store for the subscriber after the lastMessageId
- * messageId <p/>
+ * For an active subscription - retrieve messages from the store for the
+ * subscriber after the lastMessageId messageId <p/>
*
* @param clientId
* @param subscriptionName
@@ -74,27 +77,29 @@
*
* @throws Exception
*/
- public void recoverNextMessages(String clientId,String subscriptionName,int maxReturned,
- MessageRecoveryListener listener) throws Exception;
+ public void recoverNextMessages(String clientId, String subscriptionName, int maxReturned,
+ MessageRecoveryListener listener) throws Exception;
/**
* A hint to the Store to reset any batching state for a durable subsriber
- * @param clientId
- * @param subscriptionName
- *
- */
- public void resetBatching(String clientId,String subscriptionName);
-
-
+ *
+ * @param clientId
+ * @param subscriptionName
+ *
+ */
+ public void resetBatching(String clientId, String subscriptionName);
+
/**
- * Get the number of messages ready to deliver from the store to a durable subscriber
+ * Get the number of messages ready to deliver from the store to a durable
+ * subscriber
+ *
* @param clientId
* @param subscriberName
* @return the outstanding message count
* @throws IOException
*/
- public int getMessageCount(String clientId,String subscriberName) throws IOException;
-
+ public int getMessageCount(String clientId, String subscriberName) throws IOException;
+
/**
* Finds the subscriber entry for the given consumer info
*
@@ -103,7 +108,7 @@
* @return the SubscriptionInfo
* @throws IOException
*/
- public SubscriptionInfo lookupSubscription(String clientId,String subscriptionName) throws IOException;
+ public SubscriptionInfo lookupSubscription(String clientId, String subscriptionName) throws IOException;
/**
* Lists all the durable subscriptions for a given destination.
@@ -114,10 +119,12 @@
public SubscriptionInfo[] getAllSubscriptions() throws IOException;
/**
- * Inserts the subscriber info due to a subscription change <p/> If this is a new subscription and the retroactive
- * is false, then the last message sent to the topic should be set as the last message acknowledged by they new
- * subscription. Otherwise, if retroactive is true, then create the subscription without it having an acknowledged
- * message so that on recovery, all message recorded for the topic get replayed.
+ * Inserts the subscriber info due to a subscription change <p/> If this is
+ * a new subscription and the retroactive is false, then the last message
+ * sent to the topic should be set as the last message acknowledged by they
+ * new subscription. Otherwise, if retroactive is true, then create the
+ * subscription without it having an acknowledged message so that on
+ * recovery, all message recorded for the topic get replayed.
*
* @param clientId
* @param subscriptionName
@@ -126,6 +133,5 @@
* @throws IOException
*
*/
- public void addSubsciption(SubscriptionInfo subscriptionInfo, boolean retroactive)
- throws IOException;
+ public void addSubsciption(SubscriptionInfo subscriptionInfo, boolean retroactive) throws IOException;
}
Modified: activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/TopicReferenceStore.java
URL: http://svn.apache.org/viewvc/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/TopicReferenceStore.java?view=diff&rev=563982&r1=563981&r2=563982
==============================================================================
--- activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/TopicReferenceStore.java (original)
+++ activemq/trunk/activemq-core/src/main/java/org/apache/activemq/store/TopicReferenceStore.java Wed Aug 8 11:56:59 2007
@@ -31,8 +31,8 @@
*/
public interface TopicReferenceStore extends ReferenceStore, TopicMessageStore {
/**
- * Stores the last acknowledged messgeID for the given subscription so that we can recover and commence dispatching
- * messages from the last checkpoint
+ * Stores the last acknowledged messgeID for the given subscription so that
+ * we can recover and commence dispatching messages from the last checkpoint
*
* @param context
* @param clientId
@@ -41,8 +41,8 @@
* @param subscriptionPersistentId
* @throws IOException
*/
- public void acknowledge(ConnectionContext context,String clientId,String subscriptionName,MessageId messageId)
- throws IOException;
+ public void acknowledge(ConnectionContext context, String clientId, String subscriptionName,
+ MessageId messageId) throws IOException;
/**
* @param clientId
@@ -51,12 +51,14 @@
* @throws IOException
* @throws JMSException
*/
- public void deleteSubscription(String clientId,String subscriptionName) throws IOException;
+ public void deleteSubscription(String clientId, String subscriptionName) throws IOException;
/**
- * For the new subscription find the last acknowledged message ID and then find any new messages since then and
- * dispatch them to the subscription. <p/> e.g. if we dispatched some messages to a new durable topic subscriber,
- * then went down before acknowledging any messages, we need to know the correct point from which to recover from.
+ * For the new subscription find the last acknowledged message ID and then
+ * find any new messages since then and dispatch them to the subscription.
+ * <p/> e.g. if we dispatched some messages to a new durable topic
+ * subscriber, then went down before acknowledging any messages, we need to
+ * know the correct point from which to recover from.
*
* @param clientId
* @param subscriptionName
@@ -65,12 +67,12 @@
*
* @throws Exception
*/
- public void recoverSubscription(String clientId,String subscriptionName,MessageRecoveryListener listener)
- throws Exception;
+ public void recoverSubscription(String clientId, String subscriptionName, MessageRecoveryListener listener)
+ throws Exception;
/**
- * For an active subscription - retrieve messages from the store for the subscriber after the lastMessageId
- * messageId <p/>
+ * For an active subscription - retrieve messages from the store for the
+ * subscriber after the lastMessageId messageId <p/>
*
* @param clientId
* @param subscriptionName
@@ -79,27 +81,29 @@
*
* @throws Exception
*/
- public void recoverNextMessages(String clientId,String subscriptionName,int maxReturned,
- MessageRecoveryListener listener) throws Exception;
+ public void recoverNextMessages(String clientId, String subscriptionName, int maxReturned,
+ MessageRecoveryListener listener) throws Exception;
/**
* A hint to the Store to reset any batching state for a durable subsriber
- * @param clientId
- * @param subscriptionName
- *
- */
- public void resetBatching(String clientId,String subscriptionName);
-
-
+ *
+ * @param clientId
+ * @param subscriptionName
+ *
+ */
+ public void resetBatching(String clientId, String subscriptionName);
+
/**
- * Get the number of messages ready to deliver from the store to a durable subscriber
+ * Get the number of messages ready to deliver from the store to a durable
+ * subscriber
+ *
* @param clientId
* @param subscriberName
* @return the outstanding message count
* @throws IOException
*/
- public int getMessageCount(String clientId,String subscriberName) throws IOException;
-
+ public int getMessageCount(String clientId, String subscriberName) throws IOException;
+
/**
* Finds the subscriber entry for the given consumer info
*
@@ -108,7 +112,7 @@
* @return the SubscriptionInfo
* @throws IOException
*/
- public SubscriptionInfo lookupSubscription(String clientId,String subscriptionName) throws IOException;
+ public SubscriptionInfo lookupSubscription(String clientId, String subscriptionName) throws IOException;
/**
* Lists all the durable subscirptions for a given destination.
@@ -119,10 +123,12 @@
public SubscriptionInfo[] getAllSubscriptions() throws IOException;
/**
- * Inserts the subscriber info due to a subscription change <p/> If this is a new subscription and the retroactive
- * is false, then the last message sent to the topic should be set as the last message acknowledged by they new
- * subscription. Otherwise, if retroactive is true, then create the subscription without it having an acknowledged
- * message so that on recovery, all message recorded for the topic get replayed.
+ * Inserts the subscriber info due to a subscription change <p/> If this is
+ * a new subscription and the retroactive is false, then the last message
+ * sent to the topic should be set as the last message acknowledged by they
+ * new subscription. Otherwise, if retroactive is true, then create the
+ * subscription without it having an acknowledged message so that on
+ * recovery, all message recorded for the topic get replayed.
*
* @param clientId
* @param subscriptionName
@@ -131,6 +137,5 @@
* @throws IOException
*
*/
- public void addSubsciption(SubscriptionInfo subscriptionInfo,boolean retroactive)
- throws IOException;
+ public void addSubsciption(SubscriptionInfo subscriptionInfo, boolean retroactive) throws IOException;
}