001/**
002 * Licensed to the Apache Software Foundation (ASF) under one or more
003 * contributor license agreements.  See the NOTICE file distributed with
004 * this work for additional information regarding copyright ownership.
005 * The ASF licenses this file to You under the Apache License, Version 2.0
006 * (the "License"); you may not use this file except in compliance with
007 * the License.  You may obtain a copy of the License at
008 *
009 *      http://www.apache.org/licenses/LICENSE-2.0
010 *
011 * Unless required by applicable law or agreed to in writing, software
012 * distributed under the License is distributed on an "AS IS" BASIS,
013 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
014 * See the License for the specific language governing permissions and
015 * limitations under the License.
016 */
017package org.apache.activemq.transport.failover;
018
019import java.io.BufferedReader;
020import java.io.FileReader;
021import java.io.IOException;
022import java.io.InputStreamReader;
023import java.io.InterruptedIOException;
024import java.net.InetAddress;
025import java.net.MalformedURLException;
026import java.net.URI;
027import java.net.URISyntaxException;
028import java.net.URL;
029import java.util.ArrayList;
030import java.util.Collections;
031import java.util.HashSet;
032import java.util.Iterator;
033import java.util.LinkedHashMap;
034import java.util.List;
035import java.util.Map;
036import java.util.StringTokenizer;
037import java.util.concurrent.CopyOnWriteArrayList;
038import java.util.concurrent.atomic.AtomicReference;
039
040import org.apache.activemq.broker.SslContext;
041import org.apache.activemq.command.Command;
042import org.apache.activemq.command.ConnectionControl;
043import org.apache.activemq.command.ConsumerControl;
044import org.apache.activemq.command.ConnectionId;
045import org.apache.activemq.command.MessageDispatch;
046import org.apache.activemq.command.MessagePull;
047import org.apache.activemq.command.RemoveInfo;
048import org.apache.activemq.command.Response;
049
050import org.apache.activemq.state.ConnectionStateTracker;
051import org.apache.activemq.state.Tracked;
052import org.apache.activemq.thread.Task;
053import org.apache.activemq.thread.TaskRunner;
054import org.apache.activemq.thread.TaskRunnerFactory;
055import org.apache.activemq.transport.CompositeTransport;
056import org.apache.activemq.transport.DefaultTransportListener;
057import org.apache.activemq.transport.FutureResponse;
058import org.apache.activemq.transport.ResponseCallback;
059import org.apache.activemq.transport.Transport;
060import org.apache.activemq.transport.TransportFactory;
061import org.apache.activemq.transport.TransportListener;
062import org.apache.activemq.util.IOExceptionSupport;
063import org.apache.activemq.util.ServiceSupport;
064import org.apache.activemq.util.URISupport;
065import org.slf4j.Logger;
066import org.slf4j.LoggerFactory;
067
068/**
069 * A Transport that is made reliable by being able to fail over to another
070 * transport when a transport failure is detected.
071 */
072public class FailoverTransport implements CompositeTransport {
073
074    private static final Logger LOG = LoggerFactory.getLogger(FailoverTransport.class);
075    private static final int DEFAULT_INITIAL_RECONNECT_DELAY = 10;
076    private static final int INFINITE = -1;
077    private TransportListener transportListener;
078    private boolean disposed;
079    private final CopyOnWriteArrayList<URI> uris = new CopyOnWriteArrayList<URI>();
080    private final CopyOnWriteArrayList<URI> updated = new CopyOnWriteArrayList<URI>();
081
082    private final Object reconnectMutex = new Object();
083    private final Object backupMutex = new Object();
084    private final Object sleepMutex = new Object();
085    private final Object listenerMutex = new Object();
086    private final ConnectionStateTracker stateTracker = new ConnectionStateTracker();
087    private final Map<Integer, Command> requestMap = new LinkedHashMap<Integer, Command>();
088
089    private URI connectedTransportURI;
090    private URI failedConnectTransportURI;
091    private final AtomicReference<Transport> connectedTransport = new AtomicReference<Transport>();
092    private final TaskRunnerFactory reconnectTaskFactory;
093    private final TaskRunner reconnectTask;
094    private boolean started;
095    private long initialReconnectDelay = DEFAULT_INITIAL_RECONNECT_DELAY;
096    private long maxReconnectDelay = 1000 * 30;
097    private double backOffMultiplier = 2d;
098    private long timeout = INFINITE;
099    private boolean useExponentialBackOff = true;
100    private boolean randomize = true;
101    private int maxReconnectAttempts = INFINITE;
102    private int startupMaxReconnectAttempts = INFINITE;
103    private int connectFailures;
104    private int warnAfterReconnectAttempts = 10;
105    private long reconnectDelay = DEFAULT_INITIAL_RECONNECT_DELAY;
106    private Exception connectionFailure;
107    private boolean firstConnection = true;
108    // optionally always have a backup created
109    private boolean backup = false;
110    private final List<BackupTransport> backups = new CopyOnWriteArrayList<BackupTransport>();
111    private int backupPoolSize = 1;
112    private boolean trackMessages = false;
113    private boolean trackTransactionProducers = true;
114    private int maxCacheSize = 128 * 1024;
115    private final TransportListener disposedListener = new DefaultTransportListener() {};
116    private boolean updateURIsSupported = true;
117    private boolean reconnectSupported = true;
118    // remember for reconnect thread
119    private SslContext brokerSslContext;
120    private String updateURIsURL = null;
121    private boolean rebalanceUpdateURIs = true;
122    private boolean doRebalance = false;
123    private boolean connectedToPriority = false;
124
125    private boolean priorityBackup = false;
126    private final ArrayList<URI> priorityList = new ArrayList<URI>();
127    private boolean priorityBackupAvailable = false;
128    private String nestedExtraQueryOptions;
129    private boolean shuttingDown = false;
130
131    public FailoverTransport() {
132        brokerSslContext = SslContext.getCurrentSslContext();
133        stateTracker.setTrackTransactions(true);
134        // Setup a task that is used to reconnect the a connection async.
135        reconnectTaskFactory = new TaskRunnerFactory();
136        reconnectTaskFactory.init();
137        reconnectTask = reconnectTaskFactory.createTaskRunner(new Task() {
138            @Override
139            public boolean iterate() {
140                boolean result = false;
141                if (!started) {
142                    return result;
143                }
144                boolean buildBackup = true;
145                synchronized (backupMutex) {
146                    if ((connectedTransport.get() == null || doRebalance || priorityBackupAvailable) && !disposed) {
147                        result = doReconnect();
148                        buildBackup = false;
149                    }
150                }
151                if (buildBackup) {
152                    buildBackups();
153                    if (priorityBackup && !connectedToPriority) {
154                        try {
155                            doDelay();
156                            if (reconnectTask == null) {
157                                return true;
158                            }
159                            reconnectTask.wakeup();
160                        } catch (InterruptedException e) {
161                            LOG.debug("Reconnect task has been interrupted.", e);
162                        }
163                    }
164                } else {
165                    // build backups on the next iteration
166                    buildBackup = true;
167                    try {
168                        if (reconnectTask == null) {
169                            return true;
170                        }
171                        reconnectTask.wakeup();
172                    } catch (InterruptedException e) {
173                        LOG.debug("Reconnect task has been interrupted.", e);
174                    }
175                }
176                return result;
177            }
178
179        }, "ActiveMQ Failover Worker: " + System.identityHashCode(this));
180    }
181
182    private void processCommand(Object incoming) {
183        Command command = (Command) incoming;
184        if (command == null) {
185            return;
186        }
187        if (command.isResponse()) {
188            Object object = null;
189            synchronized (requestMap) {
190                object = requestMap.remove(Integer.valueOf(((Response) command).getCorrelationId()));
191            }
192            if (object != null && object.getClass() == Tracked.class) {
193                ((Tracked) object).onResponses(command);
194            }
195        }
196
197        if (command.isConnectionControl()) {
198            handleConnectionControl((ConnectionControl) command);
199        } else if (command.isConsumerControl()) {
200            ConsumerControl consumerControl = (ConsumerControl)command;
201            if (consumerControl.isClose()) {
202                stateTracker.processRemoveConsumer(consumerControl.getConsumerId(), RemoveInfo.LAST_DELIVERED_UNKNOWN);
203            }
204        }
205
206        if (transportListener != null) {
207            transportListener.onCommand(command);
208        }
209    }
210
211    private TransportListener createTransportListener(final Transport owner) {
212        return new TransportListener() {
213
214            @Override
215            public void onCommand(Object o) {
216                processCommand(o);
217            }
218
219            @Override
220            public void onException(IOException error) {
221                try {
222                    handleTransportFailure(owner, error);
223                } catch (InterruptedException e) {
224                    Thread.currentThread().interrupt();
225                    if (transportListener != null) {
226                        transportListener.onException(new InterruptedIOException());
227                    }
228                }
229            }
230
231            @Override
232            public void transportInterupted() {
233            }
234
235            @Override
236            public void transportResumed() {
237            }
238        };
239    }
240
241    public final void disposeTransport(Transport transport) {
242        transport.setTransportListener(disposedListener);
243        ServiceSupport.dispose(transport);
244    }
245
246    public final void handleTransportFailure(IOException e) throws InterruptedException {
247        handleTransportFailure(getConnectedTransport(), e);
248    }
249
250    public final void handleTransportFailure(Transport failed, IOException e) throws InterruptedException {
251        if (shuttingDown) {
252            // shutdown info sent and remote socket closed and we see that before a local close
253            // let the close do the work
254            return;
255        }
256
257        if (LOG.isTraceEnabled()) {
258            LOG.trace(this + " handleTransportFailure: " + e, e);
259        }
260
261        // could be blocked in write with the reconnectMutex held, but still needs to be whacked
262        Transport transport = null;
263
264        if (connectedTransport.compareAndSet(failed, null)) {
265            transport = failed;
266            if (transport != null) {
267                disposeTransport(transport);
268            }
269        }
270
271        synchronized (reconnectMutex) {
272            if (transport != null && connectedTransport.get() == null) {
273                boolean reconnectOk = false;
274
275                if (canReconnect()) {
276                    reconnectOk = true;
277                }
278
279                LOG.warn("Transport ({}) failed {} attempting to automatically reconnect: {}",
280                         connectedTransportURI, (reconnectOk ? "," : ", not"), e);
281
282                failedConnectTransportURI = connectedTransportURI;
283                connectedTransportURI = null;
284                connectedToPriority = false;
285
286                if (reconnectOk) {
287                    // notify before any reconnect attempt so ack state can be whacked
288                    if (transportListener != null) {
289                        transportListener.transportInterupted();
290                    }
291
292                    updated.remove(failedConnectTransportURI);
293                    reconnectTask.wakeup();
294                } else if (!isDisposed()) {
295                    propagateFailureToExceptionListener(e);
296                }
297            }
298        }
299    }
300
301    private boolean canReconnect() {
302        return started && 0 != calculateReconnectAttemptLimit();
303    }
304
305    public final void handleConnectionControl(ConnectionControl control) {
306        String reconnectStr = control.getReconnectTo();
307        if (LOG.isTraceEnabled()) {
308            LOG.trace("Received ConnectionControl: {}", control);
309        }
310
311        if (reconnectStr != null) {
312            reconnectStr = reconnectStr.trim();
313            if (reconnectStr.length() > 0) {
314                try {
315                    URI uri = new URI(reconnectStr);
316                    if (isReconnectSupported()) {
317                        reconnect(uri);
318                        LOG.info("Reconnected to: " + uri);
319                    }
320                } catch (Exception e) {
321                    LOG.error("Failed to handle ConnectionControl reconnect to " + reconnectStr, e);
322                }
323            }
324        }
325        processNewTransports(control.isRebalanceConnection(), control.getConnectedBrokers());
326    }
327
328    private final void processNewTransports(boolean rebalance, String newTransports) {
329        if (newTransports != null) {
330            newTransports = newTransports.trim();
331            if (newTransports.length() > 0 && isUpdateURIsSupported()) {
332                List<URI> list = new ArrayList<URI>();
333                StringTokenizer tokenizer = new StringTokenizer(newTransports, ",");
334                while (tokenizer.hasMoreTokens()) {
335                    String str = tokenizer.nextToken();
336                    try {
337                        URI uri = new URI(str);
338                        list.add(uri);
339                    } catch (Exception e) {
340                        LOG.error("Failed to parse broker address: " + str, e);
341                    }
342                }
343                if (list.isEmpty() == false) {
344                    try {
345                        updateURIs(rebalance, list.toArray(new URI[list.size()]));
346                    } catch (IOException e) {
347                        LOG.error("Failed to update transport URI's from: " + newTransports, e);
348                    }
349                }
350            }
351        }
352    }
353
354    @Override
355    public void start() throws Exception {
356        synchronized (reconnectMutex) {
357            LOG.debug("Started {}", this);
358            if (started) {
359                return;
360            }
361            started = true;
362            stateTracker.setMaxCacheSize(getMaxCacheSize());
363            stateTracker.setTrackMessages(isTrackMessages());
364            stateTracker.setTrackTransactionProducers(isTrackTransactionProducers());
365            if (connectedTransport.get() != null) {
366                stateTracker.restore(connectedTransport.get());
367            } else {
368                reconnect(false);
369            }
370        }
371    }
372
373    @Override
374    public void stop() throws Exception {
375        Transport transportToStop = null;
376        List<Transport> backupsToStop = new ArrayList<Transport>(backups.size());
377
378        try {
379            synchronized (reconnectMutex) {
380                if (LOG.isDebugEnabled()) {
381                    LOG.debug("Stopped {}", this);
382                }
383                if (!started) {
384                    return;
385                }
386                started = false;
387                disposed = true;
388
389                if (connectedTransport.get() != null) {
390                    transportToStop = connectedTransport.getAndSet(null);
391                }
392                reconnectMutex.notifyAll();
393            }
394            synchronized (sleepMutex) {
395                sleepMutex.notifyAll();
396            }
397        } finally {
398            reconnectTask.shutdown();
399            reconnectTaskFactory.shutdownNow();
400        }
401
402        synchronized(backupMutex) {
403            for (BackupTransport backup : backups) {
404                backup.setDisposed(true);
405                Transport transport = backup.getTransport();
406                if (transport != null) {
407                    transport.setTransportListener(disposedListener);
408                    backupsToStop.add(transport);
409                }
410            }
411            backups.clear();
412        }
413        for (Transport transport : backupsToStop) {
414            try {
415                LOG.trace("Stopped backup: {}", transport);
416                disposeTransport(transport);
417            } catch (Exception e) {
418            }
419        }
420        if (transportToStop != null) {
421            transportToStop.stop();
422        }
423    }
424
425    public long getInitialReconnectDelay() {
426        return initialReconnectDelay;
427    }
428
429    public void setInitialReconnectDelay(long initialReconnectDelay) {
430        this.initialReconnectDelay = initialReconnectDelay;
431    }
432
433    public long getMaxReconnectDelay() {
434        return maxReconnectDelay;
435    }
436
437    public void setMaxReconnectDelay(long maxReconnectDelay) {
438        this.maxReconnectDelay = maxReconnectDelay;
439    }
440
441    public long getReconnectDelay() {
442        return reconnectDelay;
443    }
444
445    public void setReconnectDelay(long reconnectDelay) {
446        this.reconnectDelay = reconnectDelay;
447    }
448
449    public double getReconnectDelayExponent() {
450        return backOffMultiplier;
451    }
452
453    public void setReconnectDelayExponent(double reconnectDelayExponent) {
454        this.backOffMultiplier = reconnectDelayExponent;
455    }
456
457    public Transport getConnectedTransport() {
458        return connectedTransport.get();
459    }
460
461    public URI getConnectedTransportURI() {
462        return connectedTransportURI;
463    }
464
465    public int getMaxReconnectAttempts() {
466        return maxReconnectAttempts;
467    }
468
469    public void setMaxReconnectAttempts(int maxReconnectAttempts) {
470        this.maxReconnectAttempts = maxReconnectAttempts;
471    }
472
473    public int getStartupMaxReconnectAttempts() {
474        return this.startupMaxReconnectAttempts;
475    }
476
477    public void setStartupMaxReconnectAttempts(int startupMaxReconnectAttempts) {
478        this.startupMaxReconnectAttempts = startupMaxReconnectAttempts;
479    }
480
481    public long getTimeout() {
482        return timeout;
483    }
484
485    public void setTimeout(long timeout) {
486        this.timeout = timeout;
487    }
488
489    /**
490     * @return Returns the randomize.
491     */
492    public boolean isRandomize() {
493        return randomize;
494    }
495
496    /**
497     * @param randomize The randomize to set.
498     */
499    public void setRandomize(boolean randomize) {
500        this.randomize = randomize;
501    }
502
503    public boolean isBackup() {
504        return backup;
505    }
506
507    public void setBackup(boolean backup) {
508        this.backup = backup;
509    }
510
511    public int getBackupPoolSize() {
512        return backupPoolSize;
513    }
514
515    public void setBackupPoolSize(int backupPoolSize) {
516        this.backupPoolSize = backupPoolSize;
517    }
518
519    public int getCurrentBackups() {
520        return this.backups.size();
521    }
522
523    public boolean isTrackMessages() {
524        return trackMessages;
525    }
526
527    public void setTrackMessages(boolean trackMessages) {
528        this.trackMessages = trackMessages;
529    }
530
531    public boolean isTrackTransactionProducers() {
532        return this.trackTransactionProducers;
533    }
534
535    public void setTrackTransactionProducers(boolean trackTransactionProducers) {
536        this.trackTransactionProducers = trackTransactionProducers;
537    }
538
539    public int getMaxCacheSize() {
540        return maxCacheSize;
541    }
542
543    public void setMaxCacheSize(int maxCacheSize) {
544        this.maxCacheSize = maxCacheSize;
545    }
546
547    public boolean isPriorityBackup() {
548        return priorityBackup;
549    }
550
551    public void setPriorityBackup(boolean priorityBackup) {
552        this.priorityBackup = priorityBackup;
553    }
554
555    public void setPriorityURIs(String priorityURIs) {
556        StringTokenizer tokenizer = new StringTokenizer(priorityURIs, ",");
557        while (tokenizer.hasMoreTokens()) {
558            String str = tokenizer.nextToken();
559            try {
560                URI uri = new URI(str);
561                priorityList.add(uri);
562            } catch (Exception e) {
563                LOG.error("Failed to parse broker address: " + str, e);
564            }
565        }
566    }
567
568    @Override
569    public void oneway(Object o) throws IOException {
570
571        Command command = (Command) o;
572        Exception error = null;
573        try {
574
575            synchronized (reconnectMutex) {
576
577                if (command != null && connectedTransport.get() == null) {
578                    if (command.isShutdownInfo()) {
579                        // Skipping send of ShutdownInfo command when not connected.
580                        return;
581                    } else if (command instanceof RemoveInfo || command.isMessageAck()) {
582                        // Simulate response to RemoveInfo command or MessageAck (as it will be stale)
583                        stateTracker.track(command);
584                        if (command.isResponseRequired()) {
585                            Response response = new Response();
586                            response.setCorrelationId(command.getCommandId());
587                            processCommand(response);
588                        }
589                        return;
590                    } else if (command instanceof MessagePull) {
591                        // Simulate response to MessagePull if timed as we can't honor that now.
592                        MessagePull pullRequest = (MessagePull) command;
593                        if (pullRequest.getTimeout() != 0) {
594                            MessageDispatch dispatch = new MessageDispatch();
595                            dispatch.setConsumerId(pullRequest.getConsumerId());
596                            dispatch.setDestination(pullRequest.getDestination());
597                            processCommand(dispatch);
598                        }
599                        return;
600                    }
601                }
602
603                // Keep trying until the message is sent.
604                for (int i = 0; !disposed; i++) {
605                    try {
606
607                        // Wait for transport to be connected.
608                        Transport transport = connectedTransport.get();
609                        long start = System.currentTimeMillis();
610                        boolean timedout = false;
611                        while (transport == null && !disposed && connectionFailure == null
612                                && !Thread.currentThread().isInterrupted() && willReconnect()) {
613
614                            LOG.trace("Waiting for transport to reconnect..: {}", command);
615                            long end = System.currentTimeMillis();
616                            if (command.isMessage() && timeout > 0 && (end - start > timeout)) {
617                                timedout = true;
618                                LOG.info("Failover timed out after {} ms", (end - start));
619                                break;
620                            }
621                            try {
622                                reconnectMutex.wait(100);
623                            } catch (InterruptedException e) {
624                                Thread.currentThread().interrupt();
625                                LOG.debug("Interupted:", e);
626                            }
627                            transport = connectedTransport.get();
628                        }
629
630                        if (transport == null) {
631                            // Previous loop may have exited due to use being
632                            // disposed.
633                            if (disposed) {
634                                error = new IOException("Transport disposed.");
635                            } else if (connectionFailure != null) {
636                                error = connectionFailure;
637                            } else if (timedout == true) {
638                                error = new IOException("Failover timeout of " + timeout + " ms reached.");
639                            } else if (!willReconnect()) {
640                                error = new IOException("Reconnect attempts of " + maxReconnectAttempts + " exceeded");
641                            } else {
642                                error = new IOException("Unexpected failure.");
643                            }
644                            break;
645                        }
646
647                        Tracked tracked = null;
648                        try {
649                            tracked = stateTracker.track(command);
650                        } catch (IOException ioe) {
651                            LOG.debug("Cannot track the command {} {}", command, ioe);
652                        }
653                        // If it was a request and it was not being tracked by
654                        // the state tracker,
655                        // then hold it in the requestMap so that we can replay
656                        // it later.
657                        synchronized (requestMap) {
658                            if (tracked != null && tracked.isWaitingForResponse()) {
659                                requestMap.put(Integer.valueOf(command.getCommandId()), tracked);
660                            } else if (tracked == null && command.isResponseRequired()) {
661                                requestMap.put(Integer.valueOf(command.getCommandId()), command);
662                            }
663                        }
664
665                        // Send the message.
666                        try {
667                            transport.oneway(command);
668                            stateTracker.trackBack(command);
669                            if (command.isShutdownInfo()) {
670                                shuttingDown = true;
671                            }
672                        } catch (IOException e) {
673
674                            // If the command was not tracked.. we will retry in
675                            // this method
676                            if (tracked == null && canReconnect()) {
677
678                                // since we will retry in this method.. take it
679                                // out of the request
680                                // map so that it is not sent 2 times on
681                                // recovery
682                                if (command.isResponseRequired()) {
683                                    requestMap.remove(Integer.valueOf(command.getCommandId()));
684                                }
685
686                                // Rethrow the exception so it will handled by
687                                // the outer catch
688                                throw e;
689                            } else {
690                                // Handle the error but allow the method to return since the
691                                // tracked commands are replayed on reconnect.
692                                LOG.debug("Send oneway attempt: {} failed for command: {}", i, command);
693                                handleTransportFailure(e);
694                            }
695                        }
696
697                        return;
698                    } catch (IOException e) {
699                        LOG.debug("Send oneway attempt: {} failed for command: {}", i, command);
700                        handleTransportFailure(e);
701                    }
702                }
703            }
704        } catch (InterruptedException e) {
705            // Some one may be trying to stop our thread.
706            Thread.currentThread().interrupt();
707            throw new InterruptedIOException();
708        }
709
710        if (!disposed) {
711            if (error != null) {
712                if (error instanceof IOException) {
713                    throw (IOException) error;
714                }
715                throw IOExceptionSupport.create(error);
716            }
717        }
718    }
719
720    private boolean willReconnect() {
721        return firstConnection || 0 != calculateReconnectAttemptLimit();
722    }
723
724    @Override
725    public FutureResponse asyncRequest(Object command, ResponseCallback responseCallback) throws IOException {
726        throw new AssertionError("Unsupported Method");
727    }
728
729    @Override
730    public Object request(Object command) throws IOException {
731        throw new AssertionError("Unsupported Method");
732    }
733
734    @Override
735    public Object request(Object command, int timeout) throws IOException {
736        throw new AssertionError("Unsupported Method");
737    }
738
739    @Override
740    public void add(boolean rebalance, URI u[]) {
741        boolean newURI = false;
742        for (URI uri : u) {
743            if (!contains(uri)) {
744                uris.add(uri);
745                newURI = true;
746            }
747        }
748        if (newURI) {
749            reconnect(rebalance);
750        }
751    }
752
753    @Override
754    public void remove(boolean rebalance, URI u[]) {
755        for (URI uri : u) {
756            uris.remove(uri);
757        }
758        // rebalance is automatic if any connected to removed/stopped broker
759    }
760
761    public void add(boolean rebalance, String u) {
762        try {
763            URI newURI = new URI(u);
764            if (contains(newURI) == false) {
765                uris.add(newURI);
766                reconnect(rebalance);
767            }
768
769        } catch (Exception e) {
770            LOG.error("Failed to parse URI: {}", u);
771        }
772    }
773
774    public void reconnect(boolean rebalance) {
775        synchronized (reconnectMutex) {
776            if (started) {
777                if (rebalance) {
778                    doRebalance = true;
779                }
780                LOG.debug("Waking up reconnect task");
781                try {
782                    reconnectTask.wakeup();
783                } catch (InterruptedException e) {
784                    Thread.currentThread().interrupt();
785                }
786            } else {
787                LOG.debug("Reconnect was triggered but transport is not started yet. Wait for start to connect the transport.");
788            }
789        }
790    }
791
792    private List<URI> getConnectList() {
793        if (!updated.isEmpty()) {
794            return updated;
795        }
796        ArrayList<URI> l = new ArrayList<URI>(uris);
797        boolean removed = false;
798        if (failedConnectTransportURI != null) {
799            removed = l.remove(failedConnectTransportURI);
800        }
801        if (randomize) {
802            // Randomly, reorder the list by random swapping
803            for (int i = 0; i < l.size(); i++) {
804                // meed parenthesis due other JDKs (see AMQ-4826)
805                int p = ((int) (Math.random() * 100)) % l.size();
806                URI t = l.get(p);
807                l.set(p, l.get(i));
808                l.set(i, t);
809            }
810        }
811        if (removed) {
812            l.add(failedConnectTransportURI);
813        }
814
815        LOG.debug("urlList connectionList:{}, from: {}", l, uris);
816
817        return l;
818    }
819
820    @Override
821    public TransportListener getTransportListener() {
822        return transportListener;
823    }
824
825    @Override
826    public void setTransportListener(TransportListener commandListener) {
827        synchronized (listenerMutex) {
828            this.transportListener = commandListener;
829            listenerMutex.notifyAll();
830        }
831    }
832
833    @Override
834    public <T> T narrow(Class<T> target) {
835
836        if (target.isAssignableFrom(getClass())) {
837            return target.cast(this);
838        }
839        Transport transport = connectedTransport.get();
840        if (transport != null) {
841            return transport.narrow(target);
842        }
843        return null;
844
845    }
846
847    protected void restoreTransport(Transport t) throws Exception, IOException {
848        t.start();
849        // send information to the broker - informing it we are an ft client
850        ConnectionControl cc = new ConnectionControl();
851        cc.setFaultTolerant(true);
852        t.oneway(cc);
853        stateTracker.restore(t);
854        Map<Integer, Command> tmpMap = null;
855        synchronized (requestMap) {
856            tmpMap = new LinkedHashMap<Integer, Command>(requestMap);
857        }
858        for (Command command : tmpMap.values()) {
859            LOG.trace("restore requestMap, replay: {}", command);
860            t.oneway(command);
861        }
862    }
863
864    public boolean isUseExponentialBackOff() {
865        return useExponentialBackOff;
866    }
867
868    public void setUseExponentialBackOff(boolean useExponentialBackOff) {
869        this.useExponentialBackOff = useExponentialBackOff;
870    }
871
872    @Override
873    public String toString() {
874        return connectedTransportURI == null ? "unconnected" : connectedTransportURI.toString();
875    }
876
877    @Override
878    public String getRemoteAddress() {
879        Transport transport = connectedTransport.get();
880        if (transport != null) {
881            return transport.getRemoteAddress();
882        }
883        return null;
884    }
885
886    @Override
887    public boolean isFaultTolerant() {
888        return true;
889    }
890
891    private void doUpdateURIsFromDisk() {
892        // If updateURIsURL is specified, read the file and add any new
893        // transport URI's to this FailOverTransport.
894        // Note: Could track file timestamp to avoid unnecessary reading.
895        String fileURL = getUpdateURIsURL();
896        if (fileURL != null) {
897            BufferedReader in = null;
898            String newUris = null;
899            StringBuffer buffer = new StringBuffer();
900
901            try {
902                in = new BufferedReader(getURLStream(fileURL));
903                while (true) {
904                    String line = in.readLine();
905                    if (line == null) {
906                        break;
907                    }
908                    buffer.append(line);
909                }
910                newUris = buffer.toString();
911            } catch (IOException ioe) {
912                LOG.error("Failed to read updateURIsURL: {} {}",fileURL, ioe);
913            } finally {
914                if (in != null) {
915                    try {
916                        in.close();
917                    } catch (IOException ioe) {
918                        // ignore
919                    }
920                }
921            }
922
923            processNewTransports(isRebalanceUpdateURIs(), newUris);
924        }
925    }
926
927    final boolean doReconnect() {
928        Exception failure = null;
929        synchronized (reconnectMutex) {
930
931            // First ensure we are up to date.
932            doUpdateURIsFromDisk();
933
934            if (disposed || connectionFailure != null) {
935                reconnectMutex.notifyAll();
936            }
937            if ((connectedTransport.get() != null && !doRebalance && !priorityBackupAvailable) || disposed || connectionFailure != null) {
938                return false;
939            } else {
940                List<URI> connectList = getConnectList();
941                if (connectList.isEmpty()) {
942                    failure = new IOException("No uris available to connect to.");
943                } else {
944                    if (doRebalance) {
945                        if (connectedToPriority || compareURIs(connectList.get(0), connectedTransportURI)) {
946                            // already connected to first in the list, no need to rebalance
947                            doRebalance = false;
948                            return false;
949                        } else {
950                            LOG.debug("Doing rebalance from: {} to {}", connectedTransportURI, connectList);
951
952                            try {
953                                Transport transport = this.connectedTransport.getAndSet(null);
954                                if (transport != null) {
955                                    disposeTransport(transport);
956                                }
957                            } catch (Exception e) {
958                                LOG.debug("Caught an exception stopping existing transport for rebalance", e);
959                            }
960                        }
961                        doRebalance = false;
962                    }
963
964                    resetReconnectDelay();
965
966                    Transport transport = null;
967                    URI uri = null;
968
969                    // If we have a backup already waiting lets try it.
970                    synchronized (backupMutex) {
971                        if ((priorityBackup || backup) && !backups.isEmpty()) {
972                            ArrayList<BackupTransport> l = new ArrayList<BackupTransport>(backups);
973                            if (randomize) {
974                                Collections.shuffle(l);
975                            }
976                            BackupTransport bt = l.remove(0);
977                            backups.remove(bt);
978                            transport = bt.getTransport();
979                            uri = bt.getUri();
980                            processCommand(bt.getBrokerInfo());
981                            if (priorityBackup && priorityBackupAvailable) {
982                                Transport old = this.connectedTransport.getAndSet(null);
983                                if (old != null) {
984                                    disposeTransport(old);
985                                }
986                                priorityBackupAvailable = false;
987                            }
988                        }
989                    }
990
991                    // When there was no backup and we are reconnecting for the first time
992                    // we honor the initialReconnectDelay before trying a new connection, after
993                    // this normal reconnect delay happens following a failed attempt.
994                    if (transport == null && !firstConnection && connectFailures == 0 && initialReconnectDelay > 0 && !disposed) {
995                        // reconnectDelay will be equal to initialReconnectDelay since we are on
996                        // the first connect attempt after we had a working connection, doDelay
997                        // will apply updates to move to the next reconnectDelay value based on
998                        // configuration.
999                        doDelay();
1000                    }
1001
1002                    Iterator<URI> iter = connectList.iterator();
1003                    while ((transport != null || iter.hasNext()) && (connectedTransport.get() == null && !disposed)) {
1004
1005                        try {
1006                            SslContext.setCurrentSslContext(brokerSslContext);
1007
1008                            // We could be starting with a backup and if so we wait to grab a
1009                            // URI from the pool until next time around.
1010                            if (transport == null) {
1011                                uri = addExtraQueryOptions(iter.next());
1012                                transport = TransportFactory.compositeConnect(uri);
1013                            }
1014
1015                            LOG.debug("Attempting {}th connect to: {}", connectFailures, uri);
1016
1017                            transport.setTransportListener(createTransportListener(transport));
1018                            transport.start();
1019
1020                            if (started && !firstConnection) {
1021                                restoreTransport(transport);
1022                            }
1023
1024                            LOG.debug("Connection established");
1025
1026                            reconnectDelay = initialReconnectDelay;
1027                            connectedTransportURI = uri;
1028                            connectedTransport.set(transport);
1029                            connectedToPriority = isPriority(connectedTransportURI);
1030                            reconnectMutex.notifyAll();
1031                            connectFailures = 0;
1032
1033                            // Make sure on initial startup, that the transportListener
1034                            // has been initialized for this instance.
1035                            synchronized (listenerMutex) {
1036                                if (transportListener == null) {
1037                                    try {
1038                                        // if it isn't set after 2secs - it probably never will be
1039                                        listenerMutex.wait(2000);
1040                                    } catch (InterruptedException ex) {
1041                                    }
1042                                }
1043                            }
1044
1045                            if (transportListener != null) {
1046                                transportListener.transportResumed();
1047                            } else {
1048                                LOG.debug("transport resumed by transport listener not set");
1049                            }
1050
1051                            if (firstConnection) {
1052                                firstConnection = false;
1053                                LOG.info("Successfully connected to {}", uri);
1054                            } else {
1055                                LOG.info("Successfully reconnected to {}", uri);
1056                            }
1057
1058                            return false;
1059                        } catch (Exception e) {
1060                            failure = e;
1061                            LOG.debug("Connect fail to: {}, reason: {}", uri, e);
1062                            if (transport != null) {
1063                                try {
1064                                    transport.stop();
1065                                    transport = null;
1066                                } catch (Exception ee) {
1067                                    LOG.debug("Stop of failed transport: {} failed with reason: {}", transport, ee);
1068                                }
1069                            }
1070                        } finally {
1071                            SslContext.setCurrentSslContext(null);
1072                        }
1073                    }
1074                }
1075            }
1076
1077            int reconnectLimit = calculateReconnectAttemptLimit();
1078
1079            connectFailures++;
1080            if (reconnectLimit != INFINITE && connectFailures >= reconnectLimit) {
1081                LOG.error("Failed to connect to {} after: {} attempt(s)", uris, connectFailures);
1082                connectionFailure = failure;
1083
1084                // Make sure on initial startup, that the transportListener has been
1085                // initialized for this instance.
1086                synchronized (listenerMutex) {
1087                    if (transportListener == null) {
1088                        try {
1089                            listenerMutex.wait(2000);
1090                        } catch (InterruptedException ex) {
1091                        }
1092                    }
1093                }
1094
1095                propagateFailureToExceptionListener(connectionFailure);
1096                return false;
1097            }
1098
1099            int warnInterval = getWarnAfterReconnectAttempts();
1100            if (warnInterval > 0 && (connectFailures % warnInterval) == 0) {
1101                LOG.warn("Failed to connect to {} after: {} attempt(s) continuing to retry.",
1102                         uris, connectFailures);
1103            }
1104        }
1105
1106        if (!disposed) {
1107            doDelay();
1108        }
1109
1110        return !disposed;
1111    }
1112
1113    private void doDelay() {
1114        if (reconnectDelay > 0) {
1115            synchronized (sleepMutex) {
1116                LOG.debug("Waiting {} ms before attempting connection", reconnectDelay);
1117                try {
1118                    sleepMutex.wait(reconnectDelay);
1119                } catch (InterruptedException e) {
1120                    Thread.currentThread().interrupt();
1121                }
1122            }
1123        }
1124
1125        if (useExponentialBackOff) {
1126            // Exponential increment of reconnect delay.
1127            reconnectDelay *= backOffMultiplier;
1128            if (reconnectDelay > maxReconnectDelay) {
1129                reconnectDelay = maxReconnectDelay;
1130            }
1131        }
1132    }
1133
1134    private void resetReconnectDelay() {
1135        if (!useExponentialBackOff || reconnectDelay == DEFAULT_INITIAL_RECONNECT_DELAY) {
1136            reconnectDelay = initialReconnectDelay;
1137        }
1138    }
1139
1140    /*
1141      * called with reconnectMutex held
1142     */
1143    private void propagateFailureToExceptionListener(Exception exception) {
1144        if (transportListener != null) {
1145            if (exception instanceof IOException) {
1146                transportListener.onException((IOException)exception);
1147            } else {
1148                transportListener.onException(IOExceptionSupport.create(exception));
1149            }
1150        }
1151        reconnectMutex.notifyAll();
1152    }
1153
1154    private int calculateReconnectAttemptLimit() {
1155        int maxReconnectValue = this.maxReconnectAttempts;
1156        if (firstConnection && this.startupMaxReconnectAttempts != INFINITE) {
1157            maxReconnectValue = this.startupMaxReconnectAttempts;
1158        }
1159        return maxReconnectValue;
1160    }
1161
1162    private boolean shouldBuildBackups() {
1163       return (backup && backups.size() < backupPoolSize) || (priorityBackup && !(priorityBackupAvailable || connectedToPriority));
1164    }
1165
1166    final boolean buildBackups() {
1167        synchronized (backupMutex) {
1168            if (!disposed && shouldBuildBackups()) {
1169                ArrayList<URI> backupList = new ArrayList<URI>(priorityList);
1170                List<URI> connectList = getConnectList();
1171                for (URI uri: connectList) {
1172                    if (!backupList.contains(uri)) {
1173                        backupList.add(uri);
1174                    }
1175                }
1176                // removed disposed backups
1177                List<BackupTransport> disposedList = new ArrayList<BackupTransport>();
1178                for (BackupTransport bt : backups) {
1179                    if (bt.isDisposed()) {
1180                        disposedList.add(bt);
1181                    }
1182                }
1183                backups.removeAll(disposedList);
1184                disposedList.clear();
1185                for (Iterator<URI> iter = backupList.iterator(); !disposed && iter.hasNext() && shouldBuildBackups(); ) {
1186                    URI uri = addExtraQueryOptions(iter.next());
1187                    if (connectedTransportURI != null && !connectedTransportURI.equals(uri)) {
1188                        try {
1189                            SslContext.setCurrentSslContext(brokerSslContext);
1190                            BackupTransport bt = new BackupTransport(this);
1191                            bt.setUri(uri);
1192                            if (!backups.contains(bt)) {
1193                                Transport t = TransportFactory.compositeConnect(uri);
1194                                t.setTransportListener(bt);
1195                                t.start();
1196                                bt.setTransport(t);
1197                                if (priorityBackup && isPriority(uri)) {
1198                                   priorityBackupAvailable = true;
1199                                   backups.add(0, bt);
1200                                   // if this priority backup overflows the pool
1201                                   // remove the backup with the lowest priority
1202                                   if (backups.size() > backupPoolSize) {
1203                                       BackupTransport disposeTransport = backups.remove(backups.size() - 1);
1204                                       disposeTransport.setDisposed(true);
1205                                       Transport transport = disposeTransport.getTransport();
1206                                       if (transport != null) {
1207                                           transport.setTransportListener(disposedListener);
1208                                           disposeTransport(transport);
1209                                       }
1210                                   }
1211                                } else {
1212                                    backups.add(bt);
1213                                }
1214                            }
1215                        } catch (Exception e) {
1216                            LOG.debug("Failed to build backup ", e);
1217                        } finally {
1218                            SslContext.setCurrentSslContext(null);
1219                        }
1220                    }
1221                }
1222            }
1223        }
1224        return false;
1225    }
1226
1227    protected boolean isPriority(URI uri) {
1228        if (!priorityBackup) {
1229            return false;
1230        }
1231
1232        if (!priorityList.isEmpty()) {
1233            return priorityList.contains(uri);
1234        }
1235        return uris.indexOf(uri) == 0;
1236    }
1237
1238    @Override
1239    public boolean isDisposed() {
1240        return disposed;
1241    }
1242
1243    @Override
1244    public boolean isConnected() {
1245        return connectedTransport.get() != null;
1246    }
1247
1248    @Override
1249    public void reconnect(URI uri) throws IOException {
1250        add(true, new URI[]{uri});
1251    }
1252
1253    @Override
1254    public boolean isReconnectSupported() {
1255        return this.reconnectSupported;
1256    }
1257
1258    public void setReconnectSupported(boolean value) {
1259        this.reconnectSupported = value;
1260    }
1261
1262    @Override
1263    public boolean isUpdateURIsSupported() {
1264        return this.updateURIsSupported;
1265    }
1266
1267    public void setUpdateURIsSupported(boolean value) {
1268        this.updateURIsSupported = value;
1269    }
1270
1271    @Override
1272    public void updateURIs(boolean rebalance, URI[] updatedURIs) throws IOException {
1273        if (isUpdateURIsSupported()) {
1274            HashSet<URI> copy = new HashSet<URI>();
1275            synchronized (reconnectMutex) {
1276                copy.addAll(this.updated);
1277                updated.clear();
1278                if (updatedURIs != null && updatedURIs.length > 0) {
1279                    for (URI uri : updatedURIs) {
1280                        if (uri != null && !updated.contains(uri)) {
1281                            updated.add(uri);
1282                        }
1283                    }
1284                }
1285            }
1286            if (!(copy.isEmpty() && updated.isEmpty()) && !copy.equals(new HashSet<URI>(updated))) {
1287                buildBackups();
1288                reconnect(rebalance);
1289            }
1290        }
1291    }
1292
1293    /**
1294     * @return the updateURIsURL
1295     */
1296    public String getUpdateURIsURL() {
1297        return this.updateURIsURL;
1298    }
1299
1300    /**
1301     * @param updateURIsURL the updateURIsURL to set
1302     */
1303    public void setUpdateURIsURL(String updateURIsURL) {
1304        this.updateURIsURL = updateURIsURL;
1305    }
1306
1307    /**
1308     * @return the rebalanceUpdateURIs
1309     */
1310    public boolean isRebalanceUpdateURIs() {
1311        return this.rebalanceUpdateURIs;
1312    }
1313
1314    /**
1315     * @param rebalanceUpdateURIs the rebalanceUpdateURIs to set
1316     */
1317    public void setRebalanceUpdateURIs(boolean rebalanceUpdateURIs) {
1318        this.rebalanceUpdateURIs = rebalanceUpdateURIs;
1319    }
1320
1321    @Override
1322    public int getReceiveCounter() {
1323        Transport transport = connectedTransport.get();
1324        if (transport == null) {
1325            return 0;
1326        }
1327        return transport.getReceiveCounter();
1328    }
1329
1330    public int getConnectFailures() {
1331        return connectFailures;
1332    }
1333
1334    public void connectionInterruptProcessingComplete(ConnectionId connectionId) {
1335        synchronized (reconnectMutex) {
1336            stateTracker.connectionInterruptProcessingComplete(this, connectionId);
1337        }
1338    }
1339
1340    public ConnectionStateTracker getStateTracker() {
1341        return stateTracker;
1342    }
1343
1344    private boolean contains(URI newURI) {
1345        boolean result = false;
1346        for (URI uri : uris) {
1347            if (compareURIs(newURI, uri)) {
1348                result = true;
1349                break;
1350            }
1351        }
1352
1353        return result;
1354    }
1355
1356    private boolean compareURIs(final URI first, final URI second) {
1357
1358        boolean result = false;
1359        if (first == null || second == null) {
1360            return result;
1361        }
1362
1363        if (first.getPort() == second.getPort()) {
1364            InetAddress firstAddr = null;
1365            InetAddress secondAddr = null;
1366            try {
1367                firstAddr = InetAddress.getByName(first.getHost());
1368                secondAddr = InetAddress.getByName(second.getHost());
1369
1370                if (firstAddr.equals(secondAddr)) {
1371                    result = true;
1372                }
1373
1374            } catch(IOException e) {
1375
1376                if (firstAddr == null) {
1377                    LOG.error("Failed to Lookup INetAddress for URI[{}] : {}", first, e);
1378                } else {
1379                    LOG.error("Failed to Lookup INetAddress for URI[{}] : {}", second, e);
1380                }
1381
1382                if (first.getHost().equalsIgnoreCase(second.getHost())) {
1383                    result = true;
1384                }
1385            }
1386        }
1387
1388        return result;
1389    }
1390
1391    private InputStreamReader getURLStream(String path) throws IOException {
1392        InputStreamReader result = null;
1393        URL url = null;
1394        try {
1395            url = new URL(path);
1396            result = new InputStreamReader(url.openStream());
1397        } catch (MalformedURLException e) {
1398            // ignore - it could be a path to a a local file
1399        }
1400        if (result == null) {
1401            result = new FileReader(path);
1402        }
1403        return result;
1404    }
1405
1406    private URI addExtraQueryOptions(URI uri) {
1407        try {
1408            if( nestedExtraQueryOptions!=null && !nestedExtraQueryOptions.isEmpty() ) {
1409                if( uri.getQuery() == null ) {
1410                    uri = URISupport.createURIWithQuery(uri, nestedExtraQueryOptions);
1411                } else {
1412                    uri = URISupport.createURIWithQuery(uri, uri.getQuery()+"&"+nestedExtraQueryOptions);
1413                }
1414            }
1415        } catch (URISyntaxException e) {
1416            throw new RuntimeException(e);
1417        }
1418        return uri;
1419    }
1420
1421    public void setNestedExtraQueryOptions(String nestedExtraQueryOptions) {
1422        this.nestedExtraQueryOptions = nestedExtraQueryOptions;
1423    }
1424
1425    public int getWarnAfterReconnectAttempts() {
1426        return warnAfterReconnectAttempts;
1427    }
1428
1429    /**
1430     * Sets the number of Connect / Reconnect attempts that must occur before a warn message
1431     * is logged indicating that the transport is not connected.  This can be useful when the
1432     * client is running inside some container or service as it give an indication of some
1433     * problem with the client connection that might not otherwise be visible.  To disable the
1434     * log messages this value should be set to a value @{code attempts <= 0}
1435     *
1436     * @param warnAfterReconnectAttempts
1437     *      The number of failed connection attempts that must happen before a warning is logged.
1438     */
1439    public void setWarnAfterReconnectAttempts(int warnAfterReconnectAttempts) {
1440        this.warnAfterReconnectAttempts = warnAfterReconnectAttempts;
1441    }
1442
1443}