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