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

Fix for https://issues.apache.org/activemq/browse/AMQ-1572 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@616506 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Robert Davies 2008-01-29 20:06:47 +00:00
parent 3249682bda
commit e210657db2
1 changed files with 65 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

70
activemq-core/src/main/java/org/apache/activemq/transport/failover/FailoverTransport.java
View File

@ -21,6 +21,7 @@ import java.io.InterruptedIOException;
import java.net.URI; import java.net.URI;
import java.util.ArrayList; import java.util.ArrayList;
import java.util.Iterator; import java.util.Iterator;
import java.util.List;
import java.util.Random; import java.util.Random;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CopyOnWriteArrayList;
@ -34,6 +35,7 @@ import org.apache.activemq.thread.DefaultThreadPools;
import org.apache.activemq.thread.Task; import org.apache.activemq.thread.Task;
import org.apache.activemq.thread.TaskRunner; import org.apache.activemq.thread.TaskRunner;
import org.apache.activemq.transport.CompositeTransport; import org.apache.activemq.transport.CompositeTransport;
import org.apache.activemq.transport.DefaultTransportListener;
import org.apache.activemq.transport.FutureResponse; import org.apache.activemq.transport.FutureResponse;
import org.apache.activemq.transport.ResponseCallback; import org.apache.activemq.transport.ResponseCallback;
import org.apache.activemq.transport.Transport; import org.apache.activemq.transport.Transport;
@ -78,6 +80,12 @@ public class FailoverTransport implements CompositeTransport {
private int connectFailures; private int connectFailures;
private long reconnectDelay = initialReconnectDelay; private long reconnectDelay = initialReconnectDelay;
private Exception connectionFailure; private Exception connectionFailure;
private boolean firstConnection = true;
//optionally always have a backup created
private boolean backup=false;
private URI backupTransportURI;
private Transport backupTransport;
private final TransportListener myTransportListener = createTransportListener(); private final TransportListener myTransportListener = createTransportListener();
@ -100,16 +108,38 @@ public class FailoverTransport implements CompositeTransport {
if (connectedTransport != null || disposed || connectionFailure != null) { if (connectedTransport != null || disposed || connectionFailure != null) {
return false; return false;
} else { } else {
ArrayList<Object> connectList = getConnectList(); List<URI> connectList = getConnectList();
if (connectList.isEmpty()) { if (connectList.isEmpty()) {
failure = new IOException("No uris available to connect to."); failure = new IOException("No uris available to connect to.");
} else { } else {
if (!useExponentialBackOff) { if (!useExponentialBackOff) {
reconnectDelay = initialReconnectDelay; reconnectDelay = initialReconnectDelay;
} }
Iterator<Object> iter = connectList.iterator(); if (backup && backupTransport != null) {
for (int i = 0; iter.hasNext() && connectedTransport == null && !disposed; i++) { Transport t = backupTransport;
URI uri = (URI)iter.next(); URI uri = backupTransportURI;
backupTransport=null;
backupTransportURI=null;
t.setTransportListener(myTransportListener);
try {
if (started) {
restoreTransport(t);
}
reconnectDelay = initialReconnectDelay;
connectedTransportURI = uri;
connectedTransport = t;
reconnectMutex.notifyAll();
connectFailures = 0;
LOG.info("Successfully reconnected to backup " + uri);
return false;
}catch (Exception e) {
LOG.debug("Backup transport failed",e);
}
}
Iterator<URI> iter = connectList.iterator();
while(iter.hasNext() && connectedTransport == null && !disposed) {
URI uri = iter.next();
try { try {
LOG.debug("Attempting connect to: " + uri); LOG.debug("Attempting connect to: " + uri);
Transport t = TransportFactory.compositeConnect(uri); Transport t = TransportFactory.compositeConnect(uri);
@ -129,7 +159,26 @@ public class FailoverTransport implements CompositeTransport {
if (transportListener != null) { if (transportListener != null) {
transportListener.transportResumed(); transportListener.transportResumed();
} }
if (firstConnection) {
firstConnection=false;
LOG.info("Successfully connected to " + uri);
if(backup) {
while(iter.hasNext() && backupTransport==null){
uri = iter.next();
try {
t = TransportFactory.compositeConnect(uri);
t.setTransportListener(new DefaultTransportListener());
t.start();
backupTransport=t;
backupTransportURI=uri;
}catch(Exception e) {
LOG.debug("Failed to create backup to " + uri,e);
}
}
}
}else {
LOG.info("Successfully reconnected to " + uri); LOG.info("Successfully reconnected to " + uri);
}
return false; return false;
} catch (Exception e) { } catch (Exception e) {
failure = e; failure = e;
@ -488,19 +537,26 @@ public class FailoverTransport implements CompositeTransport {
} }
} }
private ArrayList<Object> getConnectList() { private List<URI> getConnectList() {
ArrayList<Object> l = new ArrayList<Object>(uris); ArrayList<URI> l = new ArrayList<URI>(uris);
boolean removed = false;
if (connectedTransportURI != null) {
removed = l.remove(connectedTransportURI);
}
if (randomize) { if (randomize) {
// Randomly, reorder the list by random swapping // Randomly, reorder the list by random swapping
Random r = new Random(); Random r = new Random();
r.setSeed(System.currentTimeMillis()); r.setSeed(System.currentTimeMillis());
for (int i = 0; i < l.size(); i++) { for (int i = 0; i < l.size(); i++) {
int p = r.nextInt(l.size()); int p = r.nextInt(l.size());
Object t = l.get(p); URI t = l.get(p);
l.set(p, l.get(i)); l.set(p, l.get(i));
l.set(i, t); l.set(i, t);
} }
} }
if (removed) {
l.add(connectedTransportURI);
}
return l; return l;
} }