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