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