001/** 002 * 003 * Copyright 2009 Jive Software, 2018-2022 Florian Schmaus. 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; 018 019import java.io.IOException; 020import java.io.Reader; 021import java.io.Writer; 022import java.util.Collection; 023import java.util.HashMap; 024import java.util.Iterator; 025import java.util.LinkedHashMap; 026import java.util.LinkedList; 027import java.util.List; 028import java.util.Map; 029import java.util.Queue; 030import java.util.Set; 031import java.util.concurrent.ConcurrentLinkedQueue; 032import java.util.concurrent.CopyOnWriteArraySet; 033import java.util.concurrent.Executor; 034import java.util.concurrent.ExecutorService; 035import java.util.concurrent.Executors; 036import java.util.concurrent.ThreadFactory; 037import java.util.concurrent.TimeUnit; 038import java.util.concurrent.atomic.AtomicInteger; 039import java.util.concurrent.locks.Lock; 040import java.util.concurrent.locks.ReentrantLock; 041import java.util.logging.Level; 042import java.util.logging.Logger; 043 044import javax.net.ssl.SSLSession; 045import javax.xml.namespace.QName; 046 047import org.jivesoftware.smack.ConnectionConfiguration.SecurityMode; 048import org.jivesoftware.smack.SmackConfiguration.UnknownIqRequestReplyMode; 049import org.jivesoftware.smack.SmackException.AlreadyConnectedException; 050import org.jivesoftware.smack.SmackException.AlreadyLoggedInException; 051import org.jivesoftware.smack.SmackException.NoResponseException; 052import org.jivesoftware.smack.SmackException.NotConnectedException; 053import org.jivesoftware.smack.SmackException.NotLoggedInException; 054import org.jivesoftware.smack.SmackException.ResourceBindingNotOfferedException; 055import org.jivesoftware.smack.SmackException.SecurityRequiredByClientException; 056import org.jivesoftware.smack.SmackException.SecurityRequiredException; 057import org.jivesoftware.smack.SmackException.SmackSaslException; 058import org.jivesoftware.smack.SmackException.SmackWrappedException; 059import org.jivesoftware.smack.SmackFuture.InternalSmackFuture; 060import org.jivesoftware.smack.XMPPException.FailedNonzaException; 061import org.jivesoftware.smack.XMPPException.StreamErrorException; 062import org.jivesoftware.smack.XMPPException.XMPPErrorException; 063import org.jivesoftware.smack.compress.packet.Compress; 064import org.jivesoftware.smack.compression.XMPPInputOutputStream; 065import org.jivesoftware.smack.datatypes.UInt16; 066import org.jivesoftware.smack.debugger.SmackDebugger; 067import org.jivesoftware.smack.debugger.SmackDebuggerFactory; 068import org.jivesoftware.smack.filter.IQReplyFilter; 069import org.jivesoftware.smack.filter.StanzaFilter; 070import org.jivesoftware.smack.filter.StanzaIdFilter; 071import org.jivesoftware.smack.internal.SmackTlsContext; 072import org.jivesoftware.smack.iqrequest.IQRequestHandler; 073import org.jivesoftware.smack.packet.Bind; 074import org.jivesoftware.smack.packet.ErrorIQ; 075import org.jivesoftware.smack.packet.ExtensionElement; 076import org.jivesoftware.smack.packet.FullyQualifiedElement; 077import org.jivesoftware.smack.packet.IQ; 078import org.jivesoftware.smack.packet.Mechanisms; 079import org.jivesoftware.smack.packet.Message; 080import org.jivesoftware.smack.packet.MessageBuilder; 081import org.jivesoftware.smack.packet.MessageOrPresence; 082import org.jivesoftware.smack.packet.MessageOrPresenceBuilder; 083import org.jivesoftware.smack.packet.Nonza; 084import org.jivesoftware.smack.packet.Presence; 085import org.jivesoftware.smack.packet.PresenceBuilder; 086import org.jivesoftware.smack.packet.Session; 087import org.jivesoftware.smack.packet.Stanza; 088import org.jivesoftware.smack.packet.StanzaError; 089import org.jivesoftware.smack.packet.StanzaFactory; 090import org.jivesoftware.smack.packet.StartTls; 091import org.jivesoftware.smack.packet.StreamError; 092import org.jivesoftware.smack.packet.StreamOpen; 093import org.jivesoftware.smack.packet.TopLevelStreamElement; 094import org.jivesoftware.smack.packet.XmlEnvironment; 095import org.jivesoftware.smack.packet.id.StanzaIdSource; 096import org.jivesoftware.smack.parsing.ParsingExceptionCallback; 097import org.jivesoftware.smack.parsing.SmackParsingException; 098import org.jivesoftware.smack.provider.ExtensionElementProvider; 099import org.jivesoftware.smack.provider.NonzaProvider; 100import org.jivesoftware.smack.provider.ProviderManager; 101import org.jivesoftware.smack.sasl.SASLErrorException; 102import org.jivesoftware.smack.sasl.SASLMechanism; 103import org.jivesoftware.smack.sasl.core.SASLAnonymous; 104import org.jivesoftware.smack.sasl.packet.SaslNonza; 105import org.jivesoftware.smack.util.Async; 106import org.jivesoftware.smack.util.CollectionUtil; 107import org.jivesoftware.smack.util.Consumer; 108import org.jivesoftware.smack.util.MultiMap; 109import org.jivesoftware.smack.util.Objects; 110import org.jivesoftware.smack.util.PacketParserUtils; 111import org.jivesoftware.smack.util.ParserUtils; 112import org.jivesoftware.smack.util.Predicate; 113import org.jivesoftware.smack.util.StringUtils; 114import org.jivesoftware.smack.util.Supplier; 115import org.jivesoftware.smack.xml.XmlPullParser; 116import org.jivesoftware.smack.xml.XmlPullParserException; 117 118import org.jxmpp.jid.DomainBareJid; 119import org.jxmpp.jid.EntityBareJid; 120import org.jxmpp.jid.EntityFullJid; 121import org.jxmpp.jid.Jid; 122import org.jxmpp.jid.impl.JidCreate; 123import org.jxmpp.jid.parts.Resourcepart; 124import org.jxmpp.stringprep.XmppStringprepException; 125import org.jxmpp.util.XmppStringUtils; 126 127/** 128 * This abstract class is commonly used as super class for XMPP connection mechanisms like TCP and BOSH. Hence it 129 * provides the methods for connection state management, like {@link #connect()}, {@link #login()} and 130 * {@link #disconnect()} (which are deliberately not provided by the {@link XMPPConnection} interface). 131 * <p> 132 * <b>Note:</b> The default entry point to Smack's documentation is {@link XMPPConnection}. If you are getting started 133 * with Smack, then head over to {@link XMPPConnection} and the come back here. 134 * </p> 135 * <h2>Parsing Exceptions</h2> 136 * <p> 137 * In case a Smack parser (Provider) throws those exceptions are handled over to the {@link ParsingExceptionCallback}. A 138 * common cause for a provider throwing is illegal input, for example a non-numeric String where only Integers are 139 * allowed. Smack's <em>default behavior</em> follows the <b>"fail-hard per default"</b> principle leading to a 140 * termination of the connection on parsing exceptions. This default was chosen to make users eventually aware that they 141 * should configure their own callback and handle those exceptions to prevent the disconnect. Handle a parsing exception 142 * could be as simple as using a non-throwing no-op callback, which would cause the faulty stream element to be taken 143 * out of the stream, i.e., Smack behaves like that element was never received. 144 * </p> 145 * <p> 146 * If the parsing exception is because Smack received illegal input, then please consider informing the authors of the 147 * originating entity about that. If it was thrown because of an bug in a Smack parser, then please consider filling a 148 * bug with Smack. 149 * </p> 150 * <h3>Managing the parsing exception callback</h3> 151 * <p> 152 * The "fail-hard per default" behavior is achieved by using the 153 * {@link org.jivesoftware.smack.parsing.ExceptionThrowingCallbackWithHint} as default parsing exception callback. You 154 * can change the behavior using {@link #setParsingExceptionCallback(ParsingExceptionCallback)} to set a new callback. 155 * Use {@link org.jivesoftware.smack.SmackConfiguration#setDefaultParsingExceptionCallback(ParsingExceptionCallback)} to 156 * set the default callback. 157 * </p> 158 */ 159public abstract class AbstractXMPPConnection implements XMPPConnection { 160 private static final Logger LOGGER = Logger.getLogger(AbstractXMPPConnection.class.getName()); 161 162 protected static final SmackReactor SMACK_REACTOR; 163 164 static { 165 SMACK_REACTOR = SmackReactor.getInstance(); 166 } 167 168 /** 169 * Counter to uniquely identify connections that are created. 170 */ 171 private static final AtomicInteger connectionCounter = new AtomicInteger(0); 172 173 static { 174 Smack.ensureInitialized(); 175 } 176 177 protected enum SyncPointState { 178 initial, 179 request_sent, 180 successful, 181 } 182 183 /** 184 * A collection of ConnectionListeners which listen for connection closing 185 * and reconnection events. 186 */ 187 protected final Set<ConnectionListener> connectionListeners = 188 new CopyOnWriteArraySet<>(); 189 190 /** 191 * A collection of StanzaCollectors which collects packets for a specified filter 192 * and perform blocking and polling operations on the result queue. 193 * <p> 194 * We use a ConcurrentLinkedQueue here, because its Iterator is weakly 195 * consistent and we want {@link #invokeStanzaCollectorsAndNotifyRecvListeners(Stanza)} for-each 196 * loop to be lock free. As drawback, removing a StanzaCollector is O(n). 197 * The alternative would be a synchronized HashSet, but this would mean a 198 * synchronized block around every usage of <code>collectors</code>. 199 * </p> 200 */ 201 private final Collection<StanzaCollector> collectors = new ConcurrentLinkedQueue<>(); 202 203 private final Map<StanzaListener, ListenerWrapper> recvListeners = new LinkedHashMap<>(); 204 205 /** 206 * List of PacketListeners that will be notified synchronously when a new stanza was received. 207 */ 208 private final Map<StanzaListener, ListenerWrapper> syncRecvListeners = new LinkedHashMap<>(); 209 210 /** 211 * List of PacketListeners that will be notified asynchronously when a new stanza was received. 212 */ 213 private final Map<StanzaListener, ListenerWrapper> asyncRecvListeners = new LinkedHashMap<>(); 214 215 /** 216 * List of PacketListeners that will be notified when a new stanza was sent. 217 */ 218 private final Map<StanzaListener, ListenerWrapper> sendListeners = 219 new HashMap<>(); 220 221 /** 222 * List of PacketListeners that will be notified when a new stanza is about to be 223 * sent to the server. These interceptors may modify the stanza before it is being 224 * actually sent to the server. 225 */ 226 private final Map<StanzaListener, InterceptorWrapper> interceptors = 227 new HashMap<>(); 228 229 private final Map<Consumer<MessageBuilder>, GenericInterceptorWrapper<MessageBuilder, Message>> messageInterceptors = new HashMap<>(); 230 231 private final Map<Consumer<PresenceBuilder>, GenericInterceptorWrapper<PresenceBuilder, Presence>> presenceInterceptors = new HashMap<>(); 232 233 private XmlEnvironment incomingStreamXmlEnvironment; 234 235 protected XmlEnvironment outgoingStreamXmlEnvironment; 236 237 final MultiMap<QName, NonzaCallback> nonzaCallbacksMap = new MultiMap<>(); 238 239 protected final Lock connectionLock = new ReentrantLock(); 240 241 protected final Map<QName, FullyQualifiedElement> streamFeatures = new HashMap<>(); 242 243 /** 244 * The full JID of the authenticated user, as returned by the resource binding response of the server. 245 * <p> 246 * It is important that we don't infer the user from the login() arguments and the configurations service name, as, 247 * for example, when SASL External is used, the username is not given to login but taken from the 'external' 248 * certificate. 249 * </p> 250 */ 251 protected EntityFullJid user; 252 253 protected boolean connected = false; 254 255 /** 256 * The stream ID, see RFC 6120 § 4.7.3 257 */ 258 protected String streamId; 259 260 /** 261 * The timeout to wait for a reply in milliseconds. 262 */ 263 private long replyTimeout = SmackConfiguration.getDefaultReplyTimeout(); 264 265 /** 266 * The SmackDebugger allows to log and debug XML traffic. 267 */ 268 protected final SmackDebugger debugger; 269 270 /** 271 * The Reader which is used for the debugger. 272 */ 273 protected Reader reader; 274 275 /** 276 * The Writer which is used for the debugger. 277 */ 278 protected Writer writer; 279 280 protected SmackException currentSmackException; 281 protected XMPPException currentXmppException; 282 283 protected boolean tlsHandled; 284 285 /** 286 * Set to <code>true</code> if the last features stanza from the server has been parsed. A XMPP connection 287 * handshake can invoke multiple features stanzas, e.g. when TLS is activated a second feature 288 * stanza is send by the server. This is set to true once the last feature stanza has been 289 * parsed. 290 */ 291 protected boolean lastFeaturesReceived; 292 293 /** 294 * Set to <code>true</code> if the SASL feature has been received. 295 */ 296 protected boolean saslFeatureReceived; 297 298 /** 299 * A synchronization point which is successful if this connection has received the closing 300 * stream element from the remote end-point, i.e. the server. 301 */ 302 protected boolean closingStreamReceived; 303 304 /** 305 * The SASLAuthentication manager that is responsible for authenticating with the server. 306 */ 307 private final SASLAuthentication saslAuthentication; 308 309 /** 310 * A number to uniquely identify connections that are created. This is distinct from the 311 * connection ID, which is a value sent by the server once a connection is made. 312 */ 313 protected final int connectionCounterValue = connectionCounter.getAndIncrement(); 314 315 /** 316 * Holds the initial configuration used while creating the connection. 317 */ 318 protected final ConnectionConfiguration config; 319 320 /** 321 * Defines how the from attribute of outgoing stanzas should be handled. 322 */ 323 private FromMode fromMode = FromMode.OMITTED; 324 325 protected XMPPInputOutputStream compressionHandler; 326 327 private ParsingExceptionCallback parsingExceptionCallback = SmackConfiguration.getDefaultParsingExceptionCallback(); 328 329 /** 330 * A cached thread pool executor service with custom thread factory to set meaningful names on the threads and set 331 * them 'daemon'. 332 */ 333 private static final ExecutorService CACHED_EXECUTOR_SERVICE = Executors.newCachedThreadPool(new ThreadFactory() { 334 @Override 335 public Thread newThread(Runnable runnable) { 336 Thread thread = new Thread(runnable); 337 thread.setName("Smack Cached Executor"); 338 thread.setDaemon(true); 339 thread.setUncaughtExceptionHandler(new Thread.UncaughtExceptionHandler() { 340 @Override 341 public void uncaughtException(Thread t, Throwable e) { 342 LOGGER.log(Level.WARNING, t + " encountered uncaught exception", e); 343 } 344 }); 345 return thread; 346 } 347 }); 348 349 protected static final AsyncButOrdered<AbstractXMPPConnection> ASYNC_BUT_ORDERED = new AsyncButOrdered<>(); 350 351 protected final AsyncButOrdered<StanzaListener> inOrderListeners = new AsyncButOrdered<>(); 352 353 /** 354 * The used host to establish the connection to 355 */ 356 protected String host; 357 358 /** 359 * The used port to establish the connection to 360 */ 361 protected UInt16 port; 362 363 /** 364 * Flag that indicates if the user is currently authenticated with the server. 365 */ 366 protected boolean authenticated = false; 367 368 // TODO: Migrate to ZonedDateTime once Smack's minimum required Android SDK level is 26 (8.0, Oreo) or higher. 369 protected long authenticatedConnectionInitiallyEstablishedTimestamp; 370 371 /** 372 * Flag that indicates if the user was authenticated with the server when the connection 373 * to the server was closed (abruptly or not). 374 */ 375 protected boolean wasAuthenticated = false; 376 377 private final Map<QName, IQRequestHandler> setIqRequestHandler = new HashMap<>(); 378 private final Map<QName, IQRequestHandler> getIqRequestHandler = new HashMap<>(); 379 380 private final StanzaFactory stanzaFactory; 381 382 /** 383 * Create a new XMPPConnection to an XMPP server. 384 * 385 * @param configuration The configuration which is used to establish the connection. 386 */ 387 protected AbstractXMPPConnection(ConnectionConfiguration configuration) { 388 saslAuthentication = new SASLAuthentication(this, configuration); 389 config = configuration; 390 391 // Install the SASL Nonza callbacks. 392 buildNonzaCallback() 393 .listenFor(SaslNonza.Challenge.class, c -> { 394 try { 395 saslAuthentication.challengeReceived(c); 396 } catch (SmackException | InterruptedException e) { 397 saslAuthentication.authenticationFailed(e); 398 } 399 }) 400 .listenFor(SaslNonza.Success.class, s -> { 401 try { 402 saslAuthentication.authenticated(s); 403 } catch (SmackSaslException | NotConnectedException | InterruptedException e) { 404 saslAuthentication.authenticationFailed(e); 405 } 406 }) 407 .listenFor(SaslNonza.SASLFailure.class, f -> saslAuthentication.authenticationFailed(f)) 408 .install(); 409 410 SmackDebuggerFactory debuggerFactory = configuration.getDebuggerFactory(); 411 if (debuggerFactory != null) { 412 debugger = debuggerFactory.create(this); 413 } else { 414 debugger = null; 415 } 416 // Notify listeners that a new connection has been established 417 for (ConnectionCreationListener listener : XMPPConnectionRegistry.getConnectionCreationListeners()) { 418 listener.connectionCreated(this); 419 } 420 421 StanzaIdSource stanzaIdSource = configuration.constructStanzaIdSource(); 422 stanzaFactory = new StanzaFactory(stanzaIdSource); 423 } 424 425 /** 426 * Get the connection configuration used by this connection. 427 * 428 * @return the connection configuration. 429 */ 430 public ConnectionConfiguration getConfiguration() { 431 return config; 432 } 433 434 @Override 435 public DomainBareJid getXMPPServiceDomain() { 436 if (xmppServiceDomain != null) { 437 return xmppServiceDomain; 438 } 439 return config.getXMPPServiceDomain(); 440 } 441 442 @Override 443 public String getHost() { 444 return host; 445 } 446 447 @Override 448 public int getPort() { 449 final UInt16 port = this.port; 450 if (port == null) { 451 return -1; 452 } 453 454 return port.intValue(); 455 } 456 457 @Override 458 public abstract boolean isSecureConnection(); 459 460 protected abstract void sendStanzaInternal(Stanza packet) throws NotConnectedException, InterruptedException; 461 462 @Override 463 public boolean trySendStanza(Stanza stanza) throws NotConnectedException { 464 // Default implementation which falls back to sendStanza() as mentioned in the methods javadoc. May be 465 // overwritten by subclasses. 466 try { 467 sendStanza(stanza); 468 } catch (InterruptedException e) { 469 LOGGER.log(Level.FINER, 470 "Thread blocked in fallback implementation of trySendStanza(Stanza) was interrupted", e); 471 return false; 472 } 473 return true; 474 } 475 476 @Override 477 public boolean trySendStanza(Stanza stanza, long timeout, TimeUnit unit) 478 throws NotConnectedException, InterruptedException { 479 // Default implementation which falls back to sendStanza() as mentioned in the methods javadoc. May be 480 // overwritten by subclasses. 481 sendStanza(stanza); 482 return true; 483 } 484 485 @Override 486 public abstract void sendNonza(Nonza element) throws NotConnectedException, InterruptedException; 487 488 @Override 489 public abstract boolean isUsingCompression(); 490 491 protected void initState() { 492 currentSmackException = null; 493 currentXmppException = null; 494 saslFeatureReceived = lastFeaturesReceived = tlsHandled = false; 495 // TODO: We do not init closingStreamReceived here, as the integration tests use it to check if we waited for 496 // it. 497 } 498 499 /** 500 * Establishes a connection to the XMPP server. It basically 501 * creates and maintains a connection to the server. 502 * <p> 503 * Listeners will be preserved from a previous connection. 504 * </p> 505 * 506 * @throws XMPPException if an error occurs on the XMPP protocol level. 507 * @throws SmackException if an error occurs somewhere else besides XMPP protocol level. 508 * @throws IOException if an I/O error occurred. 509 * @return a reference to this object, to chain <code>connect()</code> with <code>login()</code>. 510 * @throws InterruptedException if the calling thread was interrupted. 511 */ 512 public synchronized AbstractXMPPConnection connect() throws SmackException, IOException, XMPPException, InterruptedException { 513 // Check if not already connected 514 throwAlreadyConnectedExceptionIfAppropriate(); 515 516 // Notify connection listeners that we are trying to connect 517 callConnectionConnectingListener(); 518 519 // Reset the connection state 520 initState(); 521 closingStreamReceived = false; 522 streamId = null; 523 524 try { 525 // Perform the actual connection to the XMPP service 526 connectInternal(); 527 528 // If TLS is required but the server doesn't offer it, disconnect 529 // from the server and throw an error. First check if we've already negotiated TLS 530 // and are secure, however (features get parsed a second time after TLS is established). 531 if (!isSecureConnection() && getConfiguration().getSecurityMode() == SecurityMode.required) { 532 throw new SecurityRequiredByClientException(); 533 } 534 } catch (SmackException | IOException | XMPPException | InterruptedException e) { 535 instantShutdown(); 536 throw e; 537 } 538 539 // If connectInternal() did not throw, then this connection must now be marked as connected. 540 assert connected; 541 542 callConnectionConnectedListener(); 543 544 return this; 545 } 546 547 /** 548 * Abstract method that concrete subclasses of XMPPConnection need to implement to perform their 549 * way of XMPP connection establishment. Implementations are required to perform an automatic 550 * login if the previous connection state was logged (authenticated). 551 * 552 * @throws SmackException if Smack detected an exceptional situation. 553 * @throws IOException if an I/O error occurred. 554 * @throws XMPPException if an XMPP protocol error was received. 555 * @throws InterruptedException if the calling thread was interrupted. 556 */ 557 protected abstract void connectInternal() throws SmackException, IOException, XMPPException, InterruptedException; 558 559 private String usedUsername, usedPassword; 560 561 /** 562 * The resourcepart used for this connection. May not be the resulting resourcepart if it's null or overridden by the XMPP service. 563 */ 564 private Resourcepart usedResource; 565 566 /** 567 * Logs in to the server using the strongest SASL mechanism supported by 568 * the server. If more than the connection's default stanza timeout elapses in each step of the 569 * authentication process without a response from the server, a 570 * {@link SmackException.NoResponseException} will be thrown. 571 * <p> 572 * Before logging in (i.e. authenticate) to the server the connection must be connected 573 * by calling {@link #connect}. 574 * </p> 575 * <p> 576 * It is possible to log in without sending an initial available presence by using 577 * {@link ConnectionConfiguration.Builder#setSendPresence(boolean)}. 578 * Finally, if you want to not pass a password and instead use a more advanced mechanism 579 * while using SASL then you may be interested in using 580 * {@link ConnectionConfiguration.Builder#setCallbackHandler(javax.security.auth.callback.CallbackHandler)}. 581 * For more advanced login settings see {@link ConnectionConfiguration}. 582 * </p> 583 * 584 * @throws XMPPException if an error occurs on the XMPP protocol level. 585 * @throws SmackException if an error occurs somewhere else besides XMPP protocol level. 586 * @throws IOException if an I/O error occurs during login. 587 * @throws InterruptedException if the calling thread was interrupted. 588 */ 589 public synchronized void login() throws XMPPException, SmackException, IOException, InterruptedException { 590 // The previously used username, password and resource take over precedence over the 591 // ones from the connection configuration 592 CharSequence username = usedUsername != null ? usedUsername : config.getUsername(); 593 String password = usedPassword != null ? usedPassword : config.getPassword(); 594 Resourcepart resource = usedResource != null ? usedResource : config.getResource(); 595 login(username, password, resource); 596 } 597 598 /** 599 * Same as {@link #login(CharSequence, String, Resourcepart)}, but takes the resource from the connection 600 * configuration. 601 * 602 * @param username TODO javadoc me please 603 * @param password TODO javadoc me please 604 * @throws XMPPException if an XMPP protocol error was received. 605 * @throws SmackException if Smack detected an exceptional situation. 606 * @throws IOException if an I/O error occurred. 607 * @throws InterruptedException if the calling thread was interrupted. 608 * @see #login 609 */ 610 public synchronized void login(CharSequence username, String password) throws XMPPException, SmackException, 611 IOException, InterruptedException { 612 login(username, password, config.getResource()); 613 } 614 615 /** 616 * Login with the given username (authorization identity). You may omit the password if a callback handler is used. 617 * If resource is null, then the server will generate one. 618 * 619 * @param username TODO javadoc me please 620 * @param password TODO javadoc me please 621 * @param resource TODO javadoc me please 622 * @throws XMPPException if an XMPP protocol error was received. 623 * @throws SmackException if Smack detected an exceptional situation. 624 * @throws IOException if an I/O error occurred. 625 * @throws InterruptedException if the calling thread was interrupted. 626 * @see #login 627 */ 628 public synchronized void login(CharSequence username, String password, Resourcepart resource) throws XMPPException, 629 SmackException, IOException, InterruptedException { 630 if (!config.allowNullOrEmptyUsername) { 631 StringUtils.requireNotNullNorEmpty(username, "Username must not be null nor empty"); 632 } 633 throwNotConnectedExceptionIfAppropriate("Did you call connect() before login()?"); 634 throwAlreadyLoggedInExceptionIfAppropriate(); 635 usedUsername = username != null ? username.toString() : null; 636 usedPassword = password; 637 usedResource = resource; 638 loginInternal(usedUsername, usedPassword, usedResource); 639 } 640 641 protected abstract void loginInternal(String username, String password, Resourcepart resource) 642 throws XMPPException, SmackException, IOException, InterruptedException; 643 644 @Override 645 public final boolean isConnected() { 646 return connected; 647 } 648 649 @Override 650 public final boolean isAuthenticated() { 651 return authenticated; 652 } 653 654 @Override 655 public final EntityFullJid getUser() { 656 return user; 657 } 658 659 @Override 660 public String getStreamId() { 661 if (!isConnected()) { 662 return null; 663 } 664 return streamId; 665 } 666 667 protected final void throwCurrentConnectionException() throws SmackException, XMPPException { 668 if (currentSmackException != null) { 669 throw currentSmackException; 670 } else if (currentXmppException != null) { 671 throw currentXmppException; 672 } 673 674 throw new AssertionError("No current connection exception set, although throwCurrentException() was called"); 675 } 676 677 protected final boolean hasCurrentConnectionException() { 678 return currentSmackException != null || currentXmppException != null; 679 } 680 681 protected final void setCurrentConnectionExceptionAndNotify(Exception exception) { 682 if (exception instanceof SmackException) { 683 currentSmackException = (SmackException) exception; 684 } else if (exception instanceof XMPPException) { 685 currentXmppException = (XMPPException) exception; 686 } else { 687 currentSmackException = new SmackException.SmackWrappedException(exception); 688 } 689 690 notifyWaitingThreads(); 691 } 692 693 /** 694 * We use an extra object for {@link #notifyWaitingThreads()} and {@link #waitFor(Supplier)}, because all state 695 * changing methods of the connection are synchronized using the connection instance as monitor. If we now would 696 * also use the connection instance for the internal process to wait for a condition, the {@link Object#wait()} 697 * would leave the monitor when it waits, which would allow for another potential call to a state changing function 698 * to proceed. 699 */ 700 private final Object internalMonitor = new Object(); 701 702 protected final void notifyWaitingThreads() { 703 synchronized (internalMonitor) { 704 internalMonitor.notifyAll(); 705 } 706 } 707 708 protected final boolean waitFor(Supplier<Boolean> condition) throws InterruptedException { 709 final long deadline = System.currentTimeMillis() + getReplyTimeout(); 710 synchronized (internalMonitor) { 711 while (!condition.get().booleanValue()) { 712 final long now = System.currentTimeMillis(); 713 if (now >= deadline) { 714 return false; 715 } 716 internalMonitor.wait(deadline - now); 717 } 718 } 719 return true; 720 } 721 722 protected final void waitForConditionOrThrowConnectionException(Supplier<Boolean> condition, String waitFor) throws InterruptedException, SmackException, XMPPException { 723 boolean success = waitFor(() -> condition.get().booleanValue() || hasCurrentConnectionException()); 724 if (hasCurrentConnectionException()) { 725 throwCurrentConnectionException(); 726 } 727 728 // If there was no connection exception and we still did not successfully wait for the condition to hold, then 729 // we ran into a no-response timeout. 730 if (!success) { 731 throw NoResponseException.newWith(this, waitFor); 732 } 733 // Otherwise we successfully awaited the condition. 734 } 735 736 protected Resourcepart bindResourceAndEstablishSession(Resourcepart resource) 737 throws SmackException, InterruptedException, XMPPException { 738 // Wait until either: 739 // - the servers last features stanza has been parsed 740 // - the timeout occurs 741 LOGGER.finer("Waiting for last features to be received before continuing with resource binding"); 742 waitForConditionOrThrowConnectionException(() -> lastFeaturesReceived, "last stream features received from server"); 743 744 if (!hasFeature(Bind.ELEMENT, Bind.NAMESPACE)) { 745 // Server never offered resource binding, which is REQUIRED in XMPP client and 746 // server implementations as per RFC6120 7.2 747 throw new ResourceBindingNotOfferedException(); 748 } 749 750 // Resource binding, see RFC6120 7. 751 // Note that we can not use IQReplyFilter here, since the users full JID is not yet 752 // available. It will become available right after the resource has been successfully bound. 753 Bind bindResource = Bind.newSet(resource); 754 StanzaCollector packetCollector = createStanzaCollectorAndSend(new StanzaIdFilter(bindResource), bindResource); 755 Bind response = packetCollector.nextResultOrThrow(); 756 // Set the connections user to the result of resource binding. It is important that we don't infer the user 757 // from the login() arguments and the configurations service name, as, for example, when SASL External is used, 758 // the username is not given to login but taken from the 'external' certificate. 759 user = response.getJid(); 760 xmppServiceDomain = user.asDomainBareJid(); 761 762 Session.Feature sessionFeature = getFeature(Session.Feature.class); 763 // Only bind the session if it's announced as stream feature by the server, is not optional and not disabled 764 // For more information see http://tools.ietf.org/html/draft-cridland-xmpp-session-01 765 if (sessionFeature != null && !sessionFeature.isOptional()) { 766 Session session = new Session(); 767 packetCollector = createStanzaCollectorAndSend(new StanzaIdFilter(session), session); 768 packetCollector.nextResultOrThrow(); 769 } 770 771 return response.getJid().getResourcepart(); 772 } 773 774 protected void afterSuccessfulLogin(final boolean resumed) throws NotConnectedException, InterruptedException { 775 if (!resumed) { 776 authenticatedConnectionInitiallyEstablishedTimestamp = System.currentTimeMillis(); 777 } 778 // Indicate that we're now authenticated. 779 this.authenticated = true; 780 781 // If debugging is enabled, change the the debug window title to include the 782 // name we are now logged-in as. 783 // If DEBUG was set to true AFTER the connection was created the debugger 784 // will be null 785 if (debugger != null) { 786 debugger.userHasLogged(user); 787 } 788 callConnectionAuthenticatedListener(resumed); 789 790 // Set presence to online. It is important that this is done after 791 // callConnectionAuthenticatedListener(), as this call will also 792 // eventually load the roster. And we should load the roster before we 793 // send the initial presence. 794 if (config.isSendPresence() && !resumed) { 795 Presence availablePresence = getStanzaFactory() 796 .buildPresenceStanza() 797 .ofType(Presence.Type.available) 798 .build(); 799 sendStanza(availablePresence); 800 } 801 } 802 803 @Override 804 public final boolean isAnonymous() { 805 return isAuthenticated() && SASLAnonymous.NAME.equals(getUsedSaslMechansism()); 806 } 807 808 /** 809 * Get the name of the SASL mechanism that was used to authenticate this connection. This returns the name of 810 * mechanism which was used the last time this connection was authenticated, and will return <code>null</code> if 811 * this connection was not authenticated before. 812 * 813 * @return the name of the used SASL mechanism. 814 * @since 4.2 815 */ 816 public final String getUsedSaslMechansism() { 817 return saslAuthentication.getNameOfLastUsedSaslMechansism(); 818 } 819 820 private DomainBareJid xmppServiceDomain; 821 822 protected Lock getConnectionLock() { 823 return connectionLock; 824 } 825 826 protected void throwNotConnectedExceptionIfAppropriate() throws NotConnectedException { 827 throwNotConnectedExceptionIfAppropriate(null); 828 } 829 830 protected void throwNotConnectedExceptionIfAppropriate(String optionalHint) throws NotConnectedException { 831 if (!isConnected()) { 832 throw new NotConnectedException(optionalHint); 833 } 834 } 835 836 protected void throwAlreadyConnectedExceptionIfAppropriate() throws AlreadyConnectedException { 837 if (isConnected()) { 838 throw new AlreadyConnectedException(); 839 } 840 } 841 842 protected void throwAlreadyLoggedInExceptionIfAppropriate() throws AlreadyLoggedInException { 843 if (isAuthenticated()) { 844 throw new AlreadyLoggedInException(); 845 } 846 } 847 848 @Override 849 public final StanzaFactory getStanzaFactory() { 850 return stanzaFactory; 851 } 852 853 @Override 854 public final void sendStanza(Stanza stanza) throws NotConnectedException, InterruptedException { 855 Objects.requireNonNull(stanza, "Stanza must not be null"); 856 assert stanza instanceof Message || stanza instanceof Presence || stanza instanceof IQ; 857 858 throwNotConnectedExceptionIfAppropriate(); 859 switch (fromMode) { 860 case OMITTED: 861 stanza.setFrom((Jid) null); 862 break; 863 case USER: 864 stanza.setFrom(getUser()); 865 break; 866 case UNCHANGED: 867 default: 868 break; 869 } 870 // Invoke interceptors for the new stanza that is about to be sent. Interceptors may modify 871 // the content of the stanza. 872 Stanza stanzaAfterInterceptors = firePacketInterceptors(stanza); 873 sendStanzaInternal(stanzaAfterInterceptors); 874 } 875 876 /** 877 * Authenticate a connection. 878 * 879 * @param username the username that is authenticating with the server. 880 * @param password the password to send to the server. 881 * @param authzid the authorization identifier (typically null). 882 * @param sslSession the optional SSL/TLS session (if one was established) 883 * @return the used SASLMechanism. 884 * @throws XMPPErrorException if there was an XMPP error returned. 885 * @throws SASLErrorException if a SASL protocol error was returned. 886 * @throws IOException if an I/O error occurred. 887 * @throws InterruptedException if the calling thread was interrupted. 888 * @throws SmackSaslException if a SASL specific error occurred. 889 * @throws NotConnectedException if the XMPP connection is not connected. 890 * @throws NoResponseException if there was no response from the remote entity. 891 * @throws SmackWrappedException in case of an exception. 892 * @see SASLAuthentication#authenticate(String, String, EntityBareJid, SSLSession) 893 */ 894 protected final SASLMechanism authenticate(String username, String password, EntityBareJid authzid, 895 SSLSession sslSession) throws XMPPErrorException, SASLErrorException, SmackSaslException, 896 NotConnectedException, NoResponseException, IOException, InterruptedException, SmackWrappedException { 897 SASLMechanism saslMechanism = saslAuthentication.authenticate(username, password, authzid, sslSession); 898 afterSaslAuthenticationSuccess(); 899 return saslMechanism; 900 } 901 902 /** 903 * Hook for subclasses right after successful SASL authentication. RFC 6120 § 6.4.6. specifies a that the initiating 904 * entity, needs to initiate a new stream in this case. But some transports, like BOSH, requires a special handling. 905 * <p> 906 * Note that we can not reset XMPPTCPConnection's parser here, because this method is invoked by the thread calling 907 * {@link #login()}, but the parser reset has to be done within the reader thread. 908 * </p> 909 * 910 * @throws NotConnectedException if the XMPP connection is not connected. 911 * @throws InterruptedException if the calling thread was interrupted. 912 * @throws SmackWrappedException in case of an exception. 913 */ 914 protected void afterSaslAuthenticationSuccess() 915 throws NotConnectedException, InterruptedException, SmackWrappedException { 916 sendStreamOpen(); 917 } 918 919 protected final boolean isSaslAuthenticated() { 920 return saslAuthentication.authenticationSuccessful(); 921 } 922 923 /** 924 * Closes the connection by setting presence to unavailable then closing the connection to 925 * the XMPP server. The XMPPConnection can still be used for connecting to the server 926 * again. 927 * 928 */ 929 public void disconnect() { 930 Presence unavailablePresence = null; 931 if (isAuthenticated()) { 932 unavailablePresence = getStanzaFactory().buildPresenceStanza() 933 .ofType(Presence.Type.unavailable) 934 .build(); 935 } 936 try { 937 disconnect(unavailablePresence); 938 } 939 catch (NotConnectedException e) { 940 LOGGER.log(Level.FINEST, "Connection is already disconnected", e); 941 } 942 } 943 944 /** 945 * Closes the connection. A custom unavailable presence is sent to the server, followed 946 * by closing the stream. The XMPPConnection can still be used for connecting to the server 947 * again. A custom unavailable presence is useful for communicating offline presence 948 * information such as "On vacation". Typically, just the status text of the presence 949 * stanza is set with online information, but most XMPP servers will deliver the full 950 * presence stanza with whatever data is set. 951 * 952 * @param unavailablePresence the optional presence stanza to send during shutdown. 953 * @throws NotConnectedException if the XMPP connection is not connected. 954 */ 955 public synchronized void disconnect(Presence unavailablePresence) throws NotConnectedException { 956 if (unavailablePresence != null) { 957 try { 958 sendStanza(unavailablePresence); 959 } catch (InterruptedException e) { 960 LOGGER.log(Level.FINE, 961 "Was interrupted while sending unavailable presence. Continuing to disconnect the connection", 962 e); 963 } 964 } 965 shutdown(); 966 callConnectionClosedListener(); 967 } 968 969 private final Object notifyConnectionErrorMonitor = new Object(); 970 971 /** 972 * Sends out a notification that there was an error with the connection 973 * and closes the connection. 974 * 975 * @param exception the exception that causes the connection close event. 976 */ 977 protected final void notifyConnectionError(final Exception exception) { 978 synchronized (notifyConnectionErrorMonitor) { 979 if (!isConnected()) { 980 LOGGER.log(Level.INFO, "Connection was already disconnected when attempting to handle " + exception, 981 exception); 982 return; 983 } 984 985 // Note that we first have to set the current connection exception and notify waiting threads, as one of them 986 // could hold the instance lock, which we also need later when calling instantShutdown(). 987 setCurrentConnectionExceptionAndNotify(exception); 988 989 // Closes the connection temporary. A if the connection supports stream management, then a reconnection is 990 // possible. Note that a connection listener of e.g. XMPPTCPConnection will drop the SM state in 991 // case the Exception is a StreamErrorException. 992 instantShutdown(); 993 994 for (StanzaCollector collector : collectors) { 995 collector.notifyConnectionError(exception); 996 } 997 998 Async.go(() -> { 999 // Notify connection listeners of the error. 1000 callConnectionClosedOnErrorListener(exception); 1001 }, AbstractXMPPConnection.this + " callConnectionClosedOnErrorListener()"); 1002 } 1003 } 1004 1005 /** 1006 * Performs an unclean disconnect and shutdown of the connection. Does not send a closing stream stanza. 1007 */ 1008 public abstract void instantShutdown(); 1009 1010 /** 1011 * Shuts the current connection down. 1012 */ 1013 protected abstract void shutdown(); 1014 1015 protected final boolean waitForClosingStreamTagFromServer() { 1016 try { 1017 waitForConditionOrThrowConnectionException(() -> closingStreamReceived, "closing stream tag from the server"); 1018 } catch (InterruptedException | SmackException | XMPPException e) { 1019 LOGGER.log(Level.INFO, "Exception while waiting for closing stream element from the server " + this, e); 1020 return false; 1021 } 1022 return true; 1023 } 1024 1025 @Override 1026 public void addConnectionListener(ConnectionListener connectionListener) { 1027 if (connectionListener == null) { 1028 return; 1029 } 1030 connectionListeners.add(connectionListener); 1031 } 1032 1033 @Override 1034 public void removeConnectionListener(ConnectionListener connectionListener) { 1035 connectionListeners.remove(connectionListener); 1036 } 1037 1038 @Override 1039 public <I extends IQ> I sendIqRequestAndWaitForResponse(IQ request) 1040 throws NoResponseException, XMPPErrorException, NotConnectedException, InterruptedException { 1041 StanzaCollector collector = createStanzaCollectorAndSend(request); 1042 IQ resultResponse = collector.nextResultOrThrow(); 1043 @SuppressWarnings("unchecked") 1044 I concreteResultResponse = (I) resultResponse; 1045 return concreteResultResponse; 1046 } 1047 1048 @Override 1049 public StanzaCollector createStanzaCollectorAndSend(IQ packet) throws NotConnectedException, InterruptedException { 1050 StanzaFilter packetFilter = new IQReplyFilter(packet, this); 1051 // Create the packet collector before sending the packet 1052 StanzaCollector packetCollector = createStanzaCollectorAndSend(packetFilter, packet); 1053 return packetCollector; 1054 } 1055 1056 @Override 1057 public StanzaCollector createStanzaCollectorAndSend(StanzaFilter packetFilter, Stanza packet) 1058 throws NotConnectedException, InterruptedException { 1059 StanzaCollector.Configuration configuration = StanzaCollector.newConfiguration() 1060 .setStanzaFilter(packetFilter) 1061 .setRequest(packet); 1062 // Create the packet collector before sending the packet 1063 StanzaCollector packetCollector = createStanzaCollector(configuration); 1064 try { 1065 // Now we can send the packet as the collector has been created 1066 sendStanza(packet); 1067 } 1068 catch (InterruptedException | NotConnectedException | RuntimeException e) { 1069 packetCollector.cancel(); 1070 throw e; 1071 } 1072 return packetCollector; 1073 } 1074 1075 @Override 1076 public StanzaCollector createStanzaCollector(StanzaFilter packetFilter) { 1077 StanzaCollector.Configuration configuration = StanzaCollector.newConfiguration().setStanzaFilter(packetFilter); 1078 return createStanzaCollector(configuration); 1079 } 1080 1081 @Override 1082 public StanzaCollector createStanzaCollector(StanzaCollector.Configuration configuration) { 1083 StanzaCollector collector = new StanzaCollector(this, configuration); 1084 // Add the collector to the list of active collectors. 1085 collectors.add(collector); 1086 return collector; 1087 } 1088 1089 @Override 1090 public void removeStanzaCollector(StanzaCollector collector) { 1091 collectors.remove(collector); 1092 } 1093 1094 @Override 1095 public final void addStanzaListener(StanzaListener stanzaListener, StanzaFilter stanzaFilter) { 1096 if (stanzaListener == null) { 1097 throw new NullPointerException("Given stanza listener must not be null"); 1098 } 1099 ListenerWrapper wrapper = new ListenerWrapper(stanzaListener, stanzaFilter); 1100 synchronized (recvListeners) { 1101 recvListeners.put(stanzaListener, wrapper); 1102 } 1103 } 1104 1105 @Override 1106 public final boolean removeStanzaListener(StanzaListener stanzaListener) { 1107 synchronized (recvListeners) { 1108 return recvListeners.remove(stanzaListener) != null; 1109 } 1110 } 1111 1112 @Override 1113 public void addSyncStanzaListener(StanzaListener packetListener, StanzaFilter packetFilter) { 1114 if (packetListener == null) { 1115 throw new NullPointerException("Packet listener is null."); 1116 } 1117 ListenerWrapper wrapper = new ListenerWrapper(packetListener, packetFilter); 1118 synchronized (syncRecvListeners) { 1119 syncRecvListeners.put(packetListener, wrapper); 1120 } 1121 } 1122 1123 @Override 1124 public boolean removeSyncStanzaListener(StanzaListener packetListener) { 1125 synchronized (syncRecvListeners) { 1126 return syncRecvListeners.remove(packetListener) != null; 1127 } 1128 } 1129 1130 @Override 1131 public void addAsyncStanzaListener(StanzaListener packetListener, StanzaFilter packetFilter) { 1132 if (packetListener == null) { 1133 throw new NullPointerException("Packet listener is null."); 1134 } 1135 ListenerWrapper wrapper = new ListenerWrapper(packetListener, packetFilter); 1136 synchronized (asyncRecvListeners) { 1137 asyncRecvListeners.put(packetListener, wrapper); 1138 } 1139 } 1140 1141 @Override 1142 public boolean removeAsyncStanzaListener(StanzaListener packetListener) { 1143 synchronized (asyncRecvListeners) { 1144 return asyncRecvListeners.remove(packetListener) != null; 1145 } 1146 } 1147 1148 @Override 1149 public void addStanzaSendingListener(StanzaListener packetListener, StanzaFilter packetFilter) { 1150 if (packetListener == null) { 1151 throw new NullPointerException("Packet listener is null."); 1152 } 1153 ListenerWrapper wrapper = new ListenerWrapper(packetListener, packetFilter); 1154 synchronized (sendListeners) { 1155 sendListeners.put(packetListener, wrapper); 1156 } 1157 } 1158 1159 @Override 1160 public void removeStanzaSendingListener(StanzaListener packetListener) { 1161 synchronized (sendListeners) { 1162 sendListeners.remove(packetListener); 1163 } 1164 } 1165 1166 /** 1167 * Process all stanza listeners for sending stanzas. 1168 * <p> 1169 * Compared to {@link #firePacketInterceptors(Stanza)}, the listeners will be invoked in a new thread. 1170 * </p> 1171 * 1172 * @param sendTopLevelStreamElement the top level stream element which just got send. 1173 */ 1174 // TODO: Rename to fireElementSendingListeners(). 1175 @SuppressWarnings("javadoc") 1176 protected void firePacketSendingListeners(final TopLevelStreamElement sendTopLevelStreamElement) { 1177 if (debugger != null) { 1178 debugger.onOutgoingStreamElement(sendTopLevelStreamElement); 1179 } 1180 1181 if (!(sendTopLevelStreamElement instanceof Stanza)) { 1182 return; 1183 } 1184 Stanza packet = (Stanza) sendTopLevelStreamElement; 1185 1186 final List<StanzaListener> listenersToNotify = new LinkedList<>(); 1187 synchronized (sendListeners) { 1188 for (ListenerWrapper listenerWrapper : sendListeners.values()) { 1189 if (listenerWrapper.filterMatches(packet)) { 1190 listenersToNotify.add(listenerWrapper.getListener()); 1191 } 1192 } 1193 } 1194 if (listenersToNotify.isEmpty()) { 1195 return; 1196 } 1197 // Notify in a new thread, because we can 1198 asyncGo(new Runnable() { 1199 @Override 1200 public void run() { 1201 for (StanzaListener listener : listenersToNotify) { 1202 try { 1203 listener.processStanza(packet); 1204 } 1205 catch (Exception e) { 1206 LOGGER.log(Level.WARNING, "Sending listener threw exception", e); 1207 continue; 1208 } 1209 } 1210 } 1211 }); 1212 } 1213 1214 @Deprecated 1215 @Override 1216 public void addStanzaInterceptor(StanzaListener packetInterceptor, 1217 StanzaFilter packetFilter) { 1218 if (packetInterceptor == null) { 1219 throw new NullPointerException("Packet interceptor is null."); 1220 } 1221 InterceptorWrapper interceptorWrapper = new InterceptorWrapper(packetInterceptor, packetFilter); 1222 synchronized (interceptors) { 1223 interceptors.put(packetInterceptor, interceptorWrapper); 1224 } 1225 } 1226 1227 @Deprecated 1228 @Override 1229 public void removeStanzaInterceptor(StanzaListener packetInterceptor) { 1230 synchronized (interceptors) { 1231 interceptors.remove(packetInterceptor); 1232 } 1233 } 1234 1235 private static <MPB extends MessageOrPresenceBuilder<MP, MPB>, MP extends MessageOrPresence<MPB>> void addInterceptor( 1236 Map<Consumer<MPB>, GenericInterceptorWrapper<MPB, MP>> interceptors, Consumer<MPB> interceptor, 1237 Predicate<MP> filter) { 1238 Objects.requireNonNull(interceptor, "Interceptor must not be null"); 1239 1240 GenericInterceptorWrapper<MPB, MP> interceptorWrapper = new GenericInterceptorWrapper<>(interceptor, filter); 1241 1242 synchronized (interceptors) { 1243 interceptors.put(interceptor, interceptorWrapper); 1244 } 1245 } 1246 1247 private static <MPB extends MessageOrPresenceBuilder<MP, MPB>, MP extends MessageOrPresence<MPB>> void removeInterceptor( 1248 Map<Consumer<MPB>, GenericInterceptorWrapper<MPB, MP>> interceptors, Consumer<MPB> interceptor) { 1249 synchronized (interceptors) { 1250 interceptors.remove(interceptor); 1251 } 1252 } 1253 1254 @Override 1255 public void addMessageInterceptor(Consumer<MessageBuilder> messageInterceptor, Predicate<Message> messageFilter) { 1256 addInterceptor(messageInterceptors, messageInterceptor, messageFilter); 1257 } 1258 1259 @Override 1260 public void removeMessageInterceptor(Consumer<MessageBuilder> messageInterceptor) { 1261 removeInterceptor(messageInterceptors, messageInterceptor); 1262 } 1263 1264 @Override 1265 public void addPresenceInterceptor(Consumer<PresenceBuilder> presenceInterceptor, 1266 Predicate<Presence> presenceFilter) { 1267 addInterceptor(presenceInterceptors, presenceInterceptor, presenceFilter); 1268 } 1269 1270 @Override 1271 public void removePresenceInterceptor(Consumer<PresenceBuilder> presenceInterceptor) { 1272 removeInterceptor(presenceInterceptors, presenceInterceptor); 1273 } 1274 1275 private static <MPB extends MessageOrPresenceBuilder<MP, MPB>, MP extends MessageOrPresence<MPB>> MP fireMessageOrPresenceInterceptors( 1276 MP messageOrPresence, Map<Consumer<MPB>, GenericInterceptorWrapper<MPB, MP>> interceptors) { 1277 List<Consumer<MPB>> interceptorsToInvoke = new LinkedList<>(); 1278 synchronized (interceptors) { 1279 for (GenericInterceptorWrapper<MPB, MP> interceptorWrapper : interceptors.values()) { 1280 if (interceptorWrapper.filterMatches(messageOrPresence)) { 1281 Consumer<MPB> interceptor = interceptorWrapper.getInterceptor(); 1282 interceptorsToInvoke.add(interceptor); 1283 } 1284 } 1285 } 1286 1287 // Avoid transforming the stanza to a builder if there is no interceptor. 1288 if (interceptorsToInvoke.isEmpty()) { 1289 return messageOrPresence; 1290 } 1291 1292 MPB builder = messageOrPresence.asBuilder(); 1293 for (Consumer<MPB> interceptor : interceptorsToInvoke) { 1294 interceptor.accept(builder); 1295 } 1296 1297 // Now that the interceptors have (probably) modified the stanza in its builder form, we need to re-assemble it. 1298 messageOrPresence = builder.build(); 1299 return messageOrPresence; 1300 } 1301 1302 /** 1303 * Process interceptors. Interceptors may modify the stanza that is about to be sent. 1304 * Since the thread that requested to send the stanza will invoke all interceptors, it 1305 * is important that interceptors perform their work as soon as possible so that the 1306 * thread does not remain blocked for a long period. 1307 * 1308 * @param packet the stanza that is going to be sent to the server. 1309 * @return the, potentially modified stanza, after the interceptors are run. 1310 */ 1311 private Stanza firePacketInterceptors(Stanza packet) { 1312 List<StanzaListener> interceptorsToInvoke = new LinkedList<>(); 1313 synchronized (interceptors) { 1314 for (InterceptorWrapper interceptorWrapper : interceptors.values()) { 1315 if (interceptorWrapper.filterMatches(packet)) { 1316 interceptorsToInvoke.add(interceptorWrapper.getInterceptor()); 1317 } 1318 } 1319 } 1320 for (StanzaListener interceptor : interceptorsToInvoke) { 1321 try { 1322 interceptor.processStanza(packet); 1323 } catch (Exception e) { 1324 LOGGER.log(Level.SEVERE, "Packet interceptor threw exception", e); 1325 } 1326 } 1327 1328 final Stanza stanzaAfterInterceptors; 1329 if (packet instanceof Message) { 1330 Message message = (Message) packet; 1331 stanzaAfterInterceptors = fireMessageOrPresenceInterceptors(message, messageInterceptors); 1332 } 1333 else if (packet instanceof Presence) { 1334 Presence presence = (Presence) packet; 1335 stanzaAfterInterceptors = fireMessageOrPresenceInterceptors(presence, presenceInterceptors); 1336 } else { 1337 // We do not (yet) support interceptors for IQ stanzas. 1338 assert packet instanceof IQ; 1339 stanzaAfterInterceptors = packet; 1340 } 1341 1342 return stanzaAfterInterceptors; 1343 } 1344 1345 /** 1346 * Initialize the {@link #debugger}. You can specify a customized {@link SmackDebugger} 1347 * by setup the system property <code>smack.debuggerClass</code> to the implementation. 1348 * 1349 * @throws IllegalStateException if the reader or writer isn't yet initialized. 1350 * @throws IllegalArgumentException if the SmackDebugger can't be loaded. 1351 */ 1352 protected void initDebugger() { 1353 if (reader == null || writer == null) { 1354 throw new NullPointerException("Reader or writer isn't initialized."); 1355 } 1356 // If debugging is enabled, we open a window and write out all network traffic. 1357 if (debugger != null) { 1358 // Obtain new reader and writer from the existing debugger 1359 reader = debugger.newConnectionReader(reader); 1360 writer = debugger.newConnectionWriter(writer); 1361 } 1362 } 1363 1364 @Override 1365 public long getReplyTimeout() { 1366 return replyTimeout; 1367 } 1368 1369 @Override 1370 public void setReplyTimeout(long timeout) { 1371 if (Long.MAX_VALUE - System.currentTimeMillis() < timeout) { 1372 throw new IllegalArgumentException("Extremely long reply timeout"); 1373 } 1374 else { 1375 replyTimeout = timeout; 1376 } 1377 } 1378 1379 private SmackConfiguration.UnknownIqRequestReplyMode unknownIqRequestReplyMode = SmackConfiguration.getUnknownIqRequestReplyMode(); 1380 1381 /** 1382 * Set how Smack behaves when an unknown IQ request has been received. 1383 * 1384 * @param unknownIqRequestReplyMode reply mode. 1385 */ 1386 public void setUnknownIqRequestReplyMode(UnknownIqRequestReplyMode unknownIqRequestReplyMode) { 1387 this.unknownIqRequestReplyMode = Objects.requireNonNull(unknownIqRequestReplyMode, "Mode must not be null"); 1388 } 1389 1390 protected final NonzaCallback.Builder buildNonzaCallback() { 1391 return new NonzaCallback.Builder(this); 1392 } 1393 1394 protected <SN extends Nonza, FN extends Nonza> SN sendAndWaitForResponse(Nonza nonza, Class<SN> successNonzaClass, 1395 Class<FN> failedNonzaClass) 1396 throws NoResponseException, NotConnectedException, InterruptedException, FailedNonzaException { 1397 NonzaCallback.Builder builder = buildNonzaCallback(); 1398 SN successNonza = NonzaCallback.sendAndWaitForResponse(builder, nonza, successNonzaClass, failedNonzaClass); 1399 return successNonza; 1400 } 1401 1402 private void maybeNotifyDebuggerAboutIncoming(TopLevelStreamElement incomingTopLevelStreamElement) { 1403 final SmackDebugger debugger = this.debugger; 1404 if (debugger != null) { 1405 debugger.onIncomingStreamElement(incomingTopLevelStreamElement); 1406 } 1407 } 1408 1409 protected final void parseAndProcessNonza(XmlPullParser parser) throws IOException, XmlPullParserException, SmackParsingException { 1410 ParserUtils.assertAtStartTag(parser); 1411 1412 final int initialDepth = parser.getDepth(); 1413 final String element = parser.getName(); 1414 final String namespace = parser.getNamespace(); 1415 final QName key = new QName(namespace, element); 1416 1417 NonzaProvider<? extends Nonza> nonzaProvider = ProviderManager.getNonzaProvider(key); 1418 if (nonzaProvider == null) { 1419 LOGGER.severe("Unknown nonza: " + key); 1420 ParserUtils.forwardToEndTagOfDepth(parser, initialDepth); 1421 return; 1422 } 1423 1424 List<NonzaCallback> nonzaCallbacks; 1425 synchronized (nonzaCallbacksMap) { 1426 nonzaCallbacks = nonzaCallbacksMap.getAll(key); 1427 nonzaCallbacks = CollectionUtil.newListWith(nonzaCallbacks); 1428 } 1429 if (nonzaCallbacks == null) { 1430 LOGGER.info("No nonza callback for " + key); 1431 ParserUtils.forwardToEndTagOfDepth(parser, initialDepth); 1432 return; 1433 } 1434 1435 Nonza nonza = nonzaProvider.parse(parser, incomingStreamXmlEnvironment); 1436 1437 maybeNotifyDebuggerAboutIncoming(nonza); 1438 1439 for (NonzaCallback nonzaCallback : nonzaCallbacks) { 1440 nonzaCallback.onNonzaReceived(nonza); 1441 } 1442 } 1443 1444 protected void parseAndProcessStanza(XmlPullParser parser) 1445 throws XmlPullParserException, IOException, InterruptedException { 1446 ParserUtils.assertAtStartTag(parser); 1447 int parserDepth = parser.getDepth(); 1448 Stanza stanza = null; 1449 try { 1450 stanza = PacketParserUtils.parseStanza(parser, incomingStreamXmlEnvironment); 1451 } 1452 catch (XmlPullParserException | SmackParsingException | IOException | IllegalArgumentException e) { 1453 CharSequence content = PacketParserUtils.parseContentDepth(parser, 1454 parserDepth); 1455 UnparseableStanza message = new UnparseableStanza(content, e); 1456 ParsingExceptionCallback callback = getParsingExceptionCallback(); 1457 if (callback != null) { 1458 callback.handleUnparsableStanza(message); 1459 } 1460 } 1461 ParserUtils.assertAtEndTag(parser); 1462 if (stanza != null) { 1463 processStanza(stanza); 1464 } 1465 } 1466 1467 /** 1468 * Processes a stanza after it's been fully parsed by looping through the installed 1469 * stanza collectors and listeners and letting them examine the stanza to see if 1470 * they are a match with the filter. 1471 * 1472 * @param stanza the stanza to process. 1473 * @throws InterruptedException if the calling thread was interrupted. 1474 */ 1475 protected void processStanza(final Stanza stanza) throws InterruptedException { 1476 assert stanza != null; 1477 1478 maybeNotifyDebuggerAboutIncoming(stanza); 1479 1480 lastStanzaReceived = System.currentTimeMillis(); 1481 // Deliver the incoming packet to listeners. 1482 invokeStanzaCollectorsAndNotifyRecvListeners(stanza); 1483 } 1484 1485 /** 1486 * Invoke {@link StanzaCollector#processStanza(Stanza)} for every 1487 * StanzaCollector with the given packet. Also notify the receive listeners with a matching stanza filter about the packet. 1488 * <p> 1489 * This method will be invoked by the connections incoming processing thread which may be shared across multiple connections and 1490 * thus it is important that no user code, e.g. in form of a callback, is invoked by this method. For the same reason, 1491 * this method must not block for an extended period of time. 1492 * </p> 1493 * 1494 * @param packet the stanza to notify the StanzaCollectors and receive listeners about. 1495 */ 1496 protected void invokeStanzaCollectorsAndNotifyRecvListeners(final Stanza packet) { 1497 if (packet instanceof IQ) { 1498 final IQ iq = (IQ) packet; 1499 if (iq.isRequestIQ()) { 1500 final IQ iqRequest = iq; 1501 final QName key = iqRequest.getChildElementQName(); 1502 IQRequestHandler iqRequestHandler; 1503 final IQ.Type type = iq.getType(); 1504 switch (type) { 1505 case set: 1506 synchronized (setIqRequestHandler) { 1507 iqRequestHandler = setIqRequestHandler.get(key); 1508 } 1509 break; 1510 case get: 1511 synchronized (getIqRequestHandler) { 1512 iqRequestHandler = getIqRequestHandler.get(key); 1513 } 1514 break; 1515 default: 1516 throw new IllegalStateException("Should only encounter IQ type 'get' or 'set'"); 1517 } 1518 if (iqRequestHandler == null) { 1519 StanzaError.Condition replyCondition; 1520 switch (unknownIqRequestReplyMode) { 1521 case doNotReply: 1522 return; 1523 case replyFeatureNotImplemented: 1524 replyCondition = StanzaError.Condition.feature_not_implemented; 1525 break; 1526 case replyServiceUnavailable: 1527 replyCondition = StanzaError.Condition.service_unavailable; 1528 break; 1529 default: 1530 throw new AssertionError(); 1531 } 1532 1533 // If the IQ stanza is of type "get" or "set" with no registered IQ request handler, then answer an 1534 // IQ of type 'error' with condition 'service-unavailable'. 1535 final ErrorIQ errorIQ = IQ.createErrorResponse(iq, StanzaError.getBuilder( 1536 replyCondition).build()); 1537 // Use async sendStanza() here, since if sendStanza() would block, then some connections, e.g. 1538 // XmppNioTcpConnection, would deadlock, as this operation is performed in the same thread that is 1539 asyncGo(() -> { 1540 try { 1541 sendStanza(errorIQ); 1542 } 1543 catch (InterruptedException | NotConnectedException e) { 1544 LOGGER.log(Level.WARNING, "Exception while sending error IQ to unkown IQ request", e); 1545 } 1546 }); 1547 } else { 1548 Executor executorService = null; 1549 switch (iqRequestHandler.getMode()) { 1550 case sync: 1551 executorService = ASYNC_BUT_ORDERED.asExecutorFor(this); 1552 break; 1553 case async: 1554 executorService = this::asyncGoLimited; 1555 break; 1556 } 1557 final IQRequestHandler finalIqRequestHandler = iqRequestHandler; 1558 executorService.execute(new Runnable() { 1559 @Override 1560 public void run() { 1561 IQ response = finalIqRequestHandler.handleIQRequest(iq); 1562 if (response == null) { 1563 // It is not ideal if the IQ request handler does not return an IQ response, because RFC 1564 // 6120 § 8.1.2 does specify that a response is mandatory. But some APIs, mostly the 1565 // file transfer one, does not always return a result, so we need to handle this case. 1566 // Also sometimes a request handler may decide that it's better to not send a response, 1567 // e.g. to avoid presence leaks. 1568 return; 1569 } 1570 1571 assert response.isResponseIQ(); 1572 1573 response.setTo(iqRequest.getFrom()); 1574 response.setStanzaId(iqRequest.getStanzaId()); 1575 try { 1576 sendStanza(response); 1577 } 1578 catch (InterruptedException | NotConnectedException e) { 1579 LOGGER.log(Level.WARNING, "Exception while sending response to IQ request", e); 1580 } 1581 } 1582 }); 1583 } 1584 // The following returns makes it impossible for packet listeners and collectors to 1585 // filter for IQ request stanzas, i.e. IQs of type 'set' or 'get'. This is the 1586 // desired behavior. 1587 return; 1588 } 1589 } 1590 1591 // First handle the async recv listeners. Note that this code is very similar to what follows a few lines below, 1592 // the only difference is that asyncRecvListeners is used here and that the packet listeners are started in 1593 // their own thread. 1594 final Collection<StanzaListener> listenersToNotify = new LinkedList<>(); 1595 extractMatchingListeners(packet, asyncRecvListeners, listenersToNotify); 1596 for (final StanzaListener listener : listenersToNotify) { 1597 asyncGoLimited(new Runnable() { 1598 @Override 1599 public void run() { 1600 try { 1601 listener.processStanza(packet); 1602 } catch (Exception e) { 1603 LOGGER.log(Level.SEVERE, "Exception in async packet listener", e); 1604 } 1605 } 1606 }); 1607 } 1608 1609 // Loop through all collectors and notify the appropriate ones. 1610 for (StanzaCollector collector : collectors) { 1611 collector.processStanza(packet); 1612 } 1613 1614 listenersToNotify.clear(); 1615 extractMatchingListeners(packet, recvListeners, listenersToNotify); 1616 for (StanzaListener stanzaListener : listenersToNotify) { 1617 inOrderListeners.performAsyncButOrdered(stanzaListener, () -> { 1618 try { 1619 stanzaListener.processStanza(packet); 1620 } 1621 catch (NotConnectedException e) { 1622 LOGGER.log(Level.WARNING, "Got not connected exception, aborting", e); 1623 } 1624 catch (Exception e) { 1625 LOGGER.log(Level.SEVERE, "Exception in packet listener", e); 1626 } 1627 }); 1628 } 1629 1630 // Notify the receive listeners interested in the packet 1631 listenersToNotify.clear(); 1632 extractMatchingListeners(packet, syncRecvListeners, listenersToNotify); 1633 // Decouple incoming stanza processing from listener invocation. Unlike async listeners, this uses a single 1634 // threaded executor service and therefore keeps the order. 1635 ASYNC_BUT_ORDERED.performAsyncButOrdered(this, new Runnable() { 1636 @Override 1637 public void run() { 1638 // As listeners are able to remove themselves and because the timepoint where it is decided to invoke a 1639 // listener is a different timepoint where the listener is actually invoked (here), we have to check 1640 // again if the listener is still active. 1641 Iterator<StanzaListener> it = listenersToNotify.iterator(); 1642 synchronized (syncRecvListeners) { 1643 while (it.hasNext()) { 1644 StanzaListener stanzaListener = it.next(); 1645 if (!syncRecvListeners.containsKey(stanzaListener)) { 1646 // The listener was removed from syncRecvListener, also remove him from listenersToNotify. 1647 it.remove(); 1648 } 1649 } 1650 } 1651 for (StanzaListener listener : listenersToNotify) { 1652 try { 1653 listener.processStanza(packet); 1654 } catch (NotConnectedException e) { 1655 LOGGER.log(Level.WARNING, "Got not connected exception, aborting", e); 1656 break; 1657 } catch (Exception e) { 1658 LOGGER.log(Level.SEVERE, "Exception in packet listener", e); 1659 } 1660 } 1661 } 1662 }); 1663 } 1664 1665 private static void extractMatchingListeners(Stanza stanza, Map<StanzaListener, ListenerWrapper> listeners, 1666 Collection<StanzaListener> listenersToNotify) { 1667 synchronized (listeners) { 1668 for (ListenerWrapper listenerWrapper : listeners.values()) { 1669 if (listenerWrapper.filterMatches(stanza)) { 1670 listenersToNotify.add(listenerWrapper.getListener()); 1671 } 1672 } 1673 } 1674 } 1675 1676 /** 1677 * Sets whether the connection has already logged in the server. This method assures that the 1678 * {@link #wasAuthenticated} flag is never reset once it has ever been set. 1679 * 1680 */ 1681 protected void setWasAuthenticated() { 1682 // Never reset the flag if the connection has ever been authenticated 1683 if (!wasAuthenticated) { 1684 wasAuthenticated = authenticated; 1685 } 1686 } 1687 1688 protected void callConnectionConnectingListener() { 1689 for (ConnectionListener listener : connectionListeners) { 1690 listener.connecting(this); 1691 } 1692 } 1693 1694 protected void callConnectionConnectedListener() { 1695 for (ConnectionListener listener : connectionListeners) { 1696 listener.connected(this); 1697 } 1698 } 1699 1700 protected void callConnectionAuthenticatedListener(boolean resumed) { 1701 for (ConnectionListener listener : connectionListeners) { 1702 try { 1703 listener.authenticated(this, resumed); 1704 } catch (Exception e) { 1705 // Catch and print any exception so we can recover 1706 // from a faulty listener and finish the shutdown process 1707 LOGGER.log(Level.SEVERE, "Exception in authenticated listener", e); 1708 } 1709 } 1710 } 1711 1712 void callConnectionClosedListener() { 1713 for (ConnectionListener listener : connectionListeners) { 1714 try { 1715 listener.connectionClosed(); 1716 } 1717 catch (Exception e) { 1718 // Catch and print any exception so we can recover 1719 // from a faulty listener and finish the shutdown process 1720 LOGGER.log(Level.SEVERE, "Error in listener while closing connection", e); 1721 } 1722 } 1723 } 1724 1725 private void callConnectionClosedOnErrorListener(Exception e) { 1726 boolean logWarning = true; 1727 if (e instanceof StreamErrorException) { 1728 StreamErrorException see = (StreamErrorException) e; 1729 if (see.getStreamError().getCondition() == StreamError.Condition.not_authorized 1730 && wasAuthenticated) { 1731 logWarning = false; 1732 LOGGER.log(Level.FINE, 1733 "Connection closed with not-authorized stream error after it was already authenticated. The account was likely deleted/unregistered on the server"); 1734 } 1735 } 1736 if (logWarning) { 1737 LOGGER.log(Level.WARNING, "Connection " + this + " closed with error", e); 1738 } 1739 for (ConnectionListener listener : connectionListeners) { 1740 try { 1741 listener.connectionClosedOnError(e); 1742 } 1743 catch (Exception e2) { 1744 // Catch and print any exception so we can recover 1745 // from a faulty listener 1746 LOGGER.log(Level.SEVERE, "Error in listener while closing connection", e2); 1747 } 1748 } 1749 } 1750 1751 /** 1752 * A wrapper class to associate a stanza filter with a listener. 1753 */ 1754 protected static class ListenerWrapper { 1755 1756 private final StanzaListener packetListener; 1757 private final StanzaFilter packetFilter; 1758 1759 /** 1760 * Create a class which associates a stanza filter with a listener. 1761 * 1762 * @param packetListener the stanza listener. 1763 * @param packetFilter the associated filter or null if it listen for all packets. 1764 */ 1765 public ListenerWrapper(StanzaListener packetListener, StanzaFilter packetFilter) { 1766 this.packetListener = packetListener; 1767 this.packetFilter = packetFilter; 1768 } 1769 1770 public boolean filterMatches(Stanza packet) { 1771 return packetFilter == null || packetFilter.accept(packet); 1772 } 1773 1774 public StanzaListener getListener() { 1775 return packetListener; 1776 } 1777 } 1778 1779 /** 1780 * A wrapper class to associate a stanza filter with an interceptor. 1781 */ 1782 @Deprecated 1783 // TODO: Remove once addStanzaInterceptor is gone. 1784 protected static class InterceptorWrapper { 1785 1786 private final StanzaListener packetInterceptor; 1787 private final StanzaFilter packetFilter; 1788 1789 /** 1790 * Create a class which associates a stanza filter with an interceptor. 1791 * 1792 * @param packetInterceptor the interceptor. 1793 * @param packetFilter the associated filter or null if it intercepts all packets. 1794 */ 1795 public InterceptorWrapper(StanzaListener packetInterceptor, StanzaFilter packetFilter) { 1796 this.packetInterceptor = packetInterceptor; 1797 this.packetFilter = packetFilter; 1798 } 1799 1800 public boolean filterMatches(Stanza packet) { 1801 return packetFilter == null || packetFilter.accept(packet); 1802 } 1803 1804 public StanzaListener getInterceptor() { 1805 return packetInterceptor; 1806 } 1807 } 1808 1809 private static final class GenericInterceptorWrapper<MPB extends MessageOrPresenceBuilder<MP, MPB>, MP extends MessageOrPresence<MPB>> { 1810 private final Consumer<MPB> stanzaInterceptor; 1811 private final Predicate<MP> stanzaFilter; 1812 1813 private GenericInterceptorWrapper(Consumer<MPB> stanzaInterceptor, Predicate<MP> stanzaFilter) { 1814 this.stanzaInterceptor = stanzaInterceptor; 1815 this.stanzaFilter = stanzaFilter; 1816 } 1817 1818 private boolean filterMatches(MP stanza) { 1819 return stanzaFilter == null || stanzaFilter.test(stanza); 1820 } 1821 1822 public Consumer<MPB> getInterceptor() { 1823 return stanzaInterceptor; 1824 } 1825 } 1826 1827 @Override 1828 public int getConnectionCounter() { 1829 return connectionCounterValue; 1830 } 1831 1832 @Override 1833 public void setFromMode(FromMode fromMode) { 1834 this.fromMode = fromMode; 1835 } 1836 1837 @Override 1838 public FromMode getFromMode() { 1839 return this.fromMode; 1840 } 1841 1842 protected final void parseFeatures(XmlPullParser parser) throws XmlPullParserException, IOException, SmackParsingException { 1843 streamFeatures.clear(); 1844 final int initialDepth = parser.getDepth(); 1845 while (true) { 1846 XmlPullParser.Event eventType = parser.next(); 1847 1848 if (eventType == XmlPullParser.Event.START_ELEMENT && parser.getDepth() == initialDepth + 1) { 1849 FullyQualifiedElement streamFeature = null; 1850 String name = parser.getName(); 1851 String namespace = parser.getNamespace(); 1852 switch (name) { 1853 case StartTls.ELEMENT: 1854 streamFeature = PacketParserUtils.parseStartTlsFeature(parser); 1855 break; 1856 case Mechanisms.ELEMENT: 1857 streamFeature = new Mechanisms(PacketParserUtils.parseMechanisms(parser)); 1858 break; 1859 case Bind.ELEMENT: 1860 streamFeature = Bind.Feature.INSTANCE; 1861 break; 1862 case Session.ELEMENT: 1863 streamFeature = PacketParserUtils.parseSessionFeature(parser); 1864 break; 1865 case Compress.Feature.ELEMENT: 1866 streamFeature = PacketParserUtils.parseCompressionFeature(parser); 1867 break; 1868 default: 1869 ExtensionElementProvider<ExtensionElement> provider = ProviderManager.getStreamFeatureProvider(name, namespace); 1870 if (provider != null) { 1871 streamFeature = provider.parse(parser, incomingStreamXmlEnvironment); 1872 } 1873 break; 1874 } 1875 if (streamFeature != null) { 1876 addStreamFeature(streamFeature); 1877 } 1878 } 1879 else if (eventType == XmlPullParser.Event.END_ELEMENT && parser.getDepth() == initialDepth) { 1880 break; 1881 } 1882 } 1883 } 1884 1885 protected final void parseFeaturesAndNotify(XmlPullParser parser) throws Exception { 1886 parseFeatures(parser); 1887 1888 if (hasFeature(Mechanisms.ELEMENT, Mechanisms.NAMESPACE)) { 1889 // Only proceed with SASL auth if TLS is disabled or if the server doesn't announce it 1890 if (!hasFeature(StartTls.ELEMENT, StartTls.NAMESPACE) 1891 || config.getSecurityMode() == SecurityMode.disabled) { 1892 tlsHandled = saslFeatureReceived = true; 1893 notifyWaitingThreads(); 1894 } 1895 } 1896 1897 // If the server reported the bind feature then we are that that we did SASL and maybe 1898 // STARTTLS. We can then report that the last 'stream:features' have been parsed 1899 if (hasFeature(Bind.ELEMENT, Bind.NAMESPACE)) { 1900 if (!hasFeature(Compress.Feature.ELEMENT, Compress.NAMESPACE) 1901 || !config.isCompressionEnabled()) { 1902 // This where the last stream features from the server, either it did not contain 1903 // compression or we disabled it. 1904 lastFeaturesReceived = true; 1905 notifyWaitingThreads(); 1906 } 1907 } 1908 afterFeaturesReceived(); 1909 } 1910 1911 @SuppressWarnings("unused") 1912 protected void afterFeaturesReceived() throws SecurityRequiredException, NotConnectedException, InterruptedException { 1913 // Default implementation does nothing 1914 } 1915 1916 @SuppressWarnings("unchecked") 1917 @Override 1918 public <F extends FullyQualifiedElement> F getFeature(QName qname) { 1919 return (F) streamFeatures.get(qname); 1920 } 1921 1922 @Override 1923 public boolean hasFeature(QName qname) { 1924 return streamFeatures.containsKey(qname); 1925 } 1926 1927 protected void addStreamFeature(FullyQualifiedElement feature) { 1928 QName key = feature.getQName(); 1929 streamFeatures.put(key, feature); 1930 } 1931 1932 @Override 1933 public SmackFuture<IQ, Exception> sendIqRequestAsync(IQ request) { 1934 return sendIqRequestAsync(request, getReplyTimeout()); 1935 } 1936 1937 @Override 1938 public SmackFuture<IQ, Exception> sendIqRequestAsync(IQ request, long timeout) { 1939 StanzaFilter replyFilter = new IQReplyFilter(request, this); 1940 return sendAsync(request, replyFilter, timeout); 1941 } 1942 1943 @Override 1944 public <S extends Stanza> SmackFuture<S, Exception> sendAsync(S stanza, final StanzaFilter replyFilter) { 1945 return sendAsync(stanza, replyFilter, getReplyTimeout()); 1946 } 1947 1948 @SuppressWarnings("FutureReturnValueIgnored") 1949 @Override 1950 public <S extends Stanza> SmackFuture<S, Exception> sendAsync(S stanza, final StanzaFilter replyFilter, long timeout) { 1951 Objects.requireNonNull(stanza, "stanza must not be null"); 1952 // While Smack allows to add PacketListeners with a PacketFilter value of 'null', we 1953 // disallow it here in the async API as it makes no sense 1954 Objects.requireNonNull(replyFilter, "replyFilter must not be null"); 1955 1956 final InternalSmackFuture<S, Exception> future = new InternalSmackFuture<>(); 1957 1958 final StanzaListener stanzaListener = new StanzaListener() { 1959 @Override 1960 public void processStanza(Stanza stanza) throws NotConnectedException, InterruptedException { 1961 boolean removed = removeAsyncStanzaListener(this); 1962 if (!removed) { 1963 // We lost a race against the "no response" handling runnable. Avoid calling the callback, as the 1964 // exception callback will be invoked (if any). 1965 return; 1966 } 1967 try { 1968 XMPPErrorException.ifHasErrorThenThrow(stanza); 1969 @SuppressWarnings("unchecked") 1970 S s = (S) stanza; 1971 future.setResult(s); 1972 } 1973 catch (XMPPErrorException exception) { 1974 future.setException(exception); 1975 } 1976 } 1977 }; 1978 schedule(new Runnable() { 1979 @Override 1980 public void run() { 1981 boolean removed = removeAsyncStanzaListener(stanzaListener); 1982 if (!removed) { 1983 // We lost a race against the stanza listener, he already removed itself because he received a 1984 // reply. There is nothing more to do here. 1985 return; 1986 } 1987 1988 // If the packetListener got removed, then it was never run and 1989 // we never received a response, inform the exception callback 1990 Exception exception; 1991 if (!isConnected()) { 1992 // If the connection is no longer connected, throw a not connected exception. 1993 exception = new NotConnectedException(AbstractXMPPConnection.this, replyFilter); 1994 } 1995 else { 1996 exception = NoResponseException.newWith(AbstractXMPPConnection.this, replyFilter); 1997 } 1998 future.setException(exception); 1999 } 2000 }, timeout, TimeUnit.MILLISECONDS); 2001 2002 addAsyncStanzaListener(stanzaListener, replyFilter); 2003 Runnable sendOperation = () -> { 2004 try { 2005 sendStanza(stanza); 2006 } 2007 catch (NotConnectedException | InterruptedException exception) { 2008 future.setException(exception); 2009 } 2010 }; 2011 if (SmackConfiguration.TRUELY_ASYNC_SENDS) { 2012 Async.go(sendOperation); 2013 } else { 2014 sendOperation.run(); 2015 } 2016 2017 return future; 2018 } 2019 2020 @SuppressWarnings("FutureReturnValueIgnored") 2021 @Override 2022 public void addOneTimeSyncCallback(final StanzaListener callback, final StanzaFilter packetFilter) { 2023 final StanzaListener packetListener = new StanzaListener() { 2024 @Override 2025 public void processStanza(Stanza packet) throws NotConnectedException, InterruptedException, NotLoggedInException { 2026 try { 2027 callback.processStanza(packet); 2028 } finally { 2029 removeSyncStanzaListener(this); 2030 } 2031 } 2032 }; 2033 addSyncStanzaListener(packetListener, packetFilter); 2034 schedule(new Runnable() { 2035 @Override 2036 public void run() { 2037 removeSyncStanzaListener(packetListener); 2038 } 2039 }, getReplyTimeout(), TimeUnit.MILLISECONDS); 2040 } 2041 2042 @Override 2043 public IQRequestHandler registerIQRequestHandler(final IQRequestHandler iqRequestHandler) { 2044 final QName key = iqRequestHandler.getQName(); 2045 switch (iqRequestHandler.getType()) { 2046 case set: 2047 synchronized (setIqRequestHandler) { 2048 return setIqRequestHandler.put(key, iqRequestHandler); 2049 } 2050 case get: 2051 synchronized (getIqRequestHandler) { 2052 return getIqRequestHandler.put(key, iqRequestHandler); 2053 } 2054 default: 2055 throw new IllegalArgumentException("Only IQ type of 'get' and 'set' allowed"); 2056 } 2057 } 2058 2059 @Override 2060 public final IQRequestHandler unregisterIQRequestHandler(IQRequestHandler iqRequestHandler) { 2061 return unregisterIQRequestHandler(iqRequestHandler.getElement(), iqRequestHandler.getNamespace(), 2062 iqRequestHandler.getType()); 2063 } 2064 2065 @Override 2066 public IQRequestHandler unregisterIQRequestHandler(String element, String namespace, IQ.Type type) { 2067 final QName key = new QName(namespace, element); 2068 switch (type) { 2069 case set: 2070 synchronized (setIqRequestHandler) { 2071 return setIqRequestHandler.remove(key); 2072 } 2073 case get: 2074 synchronized (getIqRequestHandler) { 2075 return getIqRequestHandler.remove(key); 2076 } 2077 default: 2078 throw new IllegalArgumentException("Only IQ type of 'get' and 'set' allowed"); 2079 } 2080 } 2081 2082 private long lastStanzaReceived; 2083 2084 @Override 2085 public long getLastStanzaReceived() { 2086 return lastStanzaReceived; 2087 } 2088 2089 /** 2090 * Get the timestamp when the connection was the first time authenticated, i.e., when the first successful login was 2091 * performed. Note that this value is not reset on disconnect, so it represents the timestamp from the last 2092 * authenticated connection. The value is also not reset on stream resumption. 2093 * 2094 * @return the timestamp or {@code null}. 2095 * @since 4.3.3 2096 */ 2097 public final long getAuthenticatedConnectionInitiallyEstablishedTimestamp() { 2098 return authenticatedConnectionInitiallyEstablishedTimestamp; 2099 } 2100 2101 /** 2102 * Install a parsing exception callback, which will be invoked once an exception is encountered while parsing a 2103 * stanza. 2104 * 2105 * @param callback the callback to install 2106 */ 2107 public void setParsingExceptionCallback(ParsingExceptionCallback callback) { 2108 parsingExceptionCallback = callback; 2109 } 2110 2111 /** 2112 * Get the current active parsing exception callback. 2113 * 2114 * @return the active exception callback or null if there is none 2115 */ 2116 public ParsingExceptionCallback getParsingExceptionCallback() { 2117 return parsingExceptionCallback; 2118 } 2119 2120 @Override 2121 public final String toString() { 2122 EntityFullJid localEndpoint = getUser(); 2123 String localEndpointString = localEndpoint == null ? "not-authenticated" : localEndpoint.toString(); 2124 return getClass().getSimpleName() + '[' + localEndpointString + "] (" + getConnectionCounter() + ')'; 2125 } 2126 2127 /** 2128 * A queue of deferred runnables that where not executed immediately because {@link #currentAsyncRunnables} reached 2129 * {@link #maxAsyncRunnables}. Note that we use a {@code LinkedList} in order to avoid space blowups in case the 2130 * list ever becomes very big and shrinks again. 2131 */ 2132 private final Queue<Runnable> deferredAsyncRunnables = new LinkedList<>(); 2133 2134 private int deferredAsyncRunnablesCount; 2135 2136 private int deferredAsyncRunnablesCountPrevious; 2137 2138 private int maxAsyncRunnables = SmackConfiguration.getDefaultConcurrencyLevelLimit(); 2139 2140 private int currentAsyncRunnables; 2141 2142 protected void asyncGoLimited(final Runnable runnable) { 2143 Runnable wrappedRunnable = new Runnable() { 2144 @Override 2145 public void run() { 2146 runnable.run(); 2147 2148 synchronized (deferredAsyncRunnables) { 2149 Runnable defferredRunnable = deferredAsyncRunnables.poll(); 2150 if (defferredRunnable == null) { 2151 currentAsyncRunnables--; 2152 } else { 2153 deferredAsyncRunnablesCount--; 2154 asyncGo(defferredRunnable); 2155 } 2156 } 2157 } 2158 }; 2159 2160 synchronized (deferredAsyncRunnables) { 2161 if (currentAsyncRunnables < maxAsyncRunnables) { 2162 currentAsyncRunnables++; 2163 asyncGo(wrappedRunnable); 2164 } else { 2165 deferredAsyncRunnablesCount++; 2166 deferredAsyncRunnables.add(wrappedRunnable); 2167 } 2168 2169 final int HIGH_WATERMARK = 100; 2170 final int INFORM_WATERMARK = 20; 2171 2172 final int deferredAsyncRunnablesCount = this.deferredAsyncRunnablesCount; 2173 2174 if (deferredAsyncRunnablesCount >= HIGH_WATERMARK 2175 && deferredAsyncRunnablesCountPrevious < HIGH_WATERMARK) { 2176 LOGGER.log(Level.WARNING, "High watermark of " + HIGH_WATERMARK + " simultaneous executing runnables reached"); 2177 } else if (deferredAsyncRunnablesCount >= INFORM_WATERMARK 2178 && deferredAsyncRunnablesCountPrevious < INFORM_WATERMARK) { 2179 LOGGER.log(Level.INFO, INFORM_WATERMARK + " simultaneous executing runnables reached"); 2180 } 2181 2182 deferredAsyncRunnablesCountPrevious = deferredAsyncRunnablesCount; 2183 } 2184 } 2185 2186 public void setMaxAsyncOperations(int maxAsyncOperations) { 2187 if (maxAsyncOperations < 1) { 2188 throw new IllegalArgumentException("Max async operations must be greater than 0"); 2189 } 2190 2191 synchronized (deferredAsyncRunnables) { 2192 maxAsyncRunnables = maxAsyncOperations; 2193 } 2194 } 2195 2196 protected static void asyncGo(Runnable runnable) { 2197 CACHED_EXECUTOR_SERVICE.execute(runnable); 2198 } 2199 2200 @SuppressWarnings("static-method") 2201 protected final SmackReactor getReactor() { 2202 return SMACK_REACTOR; 2203 } 2204 2205 protected static ScheduledAction schedule(Runnable runnable, long delay, TimeUnit unit) { 2206 return SMACK_REACTOR.schedule(runnable, delay, unit, ScheduledAction.Kind.NonBlocking); 2207 } 2208 2209 protected void onStreamOpen(XmlPullParser parser) { 2210 // We found an opening stream. 2211 if ("jabber:client".equals(parser.getNamespace(null))) { 2212 streamId = parser.getAttributeValue("", "id"); 2213 incomingStreamXmlEnvironment = XmlEnvironment.from(parser); 2214 2215 String reportedServerDomainString = parser.getAttributeValue("", "from"); 2216 if (reportedServerDomainString == null) { 2217 // RFC 6120 § 4.7.1. makes no explicit statement whether or not 'from' in the stream open from the server 2218 // in c2s connections is required or not. 2219 return; 2220 } 2221 DomainBareJid reportedServerDomain; 2222 try { 2223 reportedServerDomain = JidCreate.domainBareFrom(reportedServerDomainString); 2224 DomainBareJid configuredXmppServiceDomain = config.getXMPPServiceDomain(); 2225 if (!configuredXmppServiceDomain.equals(reportedServerDomain)) { 2226 LOGGER.warning("Domain reported by server '" + reportedServerDomain 2227 + "' does not match configured domain '" + configuredXmppServiceDomain + "'"); 2228 } 2229 } catch (XmppStringprepException e) { 2230 LOGGER.log(Level.WARNING, "XMPP service domain '" + reportedServerDomainString 2231 + "' as reported by server could not be transformed to a valid JID", e); 2232 } 2233 } 2234 } 2235 2236 protected void sendStreamOpen() throws NotConnectedException, InterruptedException { 2237 // If possible, provide the receiving entity of the stream open tag, i.e. the server, as much information as 2238 // possible. The 'to' attribute is *always* available. The 'from' attribute if set by the user and no external 2239 // mechanism is used to determine the local entity (user). And the 'id' attribute is available after the first 2240 // response from the server (see e.g. RFC 6120 § 9.1.1 Step 2.) 2241 CharSequence to = getXMPPServiceDomain(); 2242 CharSequence from = null; 2243 CharSequence localpart = config.getUsername(); 2244 if (localpart != null) { 2245 from = XmppStringUtils.completeJidFrom(localpart, to); 2246 } 2247 String id = getStreamId(); 2248 2249 StreamOpen streamOpen = new StreamOpen(to, from, id, config.getXmlLang(), StreamOpen.StreamContentNamespace.client); 2250 sendNonza(streamOpen); 2251 2252 XmlEnvironment.Builder xmlEnvironmentBuilder = XmlEnvironment.builder(); 2253 xmlEnvironmentBuilder.with(streamOpen); 2254 outgoingStreamXmlEnvironment = xmlEnvironmentBuilder.build(); 2255 } 2256 2257 protected final SmackTlsContext getSmackTlsContext() { 2258 return config.smackTlsContext; 2259 } 2260}