001/** 002 * 003 * Copyright 2003-2007 Jive Software. 004 * 005 * Licensed under the Apache License, Version 2.0 (the "License"); 006 * you may not use this file except in compliance with the License. 007 * 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.jivesoftware.smack.tcp; 018 019import java.io.BufferedReader; 020import java.io.IOException; 021import java.io.InputStream; 022import java.io.InputStreamReader; 023import java.io.OutputStream; 024import java.io.OutputStreamWriter; 025import java.io.Writer; 026import java.net.InetAddress; 027import java.net.InetSocketAddress; 028import java.net.Socket; 029import java.security.cert.CertificateException; 030import java.util.ArrayList; 031import java.util.Collection; 032import java.util.Iterator; 033import java.util.LinkedHashSet; 034import java.util.LinkedList; 035import java.util.List; 036import java.util.Map; 037import java.util.Set; 038import java.util.concurrent.ArrayBlockingQueue; 039import java.util.concurrent.BlockingQueue; 040import java.util.concurrent.ConcurrentHashMap; 041import java.util.concurrent.ConcurrentLinkedQueue; 042import java.util.concurrent.TimeUnit; 043import java.util.concurrent.atomic.AtomicBoolean; 044import java.util.logging.Level; 045import java.util.logging.Logger; 046 047import javax.net.SocketFactory; 048import javax.net.ssl.HostnameVerifier; 049import javax.net.ssl.SSLSession; 050import javax.net.ssl.SSLSocket; 051 052import org.jivesoftware.smack.AbstractXMPPConnection; 053import org.jivesoftware.smack.ConnectionConfiguration; 054import org.jivesoftware.smack.ConnectionConfiguration.SecurityMode; 055import org.jivesoftware.smack.ConnectionListener; 056import org.jivesoftware.smack.SmackConfiguration; 057import org.jivesoftware.smack.SmackException; 058import org.jivesoftware.smack.SmackException.AlreadyConnectedException; 059import org.jivesoftware.smack.SmackException.AlreadyLoggedInException; 060import org.jivesoftware.smack.SmackException.ConnectionException; 061import org.jivesoftware.smack.SmackException.EndpointConnectionException; 062import org.jivesoftware.smack.SmackException.NotConnectedException; 063import org.jivesoftware.smack.SmackException.NotLoggedInException; 064import org.jivesoftware.smack.SmackException.SecurityNotPossibleException; 065import org.jivesoftware.smack.SmackException.SecurityRequiredByServerException; 066import org.jivesoftware.smack.SmackFuture; 067import org.jivesoftware.smack.StanzaListener; 068import org.jivesoftware.smack.XMPPConnection; 069import org.jivesoftware.smack.XMPPException; 070import org.jivesoftware.smack.XMPPException.FailedNonzaException; 071import org.jivesoftware.smack.XMPPException.StreamErrorException; 072import org.jivesoftware.smack.compress.packet.Compress; 073import org.jivesoftware.smack.compress.packet.Compressed; 074import org.jivesoftware.smack.compression.XMPPInputOutputStream; 075import org.jivesoftware.smack.datatypes.UInt16; 076import org.jivesoftware.smack.filter.StanzaFilter; 077import org.jivesoftware.smack.internal.SmackTlsContext; 078import org.jivesoftware.smack.packet.Element; 079import org.jivesoftware.smack.packet.IQ; 080import org.jivesoftware.smack.packet.Message; 081import org.jivesoftware.smack.packet.Nonza; 082import org.jivesoftware.smack.packet.Presence; 083import org.jivesoftware.smack.packet.Stanza; 084import org.jivesoftware.smack.packet.StartTls; 085import org.jivesoftware.smack.packet.StreamError; 086import org.jivesoftware.smack.packet.StreamOpen; 087import org.jivesoftware.smack.proxy.ProxyInfo; 088import org.jivesoftware.smack.sasl.packet.SaslNonza; 089import org.jivesoftware.smack.sm.SMUtils; 090import org.jivesoftware.smack.sm.StreamManagementException; 091import org.jivesoftware.smack.sm.StreamManagementException.StreamIdDoesNotMatchException; 092import org.jivesoftware.smack.sm.StreamManagementException.StreamManagementCounterError; 093import org.jivesoftware.smack.sm.StreamManagementException.StreamManagementNotEnabledException; 094import org.jivesoftware.smack.sm.packet.StreamManagement; 095import org.jivesoftware.smack.sm.packet.StreamManagement.AckAnswer; 096import org.jivesoftware.smack.sm.packet.StreamManagement.AckRequest; 097import org.jivesoftware.smack.sm.packet.StreamManagement.Enable; 098import org.jivesoftware.smack.sm.packet.StreamManagement.Enabled; 099import org.jivesoftware.smack.sm.packet.StreamManagement.Failed; 100import org.jivesoftware.smack.sm.packet.StreamManagement.Resume; 101import org.jivesoftware.smack.sm.packet.StreamManagement.Resumed; 102import org.jivesoftware.smack.sm.packet.StreamManagement.StreamManagementFeature; 103import org.jivesoftware.smack.sm.predicates.Predicate; 104import org.jivesoftware.smack.sm.provider.ParseStreamManagement; 105import org.jivesoftware.smack.tcp.rce.RemoteXmppTcpConnectionEndpoints; 106import org.jivesoftware.smack.tcp.rce.Rfc6120TcpRemoteConnectionEndpoint; 107import org.jivesoftware.smack.util.ArrayBlockingQueueWithShutdown; 108import org.jivesoftware.smack.util.Async; 109import org.jivesoftware.smack.util.CloseableUtil; 110import org.jivesoftware.smack.util.PacketParserUtils; 111import org.jivesoftware.smack.util.StringUtils; 112import org.jivesoftware.smack.util.TLSUtils; 113import org.jivesoftware.smack.util.XmlStringBuilder; 114import org.jivesoftware.smack.util.rce.RemoteConnectionException; 115import org.jivesoftware.smack.xml.SmackXmlParser; 116import org.jivesoftware.smack.xml.XmlPullParser; 117import org.jivesoftware.smack.xml.XmlPullParserException; 118 119import org.jxmpp.jid.impl.JidCreate; 120import org.jxmpp.jid.parts.Resourcepart; 121import org.jxmpp.stringprep.XmppStringprepException; 122import org.minidns.dnsname.DnsName; 123 124/** 125 * Creates a socket connection to an XMPP server. This is the default connection 126 * to an XMPP server and is specified in the XMPP Core (RFC 6120). 127 * 128 * @see XMPPConnection 129 * @author Matt Tucker 130 */ 131public class XMPPTCPConnection extends AbstractXMPPConnection { 132 133 private static final int QUEUE_SIZE = 500; 134 private static final Logger LOGGER = Logger.getLogger(XMPPTCPConnection.class.getName()); 135 136 /** 137 * The socket which is used for this connection. 138 */ 139 private Socket socket; 140 141 /** 142 * 143 */ 144 private boolean disconnectedButResumeable = false; 145 146 private SSLSocket secureSocket; 147 148 /** 149 * Protected access level because of unit test purposes 150 */ 151 protected final PacketWriter packetWriter = new PacketWriter(); 152 153 /** 154 * Protected access level because of unit test purposes 155 */ 156 protected final PacketReader packetReader = new PacketReader(); 157 158 /** 159 * 160 */ 161 private boolean streamFeaturesAfterAuthenticationReceived; 162 163 /** 164 * 165 */ 166 private boolean compressSyncPoint; 167 168 /** 169 * The default bundle and defer callback, used for new connections. 170 * @see bundleAndDeferCallback 171 */ 172 private static BundleAndDeferCallback defaultBundleAndDeferCallback; 173 174 /** 175 * The used bundle and defer callback. 176 * <p> 177 * Although this field may be set concurrently, the 'volatile' keyword was deliberately not added, in order to avoid 178 * having a 'volatile' read within the writer threads loop. 179 * </p> 180 */ 181 private BundleAndDeferCallback bundleAndDeferCallback = defaultBundleAndDeferCallback; 182 183 private static boolean useSmDefault = true; 184 185 private static boolean useSmResumptionDefault = true; 186 187 /** 188 * The stream ID of the stream that is currently resumable, ie. the stream we hold the state 189 * for in {@link #clientHandledStanzasCount}, {@link #serverHandledStanzasCount} and 190 * {@link #unacknowledgedStanzas}. 191 */ 192 private String smSessionId; 193 194 /** 195 * Represents the state of stream management resumption. 196 * <p> 197 * Unlike other sync points, this sync point is marked volatile because it is also read by the reader thread. 198 * </p> 199 */ 200 private volatile SyncPointState smResumedSyncPoint; 201 private Failed smResumptionFailed; 202 203 /** 204 * Represents the state of stream magement. 205 * <p> 206 * This boolean is marked volatile as it is read by various threads, including the reader thread via {@link #isSmEnabled()}. 207 * </p> 208 */ 209 private volatile boolean smEnabledSyncPoint; 210 211 /** 212 * The client's preferred maximum resumption time in seconds. 213 */ 214 private int smClientMaxResumptionTime = -1; 215 216 /** 217 * The server's preferred maximum resumption time in seconds. 218 */ 219 private int smServerMaxResumptionTime = -1; 220 221 /** 222 * Indicates whether Stream Management (XEP-198) should be used if it's supported by the server. 223 */ 224 private boolean useSm = useSmDefault; 225 private boolean useSmResumption = useSmResumptionDefault; 226 227 /** 228 * The counter that the server sends the client about it's current height. For example, if the server sends 229 * {@code <a h='42'/>}, then this will be set to 42 (while also handling the {@link #unacknowledgedStanzas} queue). 230 */ 231 private long serverHandledStanzasCount = 0; 232 233 /** 234 * The counter for stanzas handled ("received") by the client. 235 * <p> 236 * Note that we don't need to synchronize this counter. Although JLS 17.7 states that reads and writes to longs are 237 * not atomic, it guarantees that there are at most 2 separate writes, one to each 32-bit half. And since 238 * {@link SMUtils#incrementHeight(long)} masks the lower 32 bit, we only operate on one half of the long and 239 * therefore have no concurrency problem because the read/write operations on one half are guaranteed to be atomic. 240 * </p> 241 */ 242 private long clientHandledStanzasCount = 0; 243 244 private BlockingQueue<Stanza> unacknowledgedStanzas; 245 246 /** 247 * Set to true if Stream Management was at least once enabled for this connection. 248 */ 249 private boolean smWasEnabledAtLeastOnce = false; 250 251 /** 252 * This listeners are invoked for every stanza that got acknowledged. 253 * <p> 254 * We use a {@link ConcurrentLinkedQueue} here in order to allow the listeners to remove 255 * themselves after they have been invoked. 256 * </p> 257 */ 258 private final Collection<StanzaListener> stanzaAcknowledgedListeners = new ConcurrentLinkedQueue<>(); 259 260 /** 261 * These listeners are invoked for every stanza that got dropped. 262 * <p> 263 * We use a {@link ConcurrentLinkedQueue} here in order to allow the listeners to remove 264 * themselves after they have been invoked. 265 * </p> 266 */ 267 private final Collection<StanzaListener> stanzaDroppedListeners = new ConcurrentLinkedQueue<>(); 268 269 /** 270 * This listeners are invoked for a acknowledged stanza that has the given stanza ID. They will 271 * only be invoked once and automatically removed after that. 272 */ 273 private final Map<String, StanzaListener> stanzaIdAcknowledgedListeners = new ConcurrentHashMap<>(); 274 275 /** 276 * Predicates that determine if an stream management ack should be requested from the server. 277 * <p> 278 * We use a linked hash set here, so that the order how the predicates are added matches the 279 * order in which they are invoked in order to determine if an ack request should be send or not. 280 * </p> 281 */ 282 private final Set<StanzaFilter> requestAckPredicates = new LinkedHashSet<>(); 283 284 @SuppressWarnings("HidingField") 285 private final XMPPTCPConnectionConfiguration config; 286 287 /** 288 * Creates a new XMPP connection over TCP (optionally using proxies). 289 * <p> 290 * Note that XMPPTCPConnection constructors do not establish a connection to the server 291 * and you must call {@link #connect()}. 292 * </p> 293 * 294 * @param config the connection configuration. 295 */ 296 public XMPPTCPConnection(XMPPTCPConnectionConfiguration config) { 297 super(config); 298 this.config = config; 299 addConnectionListener(new ConnectionListener() { 300 @Override 301 public void connectionClosedOnError(Exception e) { 302 if (e instanceof XMPPException.StreamErrorException || e instanceof StreamManagementException) { 303 dropSmState(); 304 } 305 } 306 }); 307 308 // Re-init the reader and writer in case of SASL <success/>. This is done to reset the parser since a new stream 309 // is initiated. 310 buildNonzaCallback().listenFor(SaslNonza.Success.class, s -> resetParser()).install(); 311 } 312 313 /** 314 * Creates a new XMPP connection over TCP. 315 * <p> 316 * Note that {@code jid} must be the bare JID, e.g. "user@example.org". More fine-grained control over the 317 * connection settings is available using the {@link #XMPPTCPConnection(XMPPTCPConnectionConfiguration)} 318 * constructor. 319 * </p> 320 * 321 * @param jid the bare JID used by the client. 322 * @param password the password or authentication token. 323 * @throws XmppStringprepException if the provided string is invalid. 324 */ 325 public XMPPTCPConnection(CharSequence jid, String password) throws XmppStringprepException { 326 this(XMPPTCPConnectionConfiguration.builder().setXmppAddressAndPassword(jid, password).build()); 327 } 328 329 /** 330 * Creates a new XMPP connection over TCP. 331 * <p> 332 * This is the simplest constructor for connecting to an XMPP server. Alternatively, 333 * you can get fine-grained control over connection settings using the 334 * {@link #XMPPTCPConnection(XMPPTCPConnectionConfiguration)} constructor. 335 * </p> 336 * @param username TODO javadoc me please 337 * @param password TODO javadoc me please 338 * @param serviceName TODO javadoc me please 339 * @throws XmppStringprepException if the provided string is invalid. 340 */ 341 public XMPPTCPConnection(CharSequence username, String password, String serviceName) throws XmppStringprepException { 342 this(XMPPTCPConnectionConfiguration.builder().setUsernameAndPassword(username, password).setXmppDomain( 343 JidCreate.domainBareFrom(serviceName)).build()); 344 } 345 346 @Override 347 protected void throwNotConnectedExceptionIfAppropriate() throws NotConnectedException { 348 if (packetWriter == null) { 349 throw new NotConnectedException(); 350 } 351 packetWriter.throwNotConnectedExceptionIfDoneAndResumptionNotPossible(); 352 } 353 354 @Override 355 protected void throwAlreadyConnectedExceptionIfAppropriate() throws AlreadyConnectedException { 356 if (isConnected() && !disconnectedButResumeable) { 357 throw new AlreadyConnectedException(); 358 } 359 } 360 361 @Override 362 protected void throwAlreadyLoggedInExceptionIfAppropriate() throws AlreadyLoggedInException { 363 if (isAuthenticated() && !disconnectedButResumeable) { 364 throw new AlreadyLoggedInException(); 365 } 366 } 367 368 @Override 369 protected void afterSuccessfulLogin(final boolean resumed) throws NotConnectedException, InterruptedException { 370 // Reset the flag in case it was set 371 disconnectedButResumeable = false; 372 super.afterSuccessfulLogin(resumed); 373 } 374 375 @Override 376 protected synchronized void loginInternal(String username, String password, Resourcepart resource) throws XMPPException, 377 SmackException, IOException, InterruptedException { 378 // Authenticate using SASL 379 SSLSession sslSession = secureSocket != null ? secureSocket.getSession() : null; 380 381 streamFeaturesAfterAuthenticationReceived = false; 382 authenticate(username, password, config.getAuthzid(), sslSession); 383 384 // Wait for stream features after the authentication. 385 // TODO: The name of this synchronization point "maybeCompressFeaturesReceived" is not perfect. It should be 386 // renamed to "streamFeaturesAfterAuthenticationReceived". 387 waitForConditionOrThrowConnectionException(() -> streamFeaturesAfterAuthenticationReceived, "compress features from server"); 388 389 // If compression is enabled then request the server to use stream compression. XEP-170 390 // recommends to perform stream compression before resource binding. 391 maybeEnableCompression(); 392 393 smResumedSyncPoint = SyncPointState.initial; 394 smResumptionFailed = null; 395 if (isSmResumptionPossible()) { 396 smResumedSyncPoint = SyncPointState.request_sent; 397 sendNonza(new Resume(clientHandledStanzasCount, smSessionId)); 398 waitForConditionOrConnectionException(() -> smResumedSyncPoint == SyncPointState.successful || smResumptionFailed != null, "resume previous stream"); 399 if (smResumedSyncPoint == SyncPointState.successful) { 400 // We successfully resumed the stream, be done here 401 afterSuccessfulLogin(true); 402 return; 403 } 404 // SM resumption failed, what Smack does here is to report success of 405 // lastFeaturesReceived in case of sm resumption was answered with 'failed' so that 406 // normal resource binding can be tried. 407 assert smResumptionFailed != null; 408 LOGGER.fine("Stream resumption failed, continuing with normal stream establishment process: " + smResumptionFailed); 409 } 410 411 // We either failed to resume a previous stream management (SM) session, or we did not even try. In any case, 412 // mark SM as not enabled. Most importantly, we do this prior calling bindResourceAndEstablishSession(), as the 413 // bind IQ may trigger a SM ack request, which would be invalid in the pre resource bound state. 414 smEnabledSyncPoint = false; 415 416 List<Stanza> previouslyUnackedStanzas = new LinkedList<Stanza>(); 417 if (unacknowledgedStanzas != null) { 418 // There was a previous connection with SM enabled but that was either not resumable or 419 // failed to resume. Make sure that we (re-)send the unacknowledged stanzas. 420 unacknowledgedStanzas.drainTo(previouslyUnackedStanzas); 421 // Reset unacknowledged stanzas to 'null' to signal that we never send 'enable' in this 422 // XMPP session (There maybe was an enabled in a previous XMPP session of this 423 // connection instance though). This is used in writePackets to decide if stanzas should 424 // be added to the unacknowledged stanzas queue, because they have to be added right 425 // after the 'enable' stream element has been sent. 426 dropSmState(); 427 } 428 429 // Now bind the resource. It is important to do this *after* we dropped an eventually 430 // existing Stream Management state. As otherwise <bind/> and <session/> may end up in 431 // unacknowledgedStanzas and become duplicated on reconnect. See SMACK-706. 432 bindResourceAndEstablishSession(resource); 433 434 if (isSmAvailable() && useSm) { 435 // Remove what is maybe left from previously stream managed sessions 436 serverHandledStanzasCount = 0; 437 sendNonza(new Enable(useSmResumption, smClientMaxResumptionTime)); 438 // XEP-198 3. Enabling Stream Management. If the server response to 'Enable' is 'Failed' 439 // then this is a non recoverable error and we therefore throw an exception. 440 waitForConditionOrThrowConnectionException(() -> smEnabledSyncPoint, "enabling stream mangement"); 441 synchronized (requestAckPredicates) { 442 if (requestAckPredicates.isEmpty()) { 443 // Assure that we have at lest one predicate set up that so that we request acks 444 // for the server and eventually flush some stanzas from the unacknowledged 445 // stanza queue 446 requestAckPredicates.add(Predicate.forMessagesOrAfter5Stanzas()); 447 } 448 } 449 } 450 // Inform client about failed resumption if possible, resend stanzas otherwise 451 // Process the stanzas synchronously so a client can re-queue them for transmission 452 // before it is informed about connection success 453 if (!stanzaDroppedListeners.isEmpty()) { 454 for (Stanza stanza : previouslyUnackedStanzas) { 455 for (StanzaListener listener : stanzaDroppedListeners) { 456 try { 457 listener.processStanza(stanza); 458 } 459 catch (InterruptedException | NotConnectedException | NotLoggedInException e) { 460 LOGGER.log(Level.FINER, "StanzaDroppedListener received exception", e); 461 } 462 } 463 } 464 } else { 465 for (Stanza stanza : previouslyUnackedStanzas) { 466 sendStanzaInternal(stanza); 467 } 468 } 469 470 afterSuccessfulLogin(false); 471 } 472 473 @Override 474 public boolean isSecureConnection() { 475 return secureSocket != null; 476 } 477 478 /** 479 * Shuts the current connection down. After this method returns, the connection must be ready 480 * for re-use by connect. 481 */ 482 @Override 483 protected void shutdown() { 484 if (isSmEnabled()) { 485 try { 486 // Try to send a last SM Acknowledgement. Most servers won't find this information helpful, as the SM 487 // state is dropped after a clean disconnect anyways. OTOH it doesn't hurt much either. 488 sendSmAcknowledgementInternal(); 489 } catch (InterruptedException | NotConnectedException e) { 490 LOGGER.log(Level.FINE, "Can not send final SM ack as connection is not connected", e); 491 } 492 } 493 shutdown(false); 494 } 495 496 @Override 497 public synchronized void instantShutdown() { 498 shutdown(true); 499 } 500 501 private void shutdown(boolean instant) { 502 // The writer thread may already been finished at this point, for example when the connection is in the 503 // disconnected-but-resumable state. There is no need to wait for the closing stream tag from the server in this 504 // case. 505 if (!packetWriter.done()) { 506 // First shutdown the writer, this will result in a closing stream element getting send to 507 // the server 508 LOGGER.finer(packetWriter.threadName + " shutdown()"); 509 packetWriter.shutdown(instant); 510 LOGGER.finer(packetWriter.threadName + " shutdown() returned"); 511 512 if (!instant) { 513 waitForClosingStreamTagFromServer(); 514 } 515 } 516 517 LOGGER.finer(packetReader.threadName + " shutdown()"); 518 packetReader.shutdown(); 519 LOGGER.finer(packetReader.threadName + " shutdown() returned"); 520 521 CloseableUtil.maybeClose(socket, LOGGER); 522 523 setWasAuthenticated(); 524 525 try { 526 boolean readerAndWriterThreadsTermianted = waitFor(() -> !packetWriter.running && !packetReader.running); 527 if (!readerAndWriterThreadsTermianted) { 528 LOGGER.severe("Reader and/or writer threads did not terminate timely. Writer running: " 529 + packetWriter.running + ", Reader running: " + packetReader.running); 530 } else { 531 LOGGER.fine("Reader and writer threads terminated"); 532 } 533 } catch (InterruptedException e) { 534 LOGGER.log(Level.FINE, "Interrupted while waiting for reader and writer threads to terminate", e); 535 } 536 537 if (disconnectedButResumeable) { 538 return; 539 } 540 541 // If we are able to resume the stream, then don't set 542 // connected/authenticated/usingTLS to false since we like to behave like we are still 543 // connected (e.g. sendStanza should not throw a NotConnectedException). 544 if (instant) { 545 disconnectedButResumeable = isSmResumptionPossible(); 546 if (!disconnectedButResumeable) { 547 // Reset the stream management session id to null, since the stream is no longer resumable. Note that we 548 // keep the unacknowledgedStanzas queue, because we want to resend them when we are reconnected. 549 smSessionId = null; 550 } 551 } else { 552 disconnectedButResumeable = false; 553 554 // Drop the stream management state if this is not an instant shutdown. We send 555 // a </stream> close tag and now the stream management state is no longer valid. 556 // This also prevents that we will potentially (re-)send any unavailable presence we 557 // may have send, because it got put into the unacknowledged queue and was not acknowledged before the 558 // connection terminated. 559 dropSmState(); 560 // Note that we deliberately do not reset authenticatedConnectionInitiallyEstablishedTimestamp here, so that the 561 // information is available in the connectionClosedOnError() listeners. 562 } 563 authenticated = false; 564 connected = false; 565 secureSocket = null; 566 reader = null; 567 writer = null; 568 569 initState(); 570 } 571 572 @Override 573 public void sendNonza(Nonza element) throws NotConnectedException, InterruptedException { 574 packetWriter.sendStreamElement(element); 575 } 576 577 @Override 578 protected void sendStanzaInternal(Stanza packet) throws NotConnectedException, InterruptedException { 579 packetWriter.sendStreamElement(packet); 580 if (isSmEnabled()) { 581 for (StanzaFilter requestAckPredicate : requestAckPredicates) { 582 if (requestAckPredicate.accept(packet)) { 583 requestSmAcknowledgementInternal(); 584 break; 585 } 586 } 587 } 588 } 589 590 private void connectUsingConfiguration() throws ConnectionException, IOException, InterruptedException { 591 RemoteXmppTcpConnectionEndpoints.Result<Rfc6120TcpRemoteConnectionEndpoint> result = RemoteXmppTcpConnectionEndpoints.lookup(config); 592 593 List<RemoteConnectionException<Rfc6120TcpRemoteConnectionEndpoint>> connectionExceptions = new ArrayList<>(); 594 595 SocketFactory socketFactory = config.getSocketFactory(); 596 ProxyInfo proxyInfo = config.getProxyInfo(); 597 int timeout = config.getConnectTimeout(); 598 if (socketFactory == null) { 599 socketFactory = SocketFactory.getDefault(); 600 } 601 for (Rfc6120TcpRemoteConnectionEndpoint endpoint : result.discoveredRemoteConnectionEndpoints) { 602 Iterator<? extends InetAddress> inetAddresses; 603 String host = endpoint.getHost().toString(); 604 UInt16 portUint16 = endpoint.getPort(); 605 int port = portUint16.intValue(); 606 if (proxyInfo == null) { 607 inetAddresses = endpoint.getInetAddresses().iterator(); 608 assert inetAddresses.hasNext(); 609 610 innerloop: while (inetAddresses.hasNext()) { 611 // Create a *new* Socket before every connection attempt, i.e. connect() call, since Sockets are not 612 // re-usable after a failed connection attempt. See also SMACK-724. 613 SmackFuture.SocketFuture socketFuture = new SmackFuture.SocketFuture(socketFactory); 614 615 final InetAddress inetAddress = inetAddresses.next(); 616 final InetSocketAddress inetSocketAddress = new InetSocketAddress(inetAddress, port); 617 LOGGER.finer("Trying to establish TCP connection to " + inetSocketAddress); 618 socketFuture.connectAsync(inetSocketAddress, timeout); 619 620 try { 621 socket = socketFuture.getOrThrow(); 622 } catch (IOException e) { 623 RemoteConnectionException<Rfc6120TcpRemoteConnectionEndpoint> rce = new RemoteConnectionException<>( 624 endpoint, inetAddress, e); 625 connectionExceptions.add(rce); 626 if (inetAddresses.hasNext()) { 627 continue innerloop; 628 } else { 629 break innerloop; 630 } 631 } 632 LOGGER.finer("Established TCP connection to " + inetSocketAddress); 633 // We found a host to connect to, return here 634 this.host = host; 635 this.port = portUint16; 636 return; 637 } 638 } else { 639 // TODO: Move this into the inner-loop above. There appears no reason why we should not try a proxy 640 // connection to every inet address of each connection endpoint. 641 socket = socketFactory.createSocket(); 642 StringUtils.requireNotNullNorEmpty(host, "Host of endpoint " + endpoint + " must not be null when using a Proxy"); 643 final String hostAndPort = host + " at port " + port; 644 LOGGER.finer("Trying to establish TCP connection via Proxy to " + hostAndPort); 645 try { 646 proxyInfo.getProxySocketConnection().connect(socket, host, port, timeout); 647 } catch (IOException e) { 648 CloseableUtil.maybeClose(socket, LOGGER); 649 RemoteConnectionException<Rfc6120TcpRemoteConnectionEndpoint> rce = new RemoteConnectionException<>(endpoint, null, e); 650 connectionExceptions.add(rce); 651 continue; 652 } 653 LOGGER.finer("Established TCP connection to " + hostAndPort); 654 // We found a host to connect to, return here 655 this.host = host; 656 this.port = portUint16; 657 return; 658 } 659 } 660 661 // There are no more host addresses to try 662 // throw an exception and report all tried 663 // HostAddresses in the exception 664 throw EndpointConnectionException.from(result.lookupFailures, connectionExceptions); 665 } 666 667 /** 668 * Initializes the connection by creating a stanza reader and writer and opening a 669 * XMPP stream to the server. 670 * 671 * @throws IOException if an I/O error occurred. 672 * @throws InterruptedException if the calling thread was interrupted. 673 */ 674 private void initConnection() throws IOException, InterruptedException { 675 compressionHandler = null; 676 677 // Set the reader and writer instance variables 678 initReaderAndWriter(); 679 680 // Start the writer thread. This will open an XMPP stream to the server 681 packetWriter.init(); 682 // Start the reader thread. The startup() method will block until we 683 // get an opening stream packet back from server 684 packetReader.init(); 685 } 686 687 private void initReaderAndWriter() throws IOException { 688 InputStream is = socket.getInputStream(); 689 OutputStream os = socket.getOutputStream(); 690 if (compressionHandler != null) { 691 is = compressionHandler.getInputStream(is); 692 os = compressionHandler.getOutputStream(os); 693 } 694 // OutputStreamWriter is already buffered, no need to wrap it into a BufferedWriter 695 writer = new OutputStreamWriter(os, "UTF-8"); 696 reader = new BufferedReader(new InputStreamReader(is, "UTF-8")); 697 698 // If debugging is enabled, we open a window and write out all network traffic. 699 initDebugger(); 700 } 701 702 /** 703 * The server has indicated that TLS negotiation can start. We now need to secure the 704 * existing plain connection and perform a handshake. This method won't return until the 705 * connection has finished the handshake or an error occurred while securing the connection. 706 * @throws IOException if an I/O error occurred. 707 * @throws SecurityNotPossibleException if TLS is not possible. 708 * @throws CertificateException if there is an issue with the certificate. 709 */ 710 @SuppressWarnings("LiteralClassName") 711 private void proceedTLSReceived() throws IOException, SecurityNotPossibleException, CertificateException { 712 SmackTlsContext smackTlsContext = getSmackTlsContext(); 713 714 Socket plain = socket; 715 // Secure the plain connection 716 socket = smackTlsContext.sslContext.getSocketFactory().createSocket(plain, 717 config.getXMPPServiceDomain().toString(), plain.getPort(), true); 718 719 final SSLSocket sslSocket = (SSLSocket) socket; 720 // Immediately set the enabled SSL protocols and ciphers. See SMACK-712 why this is 721 // important (at least on certain platforms) and it seems to be a good idea anyways to 722 // prevent an accidental implicit handshake. 723 TLSUtils.setEnabledProtocolsAndCiphers(sslSocket, config.getEnabledSSLProtocols(), config.getEnabledSSLCiphers()); 724 725 // Initialize the reader and writer with the new secured version 726 initReaderAndWriter(); 727 728 // Proceed to do the handshake 729 sslSocket.startHandshake(); 730 731 if (smackTlsContext.daneVerifier != null) { 732 smackTlsContext.daneVerifier.finish(sslSocket.getSession()); 733 } 734 735 final HostnameVerifier verifier = getConfiguration().getHostnameVerifier(); 736 if (verifier == null) { 737 throw new IllegalStateException("No HostnameVerifier set. Use connectionConfiguration.setHostnameVerifier() to configure."); 738 } 739 740 final String verifierHostname; 741 { 742 DnsName xmppServiceDomainDnsName = getConfiguration().getXmppServiceDomainAsDnsNameIfPossible(); 743 // Try to convert the XMPP service domain, which potentially includes Unicode characters, into ASCII 744 // Compatible Encoding (ACE) to match RFC3280 dNSname IA5String constraint. 745 // See also: https://bugzilla.mozilla.org/show_bug.cgi?id=280839#c1 746 if (xmppServiceDomainDnsName != null) { 747 verifierHostname = xmppServiceDomainDnsName.ace; 748 } 749 else { 750 LOGGER.log(Level.WARNING, "XMPP service domain name '" + getXMPPServiceDomain() 751 + "' can not be represented as DNS name. TLS X.509 certificate validiation may fail."); 752 verifierHostname = getXMPPServiceDomain().toString(); 753 } 754 } 755 756 final boolean verificationSuccessful; 757 // Verify the TLS session. 758 verificationSuccessful = verifier.verify(verifierHostname, sslSocket.getSession()); 759 if (!verificationSuccessful) { 760 throw new CertificateException( 761 "Hostname verification of certificate failed. Certificate does not authenticate " 762 + getXMPPServiceDomain()); 763 } 764 765 // Set that TLS was successful 766 secureSocket = sslSocket; 767 } 768 769 /** 770 * Returns the compression handler that can be used for one compression methods offered by the server. 771 * 772 * @return a instance of XMPPInputOutputStream or null if no suitable instance was found 773 * 774 */ 775 private static XMPPInputOutputStream maybeGetCompressionHandler(Compress.Feature compression) { 776 for (XMPPInputOutputStream handler : SmackConfiguration.getCompressionHandlers()) { 777 String method = handler.getCompressionMethod(); 778 if (compression.getMethods().contains(method)) 779 return handler; 780 } 781 return null; 782 } 783 784 @Override 785 public boolean isUsingCompression() { 786 return compressionHandler != null && compressSyncPoint; 787 } 788 789 /** 790 * <p> 791 * Starts using stream compression that will compress network traffic. Traffic can be 792 * reduced up to 90%. Therefore, stream compression is ideal when using a slow speed network 793 * connection. However, the server and the client will need to use more CPU time in order to 794 * un/compress network data so under high load the server performance might be affected. 795 * </p> 796 * <p> 797 * Stream compression has to have been previously offered by the server. Currently only the 798 * zlib method is supported by the client. Stream compression negotiation has to be done 799 * before authentication took place. 800 * </p> 801 * 802 * @throws NotConnectedException if the XMPP connection is not connected. 803 * @throws SmackException if Smack detected an exceptional situation. 804 * @throws InterruptedException if the calling thread was interrupted. 805 * @throws XMPPException if an XMPP protocol error was received. 806 */ 807 private void maybeEnableCompression() throws SmackException, InterruptedException, XMPPException { 808 if (!config.isCompressionEnabled()) { 809 return; 810 } 811 812 Compress.Feature compression = getFeature(Compress.Feature.class); 813 if (compression == null) { 814 // Server does not support compression 815 return; 816 } 817 // If stream compression was offered by the server and we want to use 818 // compression then send compression request to the server 819 if ((compressionHandler = maybeGetCompressionHandler(compression)) != null) { 820 compressSyncPoint = false; 821 sendNonza(new Compress(compressionHandler.getCompressionMethod())); 822 waitForConditionOrThrowConnectionException(() -> compressSyncPoint, "establishing stream compression"); 823 } else { 824 LOGGER.warning("Could not enable compression because no matching handler/method pair was found"); 825 } 826 } 827 828 /** 829 * Establishes a connection to the XMPP server. It basically 830 * creates and maintains a socket connection to the server. 831 * <p> 832 * Listeners will be preserved from a previous connection if the reconnection 833 * occurs after an abrupt termination. 834 * </p> 835 * 836 * @throws XMPPException if an error occurs while trying to establish the connection. 837 * @throws SmackException if Smack detected an exceptional situation. 838 * @throws IOException if an I/O error occurred. 839 * @throws InterruptedException if the calling thread was interrupted. 840 */ 841 @Override 842 protected void connectInternal() throws SmackException, IOException, XMPPException, InterruptedException { 843 // Establishes the TCP connection to the server and does setup the reader and writer. Throws an exception if 844 // there is an error establishing the connection 845 connectUsingConfiguration(); 846 847 connected = true; 848 849 // We connected successfully to the servers TCP port 850 initConnection(); 851 852 // TLS handled will be true either if TLS was established, or if it was not mandatory. 853 waitForConditionOrThrowConnectionException(() -> tlsHandled, "establishing TLS"); 854 855 // Wait with SASL auth until the SASL mechanisms have been received 856 waitForConditionOrThrowConnectionException(() -> saslFeatureReceived, "SASL mechanisms stream feature from server"); 857 } 858 859 /** 860 * For unit testing purposes 861 * 862 * @param writer TODO javadoc me please 863 */ 864 protected void setWriter(Writer writer) { 865 this.writer = writer; 866 } 867 868 @Override 869 protected void afterFeaturesReceived() throws NotConnectedException, InterruptedException, SecurityRequiredByServerException { 870 StartTls startTlsFeature = getFeature(StartTls.class); 871 if (startTlsFeature != null) { 872 if (startTlsFeature.required() && config.getSecurityMode() == SecurityMode.disabled) { 873 SecurityRequiredByServerException smackException = new SecurityRequiredByServerException(); 874 currentSmackException = smackException; 875 notifyWaitingThreads(); 876 throw smackException; 877 } 878 879 if (config.getSecurityMode() != ConnectionConfiguration.SecurityMode.disabled) { 880 sendNonza(new StartTls()); 881 } else { 882 tlsHandled = true; 883 notifyWaitingThreads(); 884 } 885 } else { 886 tlsHandled = true; 887 notifyWaitingThreads(); 888 } 889 890 if (isSaslAuthenticated()) { 891 // If we have received features after the SASL has been successfully completed, then we 892 // have also *maybe* received, as it is an optional feature, the compression feature 893 // from the server. 894 streamFeaturesAfterAuthenticationReceived = true; 895 notifyWaitingThreads(); 896 } 897 } 898 899 private void resetParser() throws IOException { 900 try { 901 packetReader.parser = SmackXmlParser.newXmlParser(reader); 902 } catch (XmlPullParserException e) { 903 throw new IOException(e); 904 } 905 } 906 907 private void openStreamAndResetParser() throws IOException, NotConnectedException, InterruptedException { 908 sendStreamOpen(); 909 resetParser(); 910 } 911 912 protected class PacketReader { 913 914 private final String threadName = "Smack Reader (" + getConnectionCounter() + ')'; 915 916 XmlPullParser parser; 917 918 private volatile boolean done; 919 920 private boolean running; 921 922 /** 923 * Initializes the reader in order to be used. The reader is initialized during the 924 * first connection and when reconnecting due to an abruptly disconnection. 925 */ 926 void init() { 927 done = false; 928 929 running = true; 930 Async.go(new Runnable() { 931 @Override 932 public void run() { 933 LOGGER.finer(threadName + " start"); 934 try { 935 parsePackets(); 936 } finally { 937 LOGGER.finer(threadName + " exit"); 938 running = false; 939 notifyWaitingThreads(); 940 } 941 } 942 }, threadName); 943 } 944 945 /** 946 * Shuts the stanza reader down. This method simply sets the 'done' flag to true. 947 */ 948 void shutdown() { 949 done = true; 950 } 951 952 /** 953 * Parse top-level packets in order to process them further. 954 */ 955 private void parsePackets() { 956 try { 957 openStreamAndResetParser(); 958 XmlPullParser.Event eventType = parser.getEventType(); 959 while (!done) { 960 switch (eventType) { 961 case START_ELEMENT: 962 final String name = parser.getName(); 963 final String namespace = parser.getNamespace(); 964 965 switch (name) { 966 case Message.ELEMENT: 967 case IQ.IQ_ELEMENT: 968 case Presence.ELEMENT: 969 try { 970 parseAndProcessStanza(parser); 971 } finally { 972 clientHandledStanzasCount = SMUtils.incrementHeight(clientHandledStanzasCount); 973 } 974 break; 975 case "stream": 976 if (StreamOpen.ETHERX_JABBER_STREAMS_NAMESPACE.equals(namespace)) { 977 onStreamOpen(parser); 978 } 979 break; 980 case "error": 981 StreamError streamError = PacketParserUtils.parseStreamError(parser); 982 // Stream errors are non recoverable, throw this exceptions. Also note that this will set 983 // this exception as current connection exceptions and notify any waiting threads. 984 throw new StreamErrorException(streamError); 985 case "features": 986 parseFeaturesAndNotify(parser); 987 break; 988 case "proceed": 989 // Secure the connection by negotiating TLS 990 proceedTLSReceived(); 991 // Send a new opening stream to the server 992 openStreamAndResetParser(); 993 break; 994 case "failure": 995 switch (namespace) { 996 case "urn:ietf:params:xml:ns:xmpp-tls": 997 // TLS negotiation has failed. The server will close the connection 998 // TODO Parse failure stanza 999 throw new SmackException.SmackMessageException("TLS negotiation has failed"); 1000 case "http://jabber.org/protocol/compress": 1001 // Stream compression has been denied. This is a recoverable 1002 // situation. It is still possible to authenticate and 1003 // use the connection but using an uncompressed connection 1004 // TODO Parse failure stanza 1005 currentSmackException = new SmackException.SmackMessageException("Could not establish compression"); 1006 notifyWaitingThreads(); 1007 break; 1008 default: 1009 parseAndProcessNonza(parser); 1010 } 1011 break; 1012 case Compressed.ELEMENT: 1013 // Server confirmed that it's possible to use stream compression. Start 1014 // stream compression 1015 // Initialize the reader and writer with the new compressed version 1016 initReaderAndWriter(); 1017 // Send a new opening stream to the server 1018 openStreamAndResetParser(); 1019 // Notify that compression is being used 1020 compressSyncPoint = true; 1021 notifyWaitingThreads(); 1022 break; 1023 case Enabled.ELEMENT: 1024 Enabled enabled = ParseStreamManagement.enabled(parser); 1025 if (enabled.isResumeSet()) { 1026 smSessionId = enabled.getId(); 1027 if (StringUtils.isNullOrEmpty(smSessionId)) { 1028 SmackException xmppException = new SmackException.SmackMessageException("Stream Management 'enabled' element with resume attribute but without session id received"); 1029 setCurrentConnectionExceptionAndNotify(xmppException); 1030 throw xmppException; 1031 } 1032 smServerMaxResumptionTime = enabled.getMaxResumptionTime(); 1033 } else { 1034 // Mark this a non-resumable stream by setting smSessionId to null 1035 smSessionId = null; 1036 } 1037 clientHandledStanzasCount = 0; 1038 smWasEnabledAtLeastOnce = true; 1039 smEnabledSyncPoint = true; 1040 notifyWaitingThreads(); 1041 break; 1042 case Failed.ELEMENT: 1043 Failed failed = ParseStreamManagement.failed(parser); 1044 if (smResumedSyncPoint == SyncPointState.request_sent) { 1045 // This is a <failed/> nonza in a response to resuming a previous stream, failure to do 1046 // so is non-fatal as we can simply continue with resource binding in this case. 1047 smResumptionFailed = failed; 1048 notifyWaitingThreads(); 1049 } else { 1050 FailedNonzaException xmppException = new FailedNonzaException(failed, failed.getStanzaErrorCondition()); 1051 setCurrentConnectionExceptionAndNotify(xmppException); 1052 } 1053 break; 1054 case Resumed.ELEMENT: 1055 Resumed resumed = ParseStreamManagement.resumed(parser); 1056 if (!smSessionId.equals(resumed.getPrevId())) { 1057 throw new StreamIdDoesNotMatchException(smSessionId, resumed.getPrevId()); 1058 } 1059 // Mark SM as enabled 1060 smEnabledSyncPoint = true; 1061 // First, drop the stanzas already handled by the server 1062 processHandledCount(resumed.getHandledCount()); 1063 // Then re-send what is left in the unacknowledged queue 1064 List<Stanza> stanzasToResend = new ArrayList<>(unacknowledgedStanzas.size()); 1065 unacknowledgedStanzas.drainTo(stanzasToResend); 1066 for (Stanza stanza : stanzasToResend) { 1067 sendStanzaInternal(stanza); 1068 } 1069 // If there where stanzas resent, then request a SM ack for them. 1070 // Writer's sendStreamElement() won't do it automatically based on 1071 // predicates. 1072 if (!stanzasToResend.isEmpty()) { 1073 requestSmAcknowledgementInternal(); 1074 } 1075 // Mark SM resumption as successful 1076 smResumedSyncPoint = SyncPointState.successful; 1077 notifyWaitingThreads(); 1078 break; 1079 case AckAnswer.ELEMENT: 1080 AckAnswer ackAnswer = ParseStreamManagement.ackAnswer(parser); 1081 processHandledCount(ackAnswer.getHandledCount()); 1082 break; 1083 case AckRequest.ELEMENT: 1084 ParseStreamManagement.ackRequest(parser); 1085 if (smEnabledSyncPoint) { 1086 sendSmAcknowledgementInternal(); 1087 } else { 1088 LOGGER.warning("SM Ack Request received while SM is not enabled"); 1089 } 1090 break; 1091 default: 1092 parseAndProcessNonza(parser); 1093 break; 1094 } 1095 break; 1096 case END_ELEMENT: 1097 final String endTagName = parser.getName(); 1098 if ("stream".equals(endTagName)) { 1099 if (!parser.getNamespace().equals("http://etherx.jabber.org/streams")) { 1100 LOGGER.warning(XMPPTCPConnection.this + " </stream> but different namespace " + parser.getNamespace()); 1101 break; 1102 } 1103 1104 // Check if the queue was already shut down before reporting success on closing stream tag 1105 // received. This avoids a race if there is a disconnect(), followed by a connect(), which 1106 // did re-start the queue again, causing this writer to assume that the queue is not 1107 // shutdown, which results in a call to disconnect(). 1108 final boolean queueWasShutdown = packetWriter.queue.isShutdown(); 1109 closingStreamReceived = true; 1110 notifyWaitingThreads(); 1111 1112 if (queueWasShutdown) { 1113 // We received a closing stream element *after* we initiated the 1114 // termination of the session by sending a closing stream element to 1115 // the server first 1116 return; 1117 } else { 1118 // We received a closing stream element from the server without us 1119 // sending a closing stream element first. This means that the 1120 // server wants to terminate the session, therefore disconnect 1121 // the connection 1122 LOGGER.info(XMPPTCPConnection.this 1123 + " received closing </stream> element." 1124 + " Server wants to terminate the connection, calling disconnect()"); 1125 ASYNC_BUT_ORDERED.performAsyncButOrdered(XMPPTCPConnection.this, new Runnable() { 1126 @Override 1127 public void run() { 1128 disconnect(); 1129 }}); 1130 } 1131 } 1132 break; 1133 case END_DOCUMENT: 1134 // END_DOCUMENT only happens in an error case, as otherwise we would see a 1135 // closing stream element before. 1136 throw new SmackException.SmackMessageException( 1137 "Parser got END_DOCUMENT event. This could happen e.g. if the server closed the connection without sending a closing stream element"); 1138 default: 1139 // Catch all for incomplete switch (MissingCasesInEnumSwitch) statement. 1140 break; 1141 } 1142 eventType = parser.next(); 1143 } 1144 } 1145 catch (Exception e) { 1146 // Set running to false since this thread will exit here and notifyConnectionError() will wait until 1147 // the reader and writer thread's 'running' value is false. Hence we need to set it to false before calling 1148 // notifyConnetctionError() below, even though run() also sets it to false. Therefore, do not remove this. 1149 running = false; 1150 1151 String ignoreReasonThread = null; 1152 1153 boolean writerThreadWasShutDown = packetWriter.queue.isShutdown(); 1154 if (writerThreadWasShutDown) { 1155 ignoreReasonThread = "writer"; 1156 } else if (done) { 1157 ignoreReasonThread = "reader"; 1158 } 1159 1160 if (ignoreReasonThread != null) { 1161 LOGGER.log(Level.FINER, "Ignoring " + e + " as " + ignoreReasonThread + " was already shut down"); 1162 return; 1163 } 1164 1165 // Close the connection and notify connection listeners of the error. 1166 notifyConnectionError(e); 1167 } 1168 } 1169 } 1170 1171 protected class PacketWriter { 1172 public static final int QUEUE_SIZE = XMPPTCPConnection.QUEUE_SIZE; 1173 public static final int UNACKKNOWLEDGED_STANZAS_QUEUE_SIZE = 1024; 1174 public static final int UNACKKNOWLEDGED_STANZAS_QUEUE_SIZE_HIGH_WATER_MARK = (int) (0.3 * UNACKKNOWLEDGED_STANZAS_QUEUE_SIZE); 1175 1176 private final String threadName = "Smack Writer (" + getConnectionCounter() + ')'; 1177 1178 private final ArrayBlockingQueueWithShutdown<Element> queue = new ArrayBlockingQueueWithShutdown<>( 1179 QUEUE_SIZE, true); 1180 1181 /** 1182 * If set, the stanza writer is shut down 1183 */ 1184 protected volatile Long shutdownTimestamp = null; 1185 1186 private volatile boolean instantShutdown; 1187 1188 /** 1189 * True if some preconditions are given to start the bundle and defer mechanism. 1190 * <p> 1191 * This will likely get set to true right after the start of the writer thread, because 1192 * {@link #nextStreamElement()} will check if {@link queue} is empty, which is probably the case, and then set 1193 * this field to true. 1194 * </p> 1195 */ 1196 private boolean shouldBundleAndDefer; 1197 1198 private boolean running; 1199 1200 /** 1201 * Initializes the writer in order to be used. It is called at the first connection and also 1202 * is invoked if the connection is disconnected by an error. 1203 */ 1204 void init() { 1205 shutdownTimestamp = null; 1206 1207 if (unacknowledgedStanzas != null) { 1208 // It's possible that there are new stanzas in the writer queue that 1209 // came in while we were disconnected but resumable, drain those into 1210 // the unacknowledged queue so that they get resent now 1211 drainWriterQueueToUnacknowledgedStanzas(); 1212 } 1213 1214 queue.start(); 1215 running = true; 1216 Async.go(new Runnable() { 1217 @Override 1218 public void run() { 1219 LOGGER.finer(threadName + " start"); 1220 try { 1221 writePackets(); 1222 } finally { 1223 LOGGER.finer(threadName + " exit"); 1224 running = false; 1225 notifyWaitingThreads(); 1226 } 1227 } 1228 }, threadName); 1229 } 1230 1231 private boolean done() { 1232 return shutdownTimestamp != null; 1233 } 1234 1235 protected void throwNotConnectedExceptionIfDoneAndResumptionNotPossible() throws NotConnectedException { 1236 final boolean done = done(); 1237 if (done) { 1238 final boolean smResumptionPossible = isSmResumptionPossible(); 1239 // Don't throw a NotConnectedException is there is an resumable stream available 1240 if (!smResumptionPossible) { 1241 throw new NotConnectedException(XMPPTCPConnection.this, "done=" + done 1242 + " smResumptionPossible=" + smResumptionPossible); 1243 } 1244 } 1245 } 1246 1247 /** 1248 * Sends the specified element to the server. 1249 * 1250 * @param element the element to send. 1251 * @throws NotConnectedException if the XMPP connection is not connected. 1252 * @throws InterruptedException if the calling thread was interrupted. 1253 */ 1254 protected void sendStreamElement(Element element) throws NotConnectedException, InterruptedException { 1255 throwNotConnectedExceptionIfDoneAndResumptionNotPossible(); 1256 try { 1257 queue.put(element); 1258 } 1259 catch (InterruptedException e) { 1260 // put() may throw an InterruptedException for two reasons: 1261 // 1. If the queue was shut down 1262 // 2. If the thread was interrupted 1263 // so we have to check which is the case 1264 throwNotConnectedExceptionIfDoneAndResumptionNotPossible(); 1265 // If the method above did not throw, then the sending thread was interrupted 1266 throw e; 1267 } 1268 } 1269 1270 /** 1271 * Shuts down the stanza writer. Once this method has been called, no further 1272 * packets will be written to the server. 1273 */ 1274 void shutdown(boolean instant) { 1275 instantShutdown = instant; 1276 queue.shutdown(); 1277 shutdownTimestamp = System.currentTimeMillis(); 1278 } 1279 1280 /** 1281 * Maybe return the next available element from the queue for writing. If the queue is shut down <b>or</b> a 1282 * spurious interrupt occurs, <code>null</code> is returned. So it is important to check the 'done' condition in 1283 * that case. 1284 * 1285 * @return the next element for writing or null. 1286 */ 1287 private Element nextStreamElement() { 1288 // It is important the we check if the queue is empty before removing an element from it 1289 if (queue.isEmpty()) { 1290 shouldBundleAndDefer = true; 1291 } 1292 Element packet = null; 1293 try { 1294 packet = queue.take(); 1295 } 1296 catch (InterruptedException e) { 1297 if (!queue.isShutdown()) { 1298 // Users shouldn't try to interrupt the packet writer thread 1299 LOGGER.log(Level.WARNING, "Writer thread was interrupted. Don't do that. Use disconnect() instead.", e); 1300 } 1301 } 1302 return packet; 1303 } 1304 1305 private void writePackets() { 1306 try { 1307 // Write out packets from the queue. 1308 while (!done()) { 1309 Element element = nextStreamElement(); 1310 if (element == null) { 1311 continue; 1312 } 1313 1314 // Get a local version of the bundle and defer callback, in case it's unset 1315 // between the null check and the method invocation 1316 final BundleAndDeferCallback localBundleAndDeferCallback = bundleAndDeferCallback; 1317 // If the preconditions are given (e.g. bundleAndDefer callback is set, queue is 1318 // empty), then we could wait a bit for further stanzas attempting to decrease 1319 // our energy consumption 1320 if (localBundleAndDeferCallback != null && isAuthenticated() && shouldBundleAndDefer) { 1321 // Reset shouldBundleAndDefer to false, nextStreamElement() will set it to true once the 1322 // queue is empty again. 1323 shouldBundleAndDefer = false; 1324 final AtomicBoolean bundlingAndDeferringStopped = new AtomicBoolean(); 1325 final int bundleAndDeferMillis = localBundleAndDeferCallback.getBundleAndDeferMillis(new BundleAndDefer( 1326 bundlingAndDeferringStopped)); 1327 if (bundleAndDeferMillis > 0) { 1328 long remainingWait = bundleAndDeferMillis; 1329 final long waitStart = System.currentTimeMillis(); 1330 synchronized (bundlingAndDeferringStopped) { 1331 while (!bundlingAndDeferringStopped.get() && remainingWait > 0) { 1332 bundlingAndDeferringStopped.wait(remainingWait); 1333 remainingWait = bundleAndDeferMillis 1334 - (System.currentTimeMillis() - waitStart); 1335 } 1336 } 1337 } 1338 } 1339 1340 Stanza packet = null; 1341 if (element instanceof Stanza) { 1342 packet = (Stanza) element; 1343 } 1344 else if (element instanceof Enable) { 1345 // The client needs to add messages to the unacknowledged stanzas queue 1346 // right after it sent 'enabled'. Stanza will be added once 1347 // unacknowledgedStanzas is not null. 1348 unacknowledgedStanzas = new ArrayBlockingQueue<>(UNACKKNOWLEDGED_STANZAS_QUEUE_SIZE); 1349 } 1350 maybeAddToUnacknowledgedStanzas(packet); 1351 1352 CharSequence elementXml = element.toXML(outgoingStreamXmlEnvironment); 1353 if (elementXml instanceof XmlStringBuilder) { 1354 try { 1355 ((XmlStringBuilder) elementXml).write(writer, outgoingStreamXmlEnvironment); 1356 } catch (NullPointerException npe) { 1357 LOGGER.log(Level.FINE, "NPE in XmlStringBuilder of " + element.getClass() + ": " + element, npe); 1358 throw npe; 1359 } 1360 } 1361 else { 1362 writer.write(elementXml.toString()); 1363 } 1364 1365 if (queue.isEmpty()) { 1366 writer.flush(); 1367 } 1368 if (packet != null) { 1369 firePacketSendingListeners(packet); 1370 } 1371 } 1372 if (!instantShutdown) { 1373 // Flush out the rest of the queue. 1374 try { 1375 while (!queue.isEmpty()) { 1376 Element packet = queue.remove(); 1377 if (packet instanceof Stanza) { 1378 Stanza stanza = (Stanza) packet; 1379 maybeAddToUnacknowledgedStanzas(stanza); 1380 } 1381 writer.write(packet.toXML().toString()); 1382 } 1383 } 1384 catch (Exception e) { 1385 LOGGER.log(Level.WARNING, 1386 "Exception flushing queue during shutdown, ignore and continue", 1387 e); 1388 } 1389 1390 // Close the stream. 1391 try { 1392 writer.write("</stream:stream>"); 1393 writer.flush(); 1394 } 1395 catch (Exception e) { 1396 LOGGER.log(Level.WARNING, "Exception writing closing stream element", e); 1397 } 1398 1399 // Delete the queue contents (hopefully nothing is left). 1400 queue.clear(); 1401 } else if (instantShutdown && isSmEnabled()) { 1402 // This was an instantShutdown and SM is enabled, drain all remaining stanzas 1403 // into the unacknowledgedStanzas queue 1404 drainWriterQueueToUnacknowledgedStanzas(); 1405 } 1406 // Do *not* close the writer here, as it will cause the socket 1407 // to get closed. But we may want to receive further stanzas 1408 // until the closing stream tag is received. The socket will be 1409 // closed in shutdown(). 1410 } 1411 catch (Exception e) { 1412 // The exception can be ignored if the the connection is 'done' 1413 // or if the it was caused because the socket got closed 1414 if (!(done() || queue.isShutdown())) { 1415 // Set running to false since this thread will exit here and notifyConnectionError() will wait until 1416 // the reader and writer thread's 'running' value is false. 1417 running = false; 1418 notifyConnectionError(e); 1419 } else { 1420 LOGGER.log(Level.FINE, "Ignoring Exception in writePackets()", e); 1421 } 1422 } 1423 } 1424 1425 private void drainWriterQueueToUnacknowledgedStanzas() { 1426 List<Element> elements = new ArrayList<>(queue.size()); 1427 queue.drainTo(elements); 1428 for (int i = 0; i < elements.size(); i++) { 1429 Element element = elements.get(i); 1430 // If the unacknowledgedStanza queue is full, then bail out with a warning message. See SMACK-844. 1431 if (unacknowledgedStanzas.remainingCapacity() == 0) { 1432 StreamManagementException.UnacknowledgedQueueFullException exception = StreamManagementException.UnacknowledgedQueueFullException 1433 .newWith(i, elements, unacknowledgedStanzas); 1434 LOGGER.log(Level.WARNING, 1435 "Some stanzas may be lost as not all could be drained to the unacknowledged stanzas queue", exception); 1436 return; 1437 } 1438 if (element instanceof Stanza) { 1439 unacknowledgedStanzas.add((Stanza) element); 1440 } 1441 } 1442 } 1443 1444 private void maybeAddToUnacknowledgedStanzas(Stanza stanza) throws IOException { 1445 // Check if the stream element should be put to the unacknowledgedStanza 1446 // queue. Note that we can not do the put() in sendStanzaInternal() and the 1447 // packet order is not stable at this point (sendStanzaInternal() can be 1448 // called concurrently). 1449 if (unacknowledgedStanzas != null && stanza != null) { 1450 // If the unacknowledgedStanza queue reaching its high water mark, request an new ack 1451 // from the server in order to drain it 1452 if (unacknowledgedStanzas.size() == UNACKKNOWLEDGED_STANZAS_QUEUE_SIZE_HIGH_WATER_MARK) { 1453 writer.write(AckRequest.INSTANCE.toXML().toString()); 1454 } 1455 1456 try { 1457 // It is important the we put the stanza in the unacknowledged stanza 1458 // queue before we put it on the wire 1459 unacknowledgedStanzas.put(stanza); 1460 } 1461 catch (InterruptedException e) { 1462 throw new IllegalStateException(e); 1463 } 1464 } 1465 } 1466 } 1467 1468 /** 1469 * Set if Stream Management should be used by default for new connections. 1470 * 1471 * @param useSmDefault true to use Stream Management for new connections. 1472 */ 1473 public static void setUseStreamManagementDefault(boolean useSmDefault) { 1474 XMPPTCPConnection.useSmDefault = useSmDefault; 1475 } 1476 1477 /** 1478 * Set if Stream Management resumption should be used by default for new connections. 1479 * 1480 * @param useSmResumptionDefault true to use Stream Management resumption for new connections. 1481 * @deprecated use {@link #setUseStreamManagementResumptionDefault(boolean)} instead. 1482 */ 1483 @Deprecated 1484 public static void setUseStreamManagementResumptiodDefault(boolean useSmResumptionDefault) { 1485 setUseStreamManagementResumptionDefault(useSmResumptionDefault); 1486 } 1487 1488 /** 1489 * Set if Stream Management resumption should be used by default for new connections. 1490 * 1491 * @param useSmResumptionDefault true to use Stream Management resumption for new connections. 1492 */ 1493 public static void setUseStreamManagementResumptionDefault(boolean useSmResumptionDefault) { 1494 if (useSmResumptionDefault) { 1495 // Also enable SM is resumption is enabled 1496 setUseStreamManagementDefault(useSmResumptionDefault); 1497 } 1498 XMPPTCPConnection.useSmResumptionDefault = useSmResumptionDefault; 1499 } 1500 1501 /** 1502 * Set if Stream Management should be used if supported by the server. 1503 * 1504 * @param useSm true to use Stream Management. 1505 */ 1506 public void setUseStreamManagement(boolean useSm) { 1507 this.useSm = useSm; 1508 } 1509 1510 /** 1511 * Set if Stream Management resumption should be used if supported by the server. 1512 * 1513 * @param useSmResumption true to use Stream Management resumption. 1514 */ 1515 public void setUseStreamManagementResumption(boolean useSmResumption) { 1516 if (useSmResumption) { 1517 // Also enable SM is resumption is enabled 1518 setUseStreamManagement(useSmResumption); 1519 } 1520 this.useSmResumption = useSmResumption; 1521 } 1522 1523 /** 1524 * Set the preferred resumption time in seconds. 1525 * @param resumptionTime the preferred resumption time in seconds 1526 */ 1527 public void setPreferredResumptionTime(int resumptionTime) { 1528 smClientMaxResumptionTime = resumptionTime; 1529 } 1530 1531 /** 1532 * Add a predicate for Stream Management acknowledgment requests. 1533 * <p> 1534 * Those predicates are used to determine when a Stream Management acknowledgement request is send to the server. 1535 * Some pre-defined predicates are found in the <code>org.jivesoftware.smack.sm.predicates</code> package. 1536 * </p> 1537 * <p> 1538 * If not predicate is configured, the {@link Predicate#forMessagesOrAfter5Stanzas()} will be used. 1539 * </p> 1540 * 1541 * @param predicate the predicate to add. 1542 * @return if the predicate was not already active. 1543 */ 1544 public boolean addRequestAckPredicate(StanzaFilter predicate) { 1545 synchronized (requestAckPredicates) { 1546 return requestAckPredicates.add(predicate); 1547 } 1548 } 1549 1550 /** 1551 * Remove the given predicate for Stream Management acknowledgment request. 1552 * @param predicate the predicate to remove. 1553 * @return true if the predicate was removed. 1554 */ 1555 public boolean removeRequestAckPredicate(StanzaFilter predicate) { 1556 synchronized (requestAckPredicates) { 1557 return requestAckPredicates.remove(predicate); 1558 } 1559 } 1560 1561 /** 1562 * Remove all predicates for Stream Management acknowledgment requests. 1563 */ 1564 public void removeAllRequestAckPredicates() { 1565 synchronized (requestAckPredicates) { 1566 requestAckPredicates.clear(); 1567 } 1568 } 1569 1570 /** 1571 * Send an unconditional Stream Management acknowledgement request to the server. 1572 * 1573 * @throws StreamManagementNotEnabledException if Stream Management is not enabled. 1574 * @throws NotConnectedException if the connection is not connected. 1575 * @throws InterruptedException if the calling thread was interrupted. 1576 */ 1577 public void requestSmAcknowledgement() throws StreamManagementNotEnabledException, NotConnectedException, InterruptedException { 1578 if (!isSmEnabled()) { 1579 throw new StreamManagementException.StreamManagementNotEnabledException(); 1580 } 1581 requestSmAcknowledgementInternal(); 1582 } 1583 1584 private void requestSmAcknowledgementInternal() throws NotConnectedException, InterruptedException { 1585 packetWriter.sendStreamElement(AckRequest.INSTANCE); 1586 } 1587 1588 /** 1589 * Send a unconditional Stream Management acknowledgment to the server. 1590 * <p> 1591 * See <a href="http://xmpp.org/extensions/xep-0198.html#acking">XEP-198: Stream Management ยง 4. Acks</a>: 1592 * "Either party MAY send an <a/> element at any time (e.g., after it has received a certain number of stanzas, 1593 * or after a certain period of time), even if it has not received an <r/> element from the other party." 1594 * </p> 1595 * 1596 * @throws StreamManagementNotEnabledException if Stream Management is not enabled. 1597 * @throws NotConnectedException if the connection is not connected. 1598 * @throws InterruptedException if the calling thread was interrupted. 1599 */ 1600 public void sendSmAcknowledgement() throws StreamManagementNotEnabledException, NotConnectedException, InterruptedException { 1601 if (!isSmEnabled()) { 1602 throw new StreamManagementException.StreamManagementNotEnabledException(); 1603 } 1604 sendSmAcknowledgementInternal(); 1605 } 1606 1607 private void sendSmAcknowledgementInternal() throws NotConnectedException, InterruptedException { 1608 AckAnswer ackAnswer = new AckAnswer(clientHandledStanzasCount); 1609 // Do net put an ack to the queue if it has already been shutdown. Some servers, like ejabberd, like to request 1610 // an ack even after we have send a stream close (and hance the queue was shutdown). If we would not check here, 1611 // then the ack would dangle around in the queue, and be send on the next re-connection attempt even before the 1612 // stream open. 1613 packetWriter.queue.putIfNotShutdown(ackAnswer); 1614 } 1615 1616 /** 1617 * Add a Stanza acknowledged listener. 1618 * <p> 1619 * Those listeners will be invoked every time a Stanza has been acknowledged by the server. The will not get 1620 * automatically removed. Consider using {@link #addStanzaIdAcknowledgedListener(String, StanzaListener)} when 1621 * possible. 1622 * </p> 1623 * 1624 * @param listener the listener to add. 1625 */ 1626 public void addStanzaAcknowledgedListener(StanzaListener listener) { 1627 stanzaAcknowledgedListeners.add(listener); 1628 } 1629 1630 /** 1631 * Remove the given Stanza acknowledged listener. 1632 * 1633 * @param listener the listener. 1634 * @return true if the listener was removed. 1635 */ 1636 public boolean removeStanzaAcknowledgedListener(StanzaListener listener) { 1637 return stanzaAcknowledgedListeners.remove(listener); 1638 } 1639 1640 /** 1641 * Remove all stanza acknowledged listeners. 1642 */ 1643 public void removeAllStanzaAcknowledgedListeners() { 1644 stanzaAcknowledgedListeners.clear(); 1645 } 1646 1647 /** 1648 * Add a Stanza dropped listener. 1649 * <p> 1650 * Those listeners will be invoked every time a Stanza has been dropped due to a failed SM resume. They will not get 1651 * automatically removed. If at least one StanzaDroppedListener is configured, no attempt will be made to retransmit 1652 * the Stanzas. 1653 * </p> 1654 * 1655 * @param listener the listener to add. 1656 * @since 4.3.3 1657 */ 1658 public void addStanzaDroppedListener(StanzaListener listener) { 1659 stanzaDroppedListeners.add(listener); 1660 } 1661 1662 /** 1663 * Remove the given Stanza dropped listener. 1664 * 1665 * @param listener the listener. 1666 * @return true if the listener was removed. 1667 * @since 4.3.3 1668 */ 1669 public boolean removeStanzaDroppedListener(StanzaListener listener) { 1670 return stanzaDroppedListeners.remove(listener); 1671 } 1672 1673 /** 1674 * Add a new Stanza ID acknowledged listener for the given ID. 1675 * <p> 1676 * The listener will be invoked if the stanza with the given ID was acknowledged by the server. It will 1677 * automatically be removed after the listener was run. 1678 * </p> 1679 * 1680 * @param id the stanza ID. 1681 * @param listener the listener to invoke. 1682 * @return the previous listener for this stanza ID or null. 1683 * @throws StreamManagementNotEnabledException if Stream Management is not enabled. 1684 */ 1685 @SuppressWarnings("FutureReturnValueIgnored") 1686 public StanzaListener addStanzaIdAcknowledgedListener(final String id, StanzaListener listener) throws StreamManagementNotEnabledException { 1687 // Prevent users from adding callbacks that will never get removed 1688 if (!smWasEnabledAtLeastOnce) { 1689 throw new StreamManagementException.StreamManagementNotEnabledException(); 1690 } 1691 // Remove the listener after max. 3 hours 1692 final int removeAfterSeconds = Math.min(getMaxSmResumptionTime(), 3 * 60 * 60); 1693 schedule(new Runnable() { 1694 @Override 1695 public void run() { 1696 stanzaIdAcknowledgedListeners.remove(id); 1697 } 1698 }, removeAfterSeconds, TimeUnit.SECONDS); 1699 return stanzaIdAcknowledgedListeners.put(id, listener); 1700 } 1701 1702 /** 1703 * Remove the Stanza ID acknowledged listener for the given ID. 1704 * 1705 * @param id the stanza ID. 1706 * @return true if the listener was found and removed, false otherwise. 1707 */ 1708 public StanzaListener removeStanzaIdAcknowledgedListener(String id) { 1709 return stanzaIdAcknowledgedListeners.remove(id); 1710 } 1711 1712 /** 1713 * Removes all Stanza ID acknowledged listeners. 1714 */ 1715 public void removeAllStanzaIdAcknowledgedListeners() { 1716 stanzaIdAcknowledgedListeners.clear(); 1717 } 1718 1719 /** 1720 * Returns true if Stream Management is supported by the server. 1721 * 1722 * @return true if Stream Management is supported by the server. 1723 */ 1724 public boolean isSmAvailable() { 1725 return hasFeature(StreamManagementFeature.ELEMENT, StreamManagement.NAMESPACE); 1726 } 1727 1728 /** 1729 * Returns true if Stream Management was successfully negotiated with the server. 1730 * 1731 * @return true if Stream Management was negotiated. 1732 */ 1733 public boolean isSmEnabled() { 1734 return smEnabledSyncPoint; 1735 } 1736 1737 /** 1738 * Returns true if the stream was successfully resumed with help of Stream Management. 1739 * 1740 * @return true if the stream was resumed. 1741 */ 1742 public boolean streamWasResumed() { 1743 return smResumedSyncPoint == SyncPointState.successful; 1744 } 1745 1746 /** 1747 * Returns true if the connection is disconnected by a Stream resumption via Stream Management is possible. 1748 * 1749 * @return true if disconnected but resumption possible. 1750 */ 1751 public boolean isDisconnectedButSmResumptionPossible() { 1752 return disconnectedButResumeable && isSmResumptionPossible(); 1753 } 1754 1755 /** 1756 * Returns true if the stream is resumable. 1757 * 1758 * @return true if the stream is resumable. 1759 */ 1760 public boolean isSmResumptionPossible() { 1761 // There is no resumable stream available 1762 if (smSessionId == null) 1763 return false; 1764 1765 final Long shutdownTimestamp = packetWriter.shutdownTimestamp; 1766 // Seems like we are already reconnected, report true 1767 if (shutdownTimestamp == null) { 1768 return true; 1769 } 1770 1771 // See if resumption time is over 1772 long current = System.currentTimeMillis(); 1773 long maxResumptionMillies = ((long) getMaxSmResumptionTime()) * 1000; 1774 if (current > shutdownTimestamp + maxResumptionMillies) { 1775 // Stream resumption is *not* possible if the current timestamp is greater then the greatest timestamp where 1776 // resumption is possible 1777 return false; 1778 } else { 1779 return true; 1780 } 1781 } 1782 1783 /** 1784 * Drop the stream management state. Sets {@link #smSessionId} and 1785 * {@link #unacknowledgedStanzas} to <code>null</code>. 1786 */ 1787 private void dropSmState() { 1788 // clientHandledCount and serverHandledCount will be reset on <enable/> and <enabled/> 1789 // respective. No need to reset them here. 1790 smSessionId = null; 1791 unacknowledgedStanzas = null; 1792 } 1793 1794 /** 1795 * Get the maximum resumption time in seconds after which a managed stream can be resumed. 1796 * <p> 1797 * This method will return {@link Integer#MAX_VALUE} if neither the client nor the server specify a maximum 1798 * resumption time. Be aware of integer overflows when using this value, e.g. do not add arbitrary values to it 1799 * without checking for overflows before. 1800 * </p> 1801 * 1802 * @return the maximum resumption time in seconds or {@link Integer#MAX_VALUE} if none set. 1803 */ 1804 public int getMaxSmResumptionTime() { 1805 int clientResumptionTime = smClientMaxResumptionTime > 0 ? smClientMaxResumptionTime : Integer.MAX_VALUE; 1806 int serverResumptionTime = smServerMaxResumptionTime > 0 ? smServerMaxResumptionTime : Integer.MAX_VALUE; 1807 return Math.min(clientResumptionTime, serverResumptionTime); 1808 } 1809 1810 private void processHandledCount(long handledCount) throws StreamManagementCounterError { 1811 long ackedStanzasCount = SMUtils.calculateDelta(handledCount, serverHandledStanzasCount); 1812 final List<Stanza> ackedStanzas = new ArrayList<>( 1813 ackedStanzasCount <= Integer.MAX_VALUE ? (int) ackedStanzasCount 1814 : Integer.MAX_VALUE); 1815 for (long i = 0; i < ackedStanzasCount; i++) { 1816 Stanza ackedStanza = unacknowledgedStanzas.poll(); 1817 // If the server ack'ed a stanza, then it must be in the 1818 // unacknowledged stanza queue. There can be no exception. 1819 if (ackedStanza == null) { 1820 throw new StreamManagementCounterError(handledCount, serverHandledStanzasCount, 1821 ackedStanzasCount, ackedStanzas); 1822 } 1823 ackedStanzas.add(ackedStanza); 1824 } 1825 1826 boolean atLeastOneStanzaAcknowledgedListener = false; 1827 if (!stanzaAcknowledgedListeners.isEmpty()) { 1828 // If stanzaAcknowledgedListeners is not empty, the we have at least one 1829 atLeastOneStanzaAcknowledgedListener = true; 1830 } 1831 else { 1832 // Otherwise we look for a matching id in the stanza *id* acknowledged listeners 1833 for (Stanza ackedStanza : ackedStanzas) { 1834 String id = ackedStanza.getStanzaId(); 1835 if (id != null && stanzaIdAcknowledgedListeners.containsKey(id)) { 1836 atLeastOneStanzaAcknowledgedListener = true; 1837 break; 1838 } 1839 } 1840 } 1841 1842 // Only spawn a new thread if there is a chance that some listener is invoked 1843 if (atLeastOneStanzaAcknowledgedListener) { 1844 asyncGo(new Runnable() { 1845 @Override 1846 public void run() { 1847 for (Stanza ackedStanza : ackedStanzas) { 1848 for (StanzaListener listener : stanzaAcknowledgedListeners) { 1849 try { 1850 listener.processStanza(ackedStanza); 1851 } 1852 catch (InterruptedException | NotConnectedException | NotLoggedInException e) { 1853 LOGGER.log(Level.FINER, "Received exception", e); 1854 } 1855 } 1856 String id = ackedStanza.getStanzaId(); 1857 if (StringUtils.isNullOrEmpty(id)) { 1858 continue; 1859 } 1860 StanzaListener listener = stanzaIdAcknowledgedListeners.remove(id); 1861 if (listener != null) { 1862 try { 1863 listener.processStanza(ackedStanza); 1864 } 1865 catch (InterruptedException | NotConnectedException | NotLoggedInException e) { 1866 LOGGER.log(Level.FINER, "Received exception", e); 1867 } 1868 } 1869 } 1870 } 1871 }); 1872 } 1873 1874 serverHandledStanzasCount = handledCount; 1875 } 1876 1877 /** 1878 * Set the default bundle and defer callback used for new connections. 1879 * 1880 * @param defaultBundleAndDeferCallback TODO javadoc me please 1881 * @see BundleAndDeferCallback 1882 * @since 4.1 1883 */ 1884 public static void setDefaultBundleAndDeferCallback(BundleAndDeferCallback defaultBundleAndDeferCallback) { 1885 XMPPTCPConnection.defaultBundleAndDeferCallback = defaultBundleAndDeferCallback; 1886 } 1887 1888 /** 1889 * Set the bundle and defer callback used for this connection. 1890 * <p> 1891 * You can use <code>null</code> as argument to reset the callback. Outgoing stanzas will then 1892 * no longer get deferred. 1893 * </p> 1894 * 1895 * @param bundleAndDeferCallback the callback or <code>null</code>. 1896 * @see BundleAndDeferCallback 1897 * @since 4.1 1898 */ 1899 public void setBundleandDeferCallback(BundleAndDeferCallback bundleAndDeferCallback) { 1900 this.bundleAndDeferCallback = bundleAndDeferCallback; 1901 } 1902 1903}