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