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