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