activemq-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From rajdav...@apache.org
Subject svn commit: r378699 - in /incubator/activemq/trunk/activemq-core/src: main/java/org/apache/activemq/ main/java/org/apache/activemq/broker/ main/java/org/apache/activemq/broker/ft/ main/java/org/apache/activemq/proxy/ main/java/org/apache/activemq/trans...
Date Sat, 18 Feb 2006 07:18:05 GMT
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) {



Mime
View raw message