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 volatile 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 volatile 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 volatile 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}