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