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