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