Author: rajdavies
Date: Fri Feb 17 23:18:01 2006
New Revision: 378699
URL: http://svn.apache.org/viewcvs?rev=378699&view=rev
Log:
Extended TransportListener to notify when transport is interrupted and resumed
Added:
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/DefaultTransportListener.java
(with props)
Modified:
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/AbstractConnection.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/ft/MasterConnector.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/proxy/ProxyConnection.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/TransportFilter.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/TransportListener.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/fanout/FanoutTransport.java
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/mock/MockTransport.java
incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/StubConnection.java
Modified:
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
URL:
http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java?rev=378699&r1=378698&r2=378699&view=diff
==============================================================================
---
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
(original)
+++
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/ActiveMQConnection.java
Fri Feb 17 23:18:01 2006
@@ -24,7 +24,6 @@
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
-
import javax.jms.Connection;
import javax.jms.ConnectionConsumer;
import javax.jms.ConnectionMetaData;
@@ -42,7 +41,6 @@
import javax.jms.TopicConnection;
import javax.jms.TopicSession;
import javax.jms.XAConnection;
-
import org.apache.activemq.command.ActiveMQDestination;
import org.apache.activemq.command.ActiveMQMessage;
import org.apache.activemq.command.ActiveMQTempDestination;
@@ -71,8 +69,8 @@
import org.apache.activemq.management.StatsCapable;
import org.apache.activemq.management.StatsImpl;
import org.apache.activemq.thread.TaskRunnerFactory;
+import org.apache.activemq.transport.DefaultTransportListener;
import org.apache.activemq.transport.Transport;
-import org.apache.activemq.transport.TransportListener;
import org.apache.activemq.util.IdGenerator;
import org.apache.activemq.util.IntrospectionSupport;
import org.apache.activemq.util.JMSExceptionSupport;
@@ -80,14 +78,13 @@
import org.apache.activemq.util.ServiceSupport;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-
import edu.emory.mathcs.backport.java.util.concurrent.ConcurrentHashMap;
import edu.emory.mathcs.backport.java.util.concurrent.CopyOnWriteArrayList;
import edu.emory.mathcs.backport.java.util.concurrent.CountDownLatch;
import edu.emory.mathcs.backport.java.util.concurrent.atomic.AtomicBoolean;
-public class ActiveMQConnection implements Connection, TopicConnection,
QueueConnection, StatsCapable, Closeable, TransportListener, StreamConnection {
+public class ActiveMQConnection extends DefaultTransportListener implements
Connection, TopicConnection, QueueConnection, StatsCapable, Closeable,
StreamConnection {
public static final TaskRunnerFactory SESSION_TASK_RUNNER = new
TaskRunnerFactory("session
Task",ThreadPriorities.INBOUND_CLIENT_SESSION,true,1000);
Modified:
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/AbstractConnection.java
URL:
http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/AbstractConnection.java?rev=378699&r1=378698&r2=378699&view=diff
==============================================================================
---
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/AbstractConnection.java
(original)
+++
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/AbstractConnection.java
Fri Feb 17 23:18:01 2006
@@ -442,6 +442,7 @@
}
public Response processRemoveSession(SessionId id) throws Throwable {
+
ConnectionId connectionId = id.getParentId();
ConnectionState cs = lookupConnectionState(connectionId);
@@ -491,22 +492,37 @@
return null;
}
- public Response processRemoveConnection(ConnectionId id) throws Throwable {
+ public Response processRemoveConnection(ConnectionId id) {
ConnectionState cs = lookupConnectionState(id);
// Cascade the connection stop to the sessions.
for (Iterator iter = cs.getSessionIds().iterator(); iter.hasNext();) {
- processRemoveSession((SessionId) iter.next());
+
+ SessionId sessionId = (SessionId) iter.next();
+ try{
+ processRemoveSession(sessionId);
+ }catch(Throwable e){
+ serviceLog.warn("Failed to remove session " + sessionId,e);
+ }
}
// Cascade the connection stop to temp destinations.
for (Iterator iter = cs.getTempDesinations().iterator();
iter.hasNext();) {
- broker.removeDestination(cs.getContext(), (ActiveMQDestination)
iter.next(), 0);
+ ActiveMQDestination dest = (ActiveMQDestination) iter.next();
+ try{
+ broker.removeDestination(cs.getContext(), dest, 0);
+ }catch(Throwable e){
+ serviceLog.warn("Failed to remove tmp destination " + dest,e);
+ }
iter.remove();
}
- broker.removeConnection(cs.getContext(), cs.getInfo(), null);
+ try{
+ broker.removeConnection(cs.getContext(), cs.getInfo(), null);
+ }catch(Throwable e){
+ serviceLog.warn("Failed to remove connection " + cs.getInfo(),e);
+ }
connectionStates.remove(id);
return null;
Modified:
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java
URL:
http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java?rev=378699&r1=378698&r2=378699&view=diff
==============================================================================
---
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java
(original)
+++
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java
Fri Feb 17 23:18:01 2006
@@ -24,6 +24,7 @@
import org.apache.activemq.command.Response;
import org.apache.activemq.command.ShutdownInfo;
import org.apache.activemq.thread.TaskRunnerFactory;
+import org.apache.activemq.transport.DefaultTransportListener;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.transport.TransportListener;
import org.apache.commons.logging.Log;
@@ -55,7 +56,7 @@
super(connector, broker, taskRunnerFactory);
connector.setBrokerName(broker.getBrokerName());
this.transport = transport;
- this.transport.setTransportListener(new TransportListener() {
+ this.transport.setTransportListener(new DefaultTransportListener() {
public void onCommand(Command command) {
Response response = service(command);
if( response!=null ) {
Modified:
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/ft/MasterConnector.java
URL:
http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/ft/MasterConnector.java?rev=378699&r1=378698&r2=378699&view=diff
==============================================================================
---
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/ft/MasterConnector.java
(original)
+++
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/ft/MasterConnector.java
Fri Feb 17 23:18:01 2006
@@ -16,8 +16,8 @@
*/
package org.apache.activemq.broker.ft;
-import edu.emory.mathcs.backport.java.util.concurrent.atomic.AtomicBoolean;
-
+import java.io.IOException;
+import java.net.URI;
import org.apache.activemq.Service;
import org.apache.activemq.broker.BrokerService;
import org.apache.activemq.broker.TransportConnector;
@@ -31,17 +31,15 @@
import org.apache.activemq.command.Response;
import org.apache.activemq.command.SessionInfo;
import org.apache.activemq.command.ShutdownInfo;
+import org.apache.activemq.transport.DefaultTransportListener;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.transport.TransportFactory;
-import org.apache.activemq.transport.TransportListener;
import org.apache.activemq.util.IdGenerator;
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.io.IOException;
-import java.net.URI;
+import edu.emory.mathcs.backport.java.util.concurrent.atomic.AtomicBoolean;
/**
* Used by a Slave Broker to Connect to the Master
@@ -83,7 +81,7 @@
remoteBroker=TransportFactory.connect(remoteURI);
log.info("Starting a network connection between "+localBroker+" and
"+remoteBroker+" has been established.");
- localBroker.setTransportListener(new TransportListener(){
+ localBroker.setTransportListener(new DefaultTransportListener(){
public void onCommand(Command command){
}
public void onException(IOException error){
@@ -93,7 +91,7 @@
}
});
- remoteBroker.setTransportListener(new TransportListener(){
+ remoteBroker.setTransportListener(new DefaultTransportListener(){
public void onCommand(Command command){
if( started.get() ) {
serviceRemoteCommand(command);
Modified:
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/proxy/ProxyConnection.java
URL:
http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/proxy/ProxyConnection.java?rev=378699&r1=378698&r2=378699&view=diff
==============================================================================
---
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/proxy/ProxyConnection.java
(original)
+++
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/proxy/ProxyConnection.java
Fri Feb 17 23:18:01 2006
@@ -17,17 +17,16 @@
package org.apache.activemq.proxy;
import java.io.IOException;
-
import org.apache.activemq.Service;
import org.apache.activemq.command.Command;
import org.apache.activemq.command.ShutdownInfo;
+import org.apache.activemq.transport.DefaultTransportListener;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.transport.TransportListener;
import org.apache.activemq.util.IOExceptionSupport;
import org.apache.activemq.util.ServiceStopper;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
-
import edu.emory.mathcs.backport.java.util.concurrent.atomic.AtomicBoolean;
class ProxyConnection implements Service {
@@ -59,7 +58,7 @@
return;
}
- this.localTransport.setTransportListener(new TransportListener() {
+ this.localTransport.setTransportListener(new
DefaultTransportListener() {
public void onCommand(Command command) {
boolean shutdown=false;
if( command.getClass() == ShutdownInfo.class ) {
@@ -81,7 +80,7 @@
}
});
- this.remoteTransport.setTransportListener(new TransportListener() {
+ this.remoteTransport.setTransportListener(new
DefaultTransportListener() {
public void onCommand(Command command) {
try {
localTransport.oneway(command);
Added:
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/DefaultTransportListener.java
URL:
http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/DefaultTransportListener.java?rev=378699&view=auto
==============================================================================
---
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/DefaultTransportListener.java
(added)
+++
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/DefaultTransportListener.java
Fri Feb 17 23:18:01 2006
@@ -0,0 +1,58 @@
+/**
+ *
+ * Copyright 2005-2006 The Apache Software Foundation
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.activemq.transport;
+
+import java.io.IOException;
+
+import org.apache.activemq.command.Command;
+
+/**
+ * An asynchronous listener of commands
+ *
+ * @version $Revision$
+ */
+public class DefaultTransportListener implements TransportListener {
+
+ /**
+ * called to process a command
+ * @param command
+ */
+ public void onCommand(Command command){
+ }
+ /**
+ * An unrecoverable exception has occured on the transport
+ * @param error
+ */
+ public void onException(IOException error){
+ }
+
+ /**
+ * The transport has suffered an interuption from which it hopes to recover
+ *
+ */
+ public void transportInterupted(){
+ }
+
+
+ /**
+ * The transport has resumed after an interuption
+ *
+ */
+ public void transportResumed(){
+ }
+
+}
Propchange:
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/DefaultTransportListener.java
------------------------------------------------------------------------------
svn:executable = *
Modified:
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/TransportFilter.java
URL:
http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/TransportFilter.java?rev=378699&r1=378698&r2=378699&view=diff
==============================================================================
---
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/TransportFilter.java
(original)
+++
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/TransportFilter.java
Fri Feb 17 23:18:01 2006
@@ -25,7 +25,7 @@
/**
* @version $Revision: 1.5 $
*/
-public class TransportFilter implements Transport, TransportListener {
+public class TransportFilter extends DefaultTransportListener implements
Transport {
final protected Transport next;
protected TransportListener commandListener;
Modified:
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/TransportListener.java
URL:
http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/TransportListener.java?rev=378699&r1=378698&r2=378699&view=diff
==============================================================================
---
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/TransportListener.java
(original)
+++
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/TransportListener.java
Fri Feb 17 23:18:01 2006
@@ -27,7 +27,28 @@
*/
public interface TransportListener {
+ /**
+ * called to process a command
+ * @param command
+ */
public void onCommand(Command command);
+ /**
+ * An unrecoverable exception has occured on the transport
+ * @param error
+ */
public void onException(IOException error);
+
+ /**
+ * The transport has suffered an interuption from which it hopes to recover
+ *
+ */
+ public void transportInterupted();
+
+
+ /**
+ * The transport has resumed after an interuption
+ *
+ */
+ public void transportResumed();
}
Modified:
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
URL:
http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java?rev=378699&r1=378698&r2=378699&view=diff
==============================================================================
---
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
(original)
+++
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
Fri Feb 17 23:18:01 2006
@@ -31,6 +31,7 @@
import org.apache.activemq.thread.Task;
import org.apache.activemq.thread.TaskRunner;
import org.apache.activemq.transport.CompositeTransport;
+import org.apache.activemq.transport.DefaultTransportListener;
import org.apache.activemq.transport.FutureResponse;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.transport.TransportFactory;
@@ -77,7 +78,7 @@
private long reconnectDelay = initialReconnectDelay;
private Exception connectionFailure;
- private final TransportListener myTransportListener = new
TransportListener() {
+ private final TransportListener myTransportListener = new
DefaultTransportListener() {
public void onCommand(Command command) {
if (command == null) {
return;
@@ -152,6 +153,9 @@
connectedTransport = t;
reconnectMutex.notifyAll();
connectFailures = 0;
+ if (transportListener != null){
+ transportListener.transportResumed();
+ }
return false;
}
catch (Exception e) {
@@ -191,6 +195,9 @@
}
private void handleTransportFailure(IOException e) throws
InterruptedException {
+ if (transportListener != null){
+ transportListener.transportInterupted();
+ }
synchronized (reconnectMutex) {
log.debug("Transport failed, starting up reconnect task", e);
if (connectedTransport != null) {
Modified:
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/fanout/FanoutTransport.java
URL:
http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/fanout/FanoutTransport.java?rev=378699&r1=378698&r2=378699&view=diff
==============================================================================
---
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/fanout/FanoutTransport.java
(original)
+++
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/fanout/FanoutTransport.java
Fri Feb 17 23:18:01 2006
@@ -31,6 +31,7 @@
import org.apache.activemq.thread.Task;
import org.apache.activemq.thread.TaskRunner;
import org.apache.activemq.transport.CompositeTransport;
+import org.apache.activemq.transport.DefaultTransportListener;
import org.apache.activemq.transport.FutureResponse;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.transport.TransportFactory;
@@ -91,7 +92,7 @@
}
}
- class FanoutTransportHandler implements TransportListener {
+ class FanoutTransportHandler extends DefaultTransportListener {
private final URI uri;
private Transport transport;
Modified:
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/mock/MockTransport.java
URL:
http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/mock/MockTransport.java?rev=378699&r1=378698&r2=378699&view=diff
==============================================================================
---
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/mock/MockTransport.java
(original)
+++
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/transport/mock/MockTransport.java
Fri Feb 17 23:18:01 2006
@@ -20,6 +20,7 @@
import org.apache.activemq.command.Command;
import org.apache.activemq.command.Response;
+import org.apache.activemq.transport.DefaultTransportListener;
import org.apache.activemq.transport.FutureResponse;
import org.apache.activemq.transport.Transport;
import org.apache.activemq.transport.TransportFilter;
@@ -29,7 +30,7 @@
/**
* @version $Revision: 1.5 $
*/
-public class MockTransport implements Transport, TransportListener {
+public class MockTransport extends DefaultTransportListener implements
Transport {
protected Transport next;
protected TransportListener commandListener;
Modified:
incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/StubConnection.java
URL:
http://svn.apache.org/viewcvs/incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/StubConnection.java?rev=378699&r1=378698&r2=378699&view=diff
==============================================================================
---
incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/StubConnection.java
(original)
+++
incubator/activemq/trunk/activemq-core/src/test/java/org/apache/activemq/broker/StubConnection.java
Fri Feb 17 23:18:01 2006
@@ -17,22 +17,17 @@
package org.apache.activemq.broker;
import java.io.IOException;
-
import org.apache.activemq.Service;
-import org.apache.activemq.broker.AbstractConnection;
-import org.apache.activemq.broker.BrokerService;
-import org.apache.activemq.broker.Connection;
import org.apache.activemq.command.Command;
import org.apache.activemq.command.ExceptionResponse;
import org.apache.activemq.command.Message;
import org.apache.activemq.command.Response;
import org.apache.activemq.command.ShutdownInfo;
import org.apache.activemq.thread.TaskRunnerFactory;
+import org.apache.activemq.transport.DefaultTransportListener;
import org.apache.activemq.transport.Transport;
-import org.apache.activemq.transport.TransportListener;
import org.apache.activemq.util.ServiceSupport;
import org.axiondb.engine.commands.ShutdownCommand;
-
import edu.emory.mathcs.backport.java.util.concurrent.BlockingQueue;
import edu.emory.mathcs.backport.java.util.concurrent.LinkedBlockingQueue;
@@ -70,7 +65,7 @@
public StubConnection(Transport transport) throws Exception {
this.transport = transport;
- transport.setTransportListener(new TransportListener() {
+ transport.setTransportListener(new DefaultTransportListener() {
public void onCommand(Command command) {
try {
if (command.getClass() == ShutdownCommand.class) {
