001/**
002 *
003 * Copyright 2009 Jive Software.
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.ArrayList;
023import java.util.Collection;
024import java.util.HashMap;
025import java.util.LinkedHashMap;
026import java.util.LinkedList;
027import java.util.List;
028import java.util.Map;
029import java.util.Set;
030import java.util.concurrent.ConcurrentLinkedQueue;
031import java.util.concurrent.CopyOnWriteArraySet;
032import java.util.concurrent.Executor;
033import java.util.concurrent.ExecutorService;
034import java.util.concurrent.Executors;
035import java.util.concurrent.ScheduledExecutorService;
036import java.util.concurrent.ScheduledFuture;
037import java.util.concurrent.ThreadFactory;
038import java.util.concurrent.TimeUnit;
039import java.util.concurrent.atomic.AtomicInteger;
040import java.util.concurrent.locks.Lock;
041import java.util.concurrent.locks.ReentrantLock;
042import java.util.logging.Level;
043import java.util.logging.Logger;
044
045import org.jivesoftware.smack.ConnectionConfiguration.SecurityMode;
046import org.jivesoftware.smack.SmackConfiguration.UnknownIqRequestReplyMode;
047import org.jivesoftware.smack.SmackException.AlreadyConnectedException;
048import org.jivesoftware.smack.SmackException.AlreadyLoggedInException;
049import org.jivesoftware.smack.SmackException.NoResponseException;
050import org.jivesoftware.smack.SmackException.NotConnectedException;
051import org.jivesoftware.smack.SmackException.NotLoggedInException;
052import org.jivesoftware.smack.SmackException.ResourceBindingNotOfferedException;
053import org.jivesoftware.smack.SmackException.SecurityRequiredByClientException;
054import org.jivesoftware.smack.SmackException.SecurityRequiredException;
055import org.jivesoftware.smack.SmackFuture.InternalSmackFuture;
056import org.jivesoftware.smack.XMPPException.StreamErrorException;
057import org.jivesoftware.smack.XMPPException.XMPPErrorException;
058import org.jivesoftware.smack.compress.packet.Compress;
059import org.jivesoftware.smack.compression.XMPPInputOutputStream;
060import org.jivesoftware.smack.debugger.SmackDebugger;
061import org.jivesoftware.smack.debugger.SmackDebuggerFactory;
062import org.jivesoftware.smack.filter.IQReplyFilter;
063import org.jivesoftware.smack.filter.StanzaFilter;
064import org.jivesoftware.smack.filter.StanzaIdFilter;
065import org.jivesoftware.smack.iqrequest.IQRequestHandler;
066import org.jivesoftware.smack.packet.Bind;
067import org.jivesoftware.smack.packet.ErrorIQ;
068import org.jivesoftware.smack.packet.ExtensionElement;
069import org.jivesoftware.smack.packet.IQ;
070import org.jivesoftware.smack.packet.Mechanisms;
071import org.jivesoftware.smack.packet.Message;
072import org.jivesoftware.smack.packet.Nonza;
073import org.jivesoftware.smack.packet.Presence;
074import org.jivesoftware.smack.packet.Session;
075import org.jivesoftware.smack.packet.Stanza;
076import org.jivesoftware.smack.packet.StanzaError;
077import org.jivesoftware.smack.packet.StartTls;
078import org.jivesoftware.smack.packet.StreamError;
079import org.jivesoftware.smack.parsing.ParsingExceptionCallback;
080import org.jivesoftware.smack.provider.ExtensionElementProvider;
081import org.jivesoftware.smack.provider.ProviderManager;
082import org.jivesoftware.smack.sasl.core.SASLAnonymous;
083import org.jivesoftware.smack.util.Async;
084import org.jivesoftware.smack.util.DNSUtil;
085import org.jivesoftware.smack.util.Objects;
086import org.jivesoftware.smack.util.PacketParserUtils;
087import org.jivesoftware.smack.util.ParserUtils;
088import org.jivesoftware.smack.util.StringUtils;
089import org.jivesoftware.smack.util.dns.HostAddress;
090
091import org.jxmpp.jid.DomainBareJid;
092import org.jxmpp.jid.EntityFullJid;
093import org.jxmpp.jid.Jid;
094import org.jxmpp.jid.parts.Resourcepart;
095import org.jxmpp.util.XmppStringUtils;
096import org.minidns.dnsname.DnsName;
097import org.xmlpull.v1.XmlPullParser;
098
099
100/**
101 * This abstract class is commonly used as super class for XMPP connection mechanisms like TCP and BOSH. Hence it
102 * provides the methods for connection state management, like {@link #connect()}, {@link #login()} and
103 * {@link #disconnect()} (which are deliberately not provided by the {@link XMPPConnection} interface).
104 * <p>
105 * <b>Note:</b> The default entry point to Smack's documentation is {@link XMPPConnection}. If you are getting started
106 * with Smack, then head over to {@link XMPPConnection} and the come back here.
107 * </p>
108 * <h2>Parsing Exceptions</h2>
109 * <p>
110 * In case a Smack parser (Provider) throws those exceptions are handled over to the {@link ParsingExceptionCallback}. A
111 * common cause for a provider throwing is illegal input, for example a non-numeric String where only Integers are
112 * allowed. Smack's <em>default behavior</em> follows the <b>"fail-hard per default"</b> principle leading to a
113 * termination of the connection on parsing exceptions. This default was chosen to make users eventually aware that they
114 * should configure their own callback and handle those exceptions to prevent the disconnect. Handle a parsing exception
115 * could be as simple as using a non-throwing no-op callback, which would cause the faulty stream element to be taken
116 * out of the stream, i.e., Smack behaves like that element was never received.
117 * </p>
118 * <p>
119 * If the parsing exception is because Smack received illegal input, then please consider informing the authors of the
120 * originating entity about that. If it was thrown because of an bug in a Smack parser, then please consider filling a
121 * bug with Smack.
122 * </p>
123 * <h3>Managing the parsing exception callback</h3>
124 * <p>
125 * The "fail-hard per default" behavior is achieved by using the
126 * {@link org.jivesoftware.smack.parsing.ExceptionThrowingCallbackWithHint} as default parsing exception callback. You
127 * can change the behavior using {@link #setParsingExceptionCallback(ParsingExceptionCallback)} to set a new callback.
128 * Use {@link org.jivesoftware.smack.SmackConfiguration#setDefaultParsingExceptionCallback(ParsingExceptionCallback)} to
129 * set the default callback.
130 * </p>
131 */
132public abstract class AbstractXMPPConnection implements XMPPConnection {
133    private static final Logger LOGGER = Logger.getLogger(AbstractXMPPConnection.class.getName());
134
135    /**
136     * Counter to uniquely identify connections that are created.
137     */
138    private static final AtomicInteger connectionCounter = new AtomicInteger(0);
139
140    static {
141        // Ensure the SmackConfiguration class is loaded by calling a method in it.
142        SmackConfiguration.getVersion();
143    }
144
145    /**
146     * A collection of ConnectionListeners which listen for connection closing
147     * and reconnection events.
148     */
149    protected final Set<ConnectionListener> connectionListeners =
150            new CopyOnWriteArraySet<>();
151
152    /**
153     * A collection of StanzaCollectors which collects packets for a specified filter
154     * and perform blocking and polling operations on the result queue.
155     * <p>
156     * We use a ConcurrentLinkedQueue here, because its Iterator is weakly
157     * consistent and we want {@link #invokeStanzaCollectorsAndNotifyRecvListeners(Stanza)} for-each
158     * loop to be lock free. As drawback, removing a StanzaCollector is O(n).
159     * The alternative would be a synchronized HashSet, but this would mean a
160     * synchronized block around every usage of <code>collectors</code>.
161     * </p>
162     */
163    private final Collection<StanzaCollector> collectors = new ConcurrentLinkedQueue<>();
164
165    /**
166     * List of PacketListeners that will be notified synchronously when a new stanza was received.
167     */
168    private final Map<StanzaListener, ListenerWrapper> syncRecvListeners = new LinkedHashMap<>();
169
170    /**
171     * List of PacketListeners that will be notified asynchronously when a new stanza was received.
172     */
173    private final Map<StanzaListener, ListenerWrapper> asyncRecvListeners = new LinkedHashMap<>();
174
175    /**
176     * List of PacketListeners that will be notified when a new stanza was sent.
177     */
178    private final Map<StanzaListener, ListenerWrapper> sendListeners =
179            new HashMap<>();
180
181    /**
182     * List of PacketListeners that will be notified when a new stanza is about to be
183     * sent to the server. These interceptors may modify the stanza before it is being
184     * actually sent to the server.
185     */
186    private final Map<StanzaListener, InterceptorWrapper> interceptors =
187            new HashMap<>();
188
189    protected final Lock connectionLock = new ReentrantLock();
190
191    protected final Map<String, ExtensionElement> streamFeatures = new HashMap<>();
192
193    /**
194     * The full JID of the authenticated user, as returned by the resource binding response of the server.
195     * <p>
196     * It is important that we don't infer the user from the login() arguments and the configurations service name, as,
197     * for example, when SASL External is used, the username is not given to login but taken from the 'external'
198     * certificate.
199     * </p>
200     */
201    protected EntityFullJid user;
202
203    protected boolean connected = false;
204
205    /**
206     * The stream ID, see RFC 6120 § 4.7.3
207     */
208    protected String streamId;
209
210    /**
211     * The timeout to wait for a reply in milliseconds.
212     */
213    private long replyTimeout = SmackConfiguration.getDefaultReplyTimeout();
214
215    /**
216     * The SmackDebugger allows to log and debug XML traffic.
217     */
218    protected final SmackDebugger debugger;
219
220    /**
221     * The Reader which is used for the debugger.
222     */
223    protected Reader reader;
224
225    /**
226     * The Writer which is used for the debugger.
227     */
228    protected Writer writer;
229
230    protected final SynchronizationPoint<SmackException> tlsHandled = new SynchronizationPoint<>(this, "establishing TLS");
231
232    /**
233     * Set to success if the last features stanza from the server has been parsed. A XMPP connection
234     * handshake can invoke multiple features stanzas, e.g. when TLS is activated a second feature
235     * stanza is send by the server. This is set to true once the last feature stanza has been
236     * parsed.
237     */
238    protected final SynchronizationPoint<SmackException> lastFeaturesReceived = new SynchronizationPoint<>(
239                    AbstractXMPPConnection.this, "last stream features received from server");
240
241    /**
242     * Set to success if the SASL feature has been received.
243     */
244    protected final SynchronizationPoint<XMPPException> saslFeatureReceived = new SynchronizationPoint<>(
245                    AbstractXMPPConnection.this, "SASL mechanisms stream feature from server");
246
247    /**
248     * The SASLAuthentication manager that is responsible for authenticating with the server.
249     */
250    protected final SASLAuthentication saslAuthentication;
251
252    /**
253     * A number to uniquely identify connections that are created. This is distinct from the
254     * connection ID, which is a value sent by the server once a connection is made.
255     */
256    protected final int connectionCounterValue = connectionCounter.getAndIncrement();
257
258    /**
259     * Holds the initial configuration used while creating the connection.
260     */
261    protected final ConnectionConfiguration config;
262
263    /**
264     * Defines how the from attribute of outgoing stanzas should be handled.
265     */
266    private FromMode fromMode = FromMode.OMITTED;
267
268    protected XMPPInputOutputStream compressionHandler;
269
270    private ParsingExceptionCallback parsingExceptionCallback = SmackConfiguration.getDefaultParsingExceptionCallback();
271
272    /**
273     * This scheduled thread pool executor is used to remove pending callbacks.
274     */
275    protected static final ScheduledExecutorService SCHEDULED_EXECUTOR_SERVICE = Executors.newSingleThreadScheduledExecutor(
276            new ThreadFactory() {
277                @Override
278                public Thread newThread(Runnable runnable) {
279                    Thread thread = new Thread(runnable);
280                    thread.setName("Smack Scheduled Executor Service");
281                    thread.setDaemon(true);
282                    return thread;
283                }
284            });
285
286    /**
287     * A cached thread pool executor service with custom thread factory to set meaningful names on the threads and set
288     * them 'daemon'.
289     */
290    private static final ExecutorService CACHED_EXECUTOR_SERVICE = Executors.newCachedThreadPool(new ThreadFactory() {
291        @Override
292        public Thread newThread(Runnable runnable) {
293            Thread thread = new Thread(runnable);
294            thread.setName("Smack Cached Executor");
295            thread.setDaemon(true);
296            return thread;
297        }
298    });
299
300    private static final AsyncButOrdered<AbstractXMPPConnection> ASYNC_BUT_ORDERED = new AsyncButOrdered<>();
301
302    /**
303     * The used host to establish the connection to
304     */
305    protected String host;
306
307    /**
308     * The used port to establish the connection to
309     */
310    protected int port;
311
312    /**
313     * Flag that indicates if the user is currently authenticated with the server.
314     */
315    protected boolean authenticated = false;
316
317    /**
318     * Flag that indicates if the user was authenticated with the server when the connection
319     * to the server was closed (abruptly or not).
320     */
321    protected boolean wasAuthenticated = false;
322
323    private final Map<String, IQRequestHandler> setIqRequestHandler = new HashMap<>();
324    private final Map<String, IQRequestHandler> getIqRequestHandler = new HashMap<>();
325
326    /**
327     * Create a new XMPPConnection to an XMPP server.
328     *
329     * @param configuration The configuration which is used to establish the connection.
330     */
331    protected AbstractXMPPConnection(ConnectionConfiguration configuration) {
332        saslAuthentication = new SASLAuthentication(this, configuration);
333        config = configuration;
334        SmackDebuggerFactory debuggerFactory = configuration.getDebuggerFactory();
335        if (debuggerFactory != null) {
336            debugger = debuggerFactory.create(this);
337        } else {
338            debugger = null;
339        }
340        // Notify listeners that a new connection has been established
341        for (ConnectionCreationListener listener : XMPPConnectionRegistry.getConnectionCreationListeners()) {
342            listener.connectionCreated(this);
343        }
344    }
345
346    /**
347     * Get the connection configuration used by this connection.
348     *
349     * @return the connection configuration.
350     */
351    public ConnectionConfiguration getConfiguration() {
352        return config;
353    }
354
355    @Override
356    public DomainBareJid getXMPPServiceDomain() {
357        if (xmppServiceDomain != null) {
358            return xmppServiceDomain;
359        }
360        return config.getXMPPServiceDomain();
361    }
362
363    @Override
364    public String getHost() {
365        return host;
366    }
367
368    @Override
369    public int getPort() {
370        return port;
371    }
372
373    @Override
374    public abstract boolean isSecureConnection();
375
376    protected abstract void sendStanzaInternal(Stanza packet) throws NotConnectedException, InterruptedException;
377
378    @Override
379    public abstract void sendNonza(Nonza element) throws NotConnectedException, InterruptedException;
380
381    @Override
382    public abstract boolean isUsingCompression();
383
384    /**
385     * Establishes a connection to the XMPP server. It basically
386     * creates and maintains a connection to the server.
387     * <p>
388     * Listeners will be preserved from a previous connection.
389     * </p>
390     *
391     * @throws XMPPException if an error occurs on the XMPP protocol level.
392     * @throws SmackException if an error occurs somewhere else besides XMPP protocol level.
393     * @throws IOException
394     * @return a reference to this object, to chain <code>connect()</code> with <code>login()</code>.
395     * @throws InterruptedException
396     */
397    public synchronized AbstractXMPPConnection connect() throws SmackException, IOException, XMPPException, InterruptedException {
398        // Check if not already connected
399        throwAlreadyConnectedExceptionIfAppropriate();
400
401        // Reset the connection state
402        saslAuthentication.init();
403        saslFeatureReceived.init();
404        lastFeaturesReceived.init();
405        tlsHandled.init();
406        streamId = null;
407
408        // Perform the actual connection to the XMPP service
409        connectInternal();
410
411        // If TLS is required but the server doesn't offer it, disconnect
412        // from the server and throw an error. First check if we've already negotiated TLS
413        // and are secure, however (features get parsed a second time after TLS is established).
414        if (!isSecureConnection() && getConfiguration().getSecurityMode() == SecurityMode.required) {
415            shutdown();
416            throw new SecurityRequiredByClientException();
417        }
418
419        // Make note of the fact that we're now connected.
420        connected = true;
421        callConnectionConnectedListener();
422
423        return this;
424    }
425
426    /**
427     * Abstract method that concrete subclasses of XMPPConnection need to implement to perform their
428     * way of XMPP connection establishment. Implementations are required to perform an automatic
429     * login if the previous connection state was logged (authenticated).
430     *
431     * @throws SmackException
432     * @throws IOException
433     * @throws XMPPException
434     * @throws InterruptedException
435     */
436    protected abstract void connectInternal() throws SmackException, IOException, XMPPException, InterruptedException;
437
438    private String usedUsername, usedPassword;
439
440    /**
441     * The resourcepart used for this connection. May not be the resulting resourcepart if it's null or overridden by the XMPP service.
442     */
443    private Resourcepart usedResource;
444
445    /**
446     * Logs in to the server using the strongest SASL mechanism supported by
447     * the server. If more than the connection's default stanza timeout elapses in each step of the
448     * authentication process without a response from the server, a
449     * {@link SmackException.NoResponseException} will be thrown.
450     * <p>
451     * Before logging in (i.e. authenticate) to the server the connection must be connected
452     * by calling {@link #connect}.
453     * </p>
454     * <p>
455     * It is possible to log in without sending an initial available presence by using
456     * {@link ConnectionConfiguration.Builder#setSendPresence(boolean)}.
457     * Finally, if you want to not pass a password and instead use a more advanced mechanism
458     * while using SASL then you may be interested in using
459     * {@link ConnectionConfiguration.Builder#setCallbackHandler(javax.security.auth.callback.CallbackHandler)}.
460     * For more advanced login settings see {@link ConnectionConfiguration}.
461     * </p>
462     *
463     * @throws XMPPException if an error occurs on the XMPP protocol level.
464     * @throws SmackException if an error occurs somewhere else besides XMPP protocol level.
465     * @throws IOException if an I/O error occurs during login.
466     * @throws InterruptedException
467     */
468    public synchronized void login() throws XMPPException, SmackException, IOException, InterruptedException {
469        // The previously used username, password and resource take over precedence over the
470        // ones from the connection configuration
471        CharSequence username = usedUsername != null ? usedUsername : config.getUsername();
472        String password = usedPassword != null ? usedPassword : config.getPassword();
473        Resourcepart resource = usedResource != null ? usedResource : config.getResource();
474        login(username, password, resource);
475    }
476
477    /**
478     * Same as {@link #login(CharSequence, String, Resourcepart)}, but takes the resource from the connection
479     * configuration.
480     *
481     * @param username
482     * @param password
483     * @throws XMPPException
484     * @throws SmackException
485     * @throws IOException
486     * @throws InterruptedException
487     * @see #login
488     */
489    public synchronized void login(CharSequence username, String password) throws XMPPException, SmackException,
490                    IOException, InterruptedException {
491        login(username, password, config.getResource());
492    }
493
494    /**
495     * Login with the given username (authorization identity). You may omit the password if a callback handler is used.
496     * If resource is null, then the server will generate one.
497     *
498     * @param username
499     * @param password
500     * @param resource
501     * @throws XMPPException
502     * @throws SmackException
503     * @throws IOException
504     * @throws InterruptedException
505     * @see #login
506     */
507    public synchronized void login(CharSequence username, String password, Resourcepart resource) throws XMPPException,
508                    SmackException, IOException, InterruptedException {
509        if (!config.allowNullOrEmptyUsername) {
510            StringUtils.requireNotNullOrEmpty(username, "Username must not be null or empty");
511        }
512        throwNotConnectedExceptionIfAppropriate("Did you call connect() before login()?");
513        throwAlreadyLoggedInExceptionIfAppropriate();
514        usedUsername = username != null ? username.toString() : null;
515        usedPassword = password;
516        usedResource = resource;
517        loginInternal(usedUsername, usedPassword, usedResource);
518    }
519
520    protected abstract void loginInternal(String username, String password, Resourcepart resource)
521                    throws XMPPException, SmackException, IOException, InterruptedException;
522
523    @Override
524    public final boolean isConnected() {
525        return connected;
526    }
527
528    @Override
529    public final boolean isAuthenticated() {
530        return authenticated;
531    }
532
533    @Override
534    public final EntityFullJid getUser() {
535        return user;
536    }
537
538    @Override
539    public String getStreamId() {
540        if (!isConnected()) {
541            return null;
542        }
543        return streamId;
544    }
545
546    protected void bindResourceAndEstablishSession(Resourcepart resource) throws XMPPErrorException,
547                    SmackException, InterruptedException {
548
549        // Wait until either:
550        // - the servers last features stanza has been parsed
551        // - the timeout occurs
552        LOGGER.finer("Waiting for last features to be received before continuing with resource binding");
553        lastFeaturesReceived.checkIfSuccessOrWaitOrThrow();
554
555
556        if (!hasFeature(Bind.ELEMENT, Bind.NAMESPACE)) {
557            // Server never offered resource binding, which is REQUIRED in XMPP client and
558            // server implementations as per RFC6120 7.2
559            throw new ResourceBindingNotOfferedException();
560        }
561
562        // Resource binding, see RFC6120 7.
563        // Note that we can not use IQReplyFilter here, since the users full JID is not yet
564        // available. It will become available right after the resource has been successfully bound.
565        Bind bindResource = Bind.newSet(resource);
566        StanzaCollector packetCollector = createStanzaCollectorAndSend(new StanzaIdFilter(bindResource), bindResource);
567        Bind response = packetCollector.nextResultOrThrow();
568        // Set the connections user to the result of resource binding. It is important that we don't infer the user
569        // from the login() arguments and the configurations service name, as, for example, when SASL External is used,
570        // the username is not given to login but taken from the 'external' certificate.
571        user = response.getJid();
572        xmppServiceDomain = user.asDomainBareJid();
573
574        Session.Feature sessionFeature = getFeature(Session.ELEMENT, Session.NAMESPACE);
575        // Only bind the session if it's announced as stream feature by the server, is not optional and not disabled
576        // For more information see http://tools.ietf.org/html/draft-cridland-xmpp-session-01
577        if (sessionFeature != null && !sessionFeature.isOptional()) {
578            Session session = new Session();
579            packetCollector = createStanzaCollectorAndSend(new StanzaIdFilter(session), session);
580            packetCollector.nextResultOrThrow();
581        }
582    }
583
584    protected void afterSuccessfulLogin(final boolean resumed) throws NotConnectedException, InterruptedException {
585        // Indicate that we're now authenticated.
586        this.authenticated = true;
587
588        // If debugging is enabled, change the the debug window title to include the
589        // name we are now logged-in as.
590        // If DEBUG was set to true AFTER the connection was created the debugger
591        // will be null
592        if (debugger != null) {
593            debugger.userHasLogged(user);
594        }
595        callConnectionAuthenticatedListener(resumed);
596
597        // Set presence to online. It is important that this is done after
598        // callConnectionAuthenticatedListener(), as this call will also
599        // eventually load the roster. And we should load the roster before we
600        // send the initial presence.
601        if (config.isSendPresence() && !resumed) {
602            sendStanza(new Presence(Presence.Type.available));
603        }
604    }
605
606    @Override
607    public final boolean isAnonymous() {
608        return isAuthenticated() && SASLAnonymous.NAME.equals(getUsedSaslMechansism());
609    }
610
611    /**
612     * Get the name of the SASL mechanism that was used to authenticate this connection. This returns the name of
613     * mechanism which was used the last time this connection was authenticated, and will return <code>null</code> if
614     * this connection was not authenticated before.
615     *
616     * @return the name of the used SASL mechanism.
617     * @since 4.2
618     */
619    public final String getUsedSaslMechansism() {
620        return saslAuthentication.getNameOfLastUsedSaslMechansism();
621    }
622
623    private DomainBareJid xmppServiceDomain;
624
625    protected List<HostAddress> hostAddresses;
626
627    /**
628     * Populates {@link #hostAddresses} with the resolved addresses or with the configured host address. If no host
629     * address was configured and all lookups failed, for example with NX_DOMAIN, then {@link #hostAddresses} will be
630     * populated with the empty list.
631     *
632     * @return a list of host addresses where DNS (SRV) RR resolution failed.
633     */
634    protected List<HostAddress> populateHostAddresses() {
635        List<HostAddress> failedAddresses = new LinkedList<>();
636        if (config.hostAddress != null) {
637            hostAddresses = new ArrayList<>(1);
638            HostAddress hostAddress = new HostAddress(config.port, config.hostAddress);
639            hostAddresses.add(hostAddress);
640        }
641        else if (config.host != null) {
642            hostAddresses = new ArrayList<>(1);
643            HostAddress hostAddress = DNSUtil.getDNSResolver().lookupHostAddress(config.host, config.port, failedAddresses, config.getDnssecMode());
644            if (hostAddress != null) {
645                hostAddresses.add(hostAddress);
646            }
647        } else {
648            // N.B.: Important to use config.serviceName and not AbstractXMPPConnection.serviceName
649            DnsName dnsName = DnsName.from(config.getXMPPServiceDomain());
650            hostAddresses = DNSUtil.resolveXMPPServiceDomain(dnsName, failedAddresses, config.getDnssecMode());
651        }
652        // Either the populated host addresses are not empty *or* there must be at least one failed address.
653        assert (!hostAddresses.isEmpty() || !failedAddresses.isEmpty());
654        return failedAddresses;
655    }
656
657    protected Lock getConnectionLock() {
658        return connectionLock;
659    }
660
661    protected void throwNotConnectedExceptionIfAppropriate() throws NotConnectedException {
662        throwNotConnectedExceptionIfAppropriate(null);
663    }
664
665    protected void throwNotConnectedExceptionIfAppropriate(String optionalHint) throws NotConnectedException {
666        if (!isConnected()) {
667            throw new NotConnectedException(optionalHint);
668        }
669    }
670
671    protected void throwAlreadyConnectedExceptionIfAppropriate() throws AlreadyConnectedException {
672        if (isConnected()) {
673            throw new AlreadyConnectedException();
674        }
675    }
676
677    protected void throwAlreadyLoggedInExceptionIfAppropriate() throws AlreadyLoggedInException {
678        if (isAuthenticated()) {
679            throw new AlreadyLoggedInException();
680        }
681    }
682
683    @Override
684    public void sendStanza(Stanza stanza) throws NotConnectedException, InterruptedException {
685        Objects.requireNonNull(stanza, "Stanza must not be null");
686        assert (stanza instanceof Message || stanza instanceof Presence || stanza instanceof IQ);
687
688        throwNotConnectedExceptionIfAppropriate();
689        switch (fromMode) {
690        case OMITTED:
691            stanza.setFrom((Jid) null);
692            break;
693        case USER:
694            stanza.setFrom(getUser());
695            break;
696        case UNCHANGED:
697        default:
698            break;
699        }
700        // Invoke interceptors for the new stanza that is about to be sent. Interceptors may modify
701        // the content of the stanza.
702        firePacketInterceptors(stanza);
703        sendStanzaInternal(stanza);
704    }
705
706    /**
707     * Returns the SASLAuthentication manager that is responsible for authenticating with
708     * the server.
709     *
710     * @return the SASLAuthentication manager that is responsible for authenticating with
711     *         the server.
712     */
713    protected SASLAuthentication getSASLAuthentication() {
714        return saslAuthentication;
715    }
716
717    /**
718     * Closes the connection by setting presence to unavailable then closing the connection to
719     * the XMPP server. The XMPPConnection can still be used for connecting to the server
720     * again.
721     *
722     */
723    public void disconnect() {
724        Presence unavailablePresence = null;
725        if (isAuthenticated()) {
726            unavailablePresence = new Presence(Presence.Type.unavailable);
727        }
728        try {
729            disconnect(unavailablePresence);
730        }
731        catch (NotConnectedException e) {
732            LOGGER.log(Level.FINEST, "Connection is already disconnected", e);
733        }
734    }
735
736    /**
737     * Closes the connection. A custom unavailable presence is sent to the server, followed
738     * by closing the stream. The XMPPConnection can still be used for connecting to the server
739     * again. A custom unavailable presence is useful for communicating offline presence
740     * information such as "On vacation". Typically, just the status text of the presence
741     * stanza is set with online information, but most XMPP servers will deliver the full
742     * presence stanza with whatever data is set.
743     *
744     * @param unavailablePresence the optional presence stanza to send during shutdown.
745     * @throws NotConnectedException
746     */
747    public synchronized void disconnect(Presence unavailablePresence) throws NotConnectedException {
748        if (unavailablePresence != null) {
749            try {
750                sendStanza(unavailablePresence);
751            } catch (InterruptedException e) {
752                LOGGER.log(Level.FINE,
753                        "Was interrupted while sending unavailable presence. Continuing to disconnect the connection",
754                        e);
755            }
756        }
757        shutdown();
758        callConnectionClosedListener();
759    }
760
761    /**
762     * Shuts the current connection down.
763     */
764    protected abstract void shutdown();
765
766    @Override
767    public void addConnectionListener(ConnectionListener connectionListener) {
768        if (connectionListener == null) {
769            return;
770        }
771        connectionListeners.add(connectionListener);
772    }
773
774    @Override
775    public void removeConnectionListener(ConnectionListener connectionListener) {
776        connectionListeners.remove(connectionListener);
777    }
778
779    @Override
780    public <I extends IQ> I sendIqRequestAndWaitForResponse(IQ request)
781            throws NoResponseException, XMPPErrorException, NotConnectedException, InterruptedException {
782        StanzaCollector collector = createStanzaCollectorAndSend(request);
783        IQ resultResponse = collector.nextResultOrThrow();
784        @SuppressWarnings("unchecked")
785        I concreteResultResponse = (I) resultResponse;
786        return concreteResultResponse;
787    }
788
789    @Override
790    public StanzaCollector createStanzaCollectorAndSend(IQ packet) throws NotConnectedException, InterruptedException {
791        StanzaFilter packetFilter = new IQReplyFilter(packet, this);
792        // Create the packet collector before sending the packet
793        StanzaCollector packetCollector = createStanzaCollectorAndSend(packetFilter, packet);
794        return packetCollector;
795    }
796
797    @Override
798    public StanzaCollector createStanzaCollectorAndSend(StanzaFilter packetFilter, Stanza packet)
799                    throws NotConnectedException, InterruptedException {
800        StanzaCollector.Configuration configuration = StanzaCollector.newConfiguration()
801                        .setStanzaFilter(packetFilter)
802                        .setRequest(packet);
803        // Create the packet collector before sending the packet
804        StanzaCollector packetCollector = createStanzaCollector(configuration);
805        try {
806            // Now we can send the packet as the collector has been created
807            sendStanza(packet);
808        }
809        catch (InterruptedException | NotConnectedException | RuntimeException e) {
810            packetCollector.cancel();
811            throw e;
812        }
813        return packetCollector;
814    }
815
816    @Override
817    public StanzaCollector createStanzaCollector(StanzaFilter packetFilter) {
818        StanzaCollector.Configuration configuration = StanzaCollector.newConfiguration().setStanzaFilter(packetFilter);
819        return createStanzaCollector(configuration);
820    }
821
822    @Override
823    public StanzaCollector createStanzaCollector(StanzaCollector.Configuration configuration) {
824        StanzaCollector collector = new StanzaCollector(this, configuration);
825        // Add the collector to the list of active collectors.
826        collectors.add(collector);
827        return collector;
828    }
829
830    @Override
831    public void removeStanzaCollector(StanzaCollector collector) {
832        collectors.remove(collector);
833    }
834
835    @Override
836    public void addSyncStanzaListener(StanzaListener packetListener, StanzaFilter packetFilter) {
837        if (packetListener == null) {
838            throw new NullPointerException("Packet listener is null.");
839        }
840        ListenerWrapper wrapper = new ListenerWrapper(packetListener, packetFilter);
841        synchronized (syncRecvListeners) {
842            syncRecvListeners.put(packetListener, wrapper);
843        }
844    }
845
846    @Override
847    public boolean removeSyncStanzaListener(StanzaListener packetListener) {
848        synchronized (syncRecvListeners) {
849            return syncRecvListeners.remove(packetListener) != null;
850        }
851    }
852
853    @Override
854    public void addAsyncStanzaListener(StanzaListener packetListener, StanzaFilter packetFilter) {
855        if (packetListener == null) {
856            throw new NullPointerException("Packet listener is null.");
857        }
858        ListenerWrapper wrapper = new ListenerWrapper(packetListener, packetFilter);
859        synchronized (asyncRecvListeners) {
860            asyncRecvListeners.put(packetListener, wrapper);
861        }
862    }
863
864    @Override
865    public boolean removeAsyncStanzaListener(StanzaListener packetListener) {
866        synchronized (asyncRecvListeners) {
867            return asyncRecvListeners.remove(packetListener) != null;
868        }
869    }
870
871    @Deprecated
872    @Override
873    public void addPacketSendingListener(StanzaListener packetListener, StanzaFilter packetFilter) {
874        addStanzaSendingListener(packetListener, packetFilter);
875    }
876
877    @Override
878    public void addStanzaSendingListener(StanzaListener packetListener, StanzaFilter packetFilter) {
879        if (packetListener == null) {
880            throw new NullPointerException("Packet listener is null.");
881        }
882        ListenerWrapper wrapper = new ListenerWrapper(packetListener, packetFilter);
883        synchronized (sendListeners) {
884            sendListeners.put(packetListener, wrapper);
885        }
886    }
887
888    @Deprecated
889    @Override
890    public void removePacketSendingListener(StanzaListener packetListener) {
891        removeStanzaSendingListener(packetListener);
892    }
893
894    @Override
895    public void removeStanzaSendingListener(StanzaListener packetListener) {
896        synchronized (sendListeners) {
897            sendListeners.remove(packetListener);
898        }
899    }
900
901    /**
902     * Process all stanza listeners for sending packets.
903     * <p>
904     * Compared to {@link #firePacketInterceptors(Stanza)}, the listeners will be invoked in a new thread.
905     * </p>
906     *
907     * @param packet the stanza to process.
908     */
909    @SuppressWarnings("javadoc")
910    protected void firePacketSendingListeners(final Stanza packet) {
911        final SmackDebugger debugger = this.debugger;
912        if (debugger != null) {
913            debugger.onOutgoingStreamElement(packet);
914        }
915
916        final List<StanzaListener> listenersToNotify = new LinkedList<>();
917        synchronized (sendListeners) {
918            for (ListenerWrapper listenerWrapper : sendListeners.values()) {
919                if (listenerWrapper.filterMatches(packet)) {
920                    listenersToNotify.add(listenerWrapper.getListener());
921                }
922            }
923        }
924        if (listenersToNotify.isEmpty()) {
925            return;
926        }
927        // Notify in a new thread, because we can
928        asyncGo(new Runnable() {
929            @Override
930            public void run() {
931                for (StanzaListener listener : listenersToNotify) {
932                    try {
933                        listener.processStanza(packet);
934                    }
935                    catch (Exception e) {
936                        LOGGER.log(Level.WARNING, "Sending listener threw exception", e);
937                        continue;
938                    }
939                }
940            }
941        });
942    }
943
944    @Deprecated
945    @Override
946    public void addPacketInterceptor(StanzaListener packetInterceptor,
947                                     StanzaFilter packetFilter) {
948        addStanzaInterceptor(packetInterceptor, packetFilter);
949    }
950
951    @Override
952    public void addStanzaInterceptor(StanzaListener packetInterceptor,
953            StanzaFilter packetFilter) {
954        if (packetInterceptor == null) {
955            throw new NullPointerException("Packet interceptor is null.");
956        }
957        InterceptorWrapper interceptorWrapper = new InterceptorWrapper(packetInterceptor, packetFilter);
958        synchronized (interceptors) {
959            interceptors.put(packetInterceptor, interceptorWrapper);
960        }
961    }
962
963    @Deprecated
964    @Override
965    public void removePacketInterceptor(StanzaListener packetInterceptor) {
966        removeStanzaInterceptor(packetInterceptor);
967    }
968
969    @Override
970    public void removeStanzaInterceptor(StanzaListener packetInterceptor) {
971        synchronized (interceptors) {
972            interceptors.remove(packetInterceptor);
973        }
974    }
975
976    /**
977     * Process interceptors. Interceptors may modify the stanza that is about to be sent.
978     * Since the thread that requested to send the stanza will invoke all interceptors, it
979     * is important that interceptors perform their work as soon as possible so that the
980     * thread does not remain blocked for a long period.
981     *
982     * @param packet the stanza that is going to be sent to the server
983     */
984    private void firePacketInterceptors(Stanza packet) {
985        List<StanzaListener> interceptorsToInvoke = new LinkedList<>();
986        synchronized (interceptors) {
987            for (InterceptorWrapper interceptorWrapper : interceptors.values()) {
988                if (interceptorWrapper.filterMatches(packet)) {
989                    interceptorsToInvoke.add(interceptorWrapper.getInterceptor());
990                }
991            }
992        }
993        for (StanzaListener interceptor : interceptorsToInvoke) {
994            try {
995                interceptor.processStanza(packet);
996            } catch (Exception e) {
997                LOGGER.log(Level.SEVERE, "Packet interceptor threw exception", e);
998            }
999        }
1000    }
1001
1002    /**
1003     * Initialize the {@link #debugger}. You can specify a customized {@link SmackDebugger}
1004     * by setup the system property <code>smack.debuggerClass</code> to the implementation.
1005     *
1006     * @throws IllegalStateException if the reader or writer isn't yet initialized.
1007     * @throws IllegalArgumentException if the SmackDebugger can't be loaded.
1008     */
1009    protected void initDebugger() {
1010        if (reader == null || writer == null) {
1011            throw new NullPointerException("Reader or writer isn't initialized.");
1012        }
1013        // If debugging is enabled, we open a window and write out all network traffic.
1014        if (debugger != null) {
1015            // Obtain new reader and writer from the existing debugger
1016            reader = debugger.newConnectionReader(reader);
1017            writer = debugger.newConnectionWriter(writer);
1018        }
1019    }
1020
1021    @Override
1022    public long getReplyTimeout() {
1023        return replyTimeout;
1024    }
1025
1026    @Override
1027    public void setReplyTimeout(long timeout) {
1028        replyTimeout = timeout;
1029    }
1030
1031    private SmackConfiguration.UnknownIqRequestReplyMode unknownIqRequestReplyMode = SmackConfiguration.getUnknownIqRequestReplyMode();
1032
1033    /**
1034     * Set how Smack behaves when an unknown IQ request has been received.
1035     *
1036     * @param unknownIqRequestReplyMode reply mode.
1037     */
1038    public void setUnknownIqRequestReplyMode(UnknownIqRequestReplyMode unknownIqRequestReplyMode) {
1039        this.unknownIqRequestReplyMode = Objects.requireNonNull(unknownIqRequestReplyMode, "Mode must not be null");
1040    }
1041
1042    protected void parseAndProcessStanza(XmlPullParser parser) throws Exception {
1043        ParserUtils.assertAtStartTag(parser);
1044        int parserDepth = parser.getDepth();
1045        Stanza stanza = null;
1046        try {
1047            stanza = PacketParserUtils.parseStanza(parser);
1048        }
1049        catch (Exception e) {
1050            CharSequence content = PacketParserUtils.parseContentDepth(parser,
1051                            parserDepth);
1052            UnparseableStanza message = new UnparseableStanza(content, e);
1053            ParsingExceptionCallback callback = getParsingExceptionCallback();
1054            if (callback != null) {
1055                callback.handleUnparsableStanza(message);
1056            }
1057        }
1058        ParserUtils.assertAtEndTag(parser);
1059        if (stanza != null) {
1060            processStanza(stanza);
1061        }
1062    }
1063
1064    /**
1065     * Processes a stanza after it's been fully parsed by looping through the installed
1066     * stanza collectors and listeners and letting them examine the stanza to see if
1067     * they are a match with the filter.
1068     *
1069     * @param stanza the stanza to process.
1070     * @throws InterruptedException
1071     */
1072    protected void processStanza(final Stanza stanza) throws InterruptedException {
1073        assert (stanza != null);
1074
1075        final SmackDebugger debugger = this.debugger;
1076        if (debugger != null) {
1077            debugger.onIncomingStreamElement(stanza);
1078        }
1079
1080        lastStanzaReceived = System.currentTimeMillis();
1081        // Deliver the incoming packet to listeners.
1082        invokeStanzaCollectorsAndNotifyRecvListeners(stanza);
1083    }
1084
1085    /**
1086     * Invoke {@link StanzaCollector#processStanza(Stanza)} for every
1087     * StanzaCollector with the given packet. Also notify the receive listeners with a matching stanza filter about the packet.
1088     * <p>
1089     * This method will be invoked by the connections incoming processing thread which may be shared across multiple connections and
1090     * thus it is important that no user code, e.g. in form of a callback, is invoked by this method. For the same reason,
1091     * this method must not block for an extended period of time.
1092     * </p>
1093     *
1094     * @param packet the stanza to notify the StanzaCollectors and receive listeners about.
1095     */
1096    protected void invokeStanzaCollectorsAndNotifyRecvListeners(final Stanza packet) {
1097        if (packet instanceof IQ) {
1098            final IQ iq = (IQ) packet;
1099            if (iq.isRequestIQ()) {
1100                final IQ iqRequest = iq;
1101                final String key = XmppStringUtils.generateKey(iq.getChildElementName(), iq.getChildElementNamespace());
1102                IQRequestHandler iqRequestHandler;
1103                final IQ.Type type = iq.getType();
1104                switch (type) {
1105                case set:
1106                    synchronized (setIqRequestHandler) {
1107                        iqRequestHandler = setIqRequestHandler.get(key);
1108                    }
1109                    break;
1110                case get:
1111                    synchronized (getIqRequestHandler) {
1112                        iqRequestHandler = getIqRequestHandler.get(key);
1113                    }
1114                    break;
1115                default:
1116                    throw new IllegalStateException("Should only encounter IQ type 'get' or 'set'");
1117                }
1118                if (iqRequestHandler == null) {
1119                    StanzaError.Condition replyCondition;
1120                    switch (unknownIqRequestReplyMode) {
1121                    case doNotReply:
1122                        return;
1123                    case replyFeatureNotImplemented:
1124                        replyCondition = StanzaError.Condition.feature_not_implemented;
1125                        break;
1126                    case replyServiceUnavailable:
1127                        replyCondition = StanzaError.Condition.service_unavailable;
1128                        break;
1129                    default:
1130                        throw new AssertionError();
1131                    }
1132
1133                    // If the IQ stanza is of type "get" or "set" with no registered IQ request handler, then answer an
1134                    // IQ of type 'error' with condition 'service-unavailable'.
1135                    ErrorIQ errorIQ = IQ.createErrorResponse(iq, StanzaError.getBuilder((
1136                                    replyCondition)));
1137                    try {
1138                        sendStanza(errorIQ);
1139                    }
1140                    catch (InterruptedException | NotConnectedException e) {
1141                        LOGGER.log(Level.WARNING, "Exception while sending error IQ to unkown IQ request", e);
1142                    }
1143                } else {
1144                    Executor executorService = null;
1145                    switch (iqRequestHandler.getMode()) {
1146                    case sync:
1147                        executorService = ASYNC_BUT_ORDERED.asExecutorFor(this);
1148                        break;
1149                    case async:
1150                        executorService = CACHED_EXECUTOR_SERVICE;
1151                        break;
1152                    }
1153                    final IQRequestHandler finalIqRequestHandler = iqRequestHandler;
1154                    executorService.execute(new Runnable() {
1155                        @Override
1156                        public void run() {
1157                            IQ response = finalIqRequestHandler.handleIQRequest(iq);
1158                            if (response == null) {
1159                                // It is not ideal if the IQ request handler does not return an IQ response, because RFC
1160                                // 6120 § 8.1.2 does specify that a response is mandatory. But some APIs, mostly the
1161                                // file transfer one, does not always return a result, so we need to handle this case.
1162                                // Also sometimes a request handler may decide that it's better to not send a response,
1163                                // e.g. to avoid presence leaks.
1164                                return;
1165                            }
1166
1167                            assert (response.getType() == IQ.Type.result || response.getType() == IQ.Type.error);
1168
1169                            response.setTo(iqRequest.getFrom());
1170                            response.setStanzaId(iqRequest.getStanzaId());
1171                            try {
1172                                sendStanza(response);
1173                            }
1174                            catch (InterruptedException | NotConnectedException e) {
1175                                LOGGER.log(Level.WARNING, "Exception while sending response to IQ request", e);
1176                            }
1177                        }
1178                    });
1179                }
1180                // The following returns makes it impossible for packet listeners and collectors to
1181                // filter for IQ request stanzas, i.e. IQs of type 'set' or 'get'. This is the
1182                // desired behavior.
1183                return;
1184            }
1185        }
1186
1187        // First handle the async recv listeners. Note that this code is very similar to what follows a few lines below,
1188        // the only difference is that asyncRecvListeners is used here and that the packet listeners are started in
1189        // their own thread.
1190        final Collection<StanzaListener> listenersToNotify = new LinkedList<>();
1191        synchronized (asyncRecvListeners) {
1192            for (ListenerWrapper listenerWrapper : asyncRecvListeners.values()) {
1193                if (listenerWrapper.filterMatches(packet)) {
1194                    listenersToNotify.add(listenerWrapper.getListener());
1195                }
1196            }
1197        }
1198
1199        for (final StanzaListener listener : listenersToNotify) {
1200            asyncGo(new Runnable() {
1201                @Override
1202                public void run() {
1203                    try {
1204                        listener.processStanza(packet);
1205                    } catch (Exception e) {
1206                        LOGGER.log(Level.SEVERE, "Exception in async packet listener", e);
1207                    }
1208                }
1209            });
1210        }
1211
1212        // Loop through all collectors and notify the appropriate ones.
1213        for (StanzaCollector collector : collectors) {
1214            collector.processStanza(packet);
1215        }
1216
1217        // Notify the receive listeners interested in the packet
1218        listenersToNotify.clear();
1219        synchronized (syncRecvListeners) {
1220            for (ListenerWrapper listenerWrapper : syncRecvListeners.values()) {
1221                if (listenerWrapper.filterMatches(packet)) {
1222                    listenersToNotify.add(listenerWrapper.getListener());
1223                }
1224            }
1225        }
1226
1227        // Decouple incoming stanza processing from listener invocation. Unlike async listeners, this uses a single
1228        // threaded executor service and therefore keeps the order.
1229        ASYNC_BUT_ORDERED.performAsyncButOrdered(this, new Runnable() {
1230            @Override
1231            public void run() {
1232                for (StanzaListener listener : listenersToNotify) {
1233                    try {
1234                        listener.processStanza(packet);
1235                    } catch (NotConnectedException e) {
1236                        LOGGER.log(Level.WARNING, "Got not connected exception, aborting", e);
1237                        break;
1238                    } catch (Exception e) {
1239                        LOGGER.log(Level.SEVERE, "Exception in packet listener", e);
1240                    }
1241                }
1242            }
1243        });
1244    }
1245
1246    /**
1247     * Sets whether the connection has already logged in the server. This method assures that the
1248     * {@link #wasAuthenticated} flag is never reset once it has ever been set.
1249     *
1250     */
1251    protected void setWasAuthenticated() {
1252        // Never reset the flag if the connection has ever been authenticated
1253        if (!wasAuthenticated) {
1254            wasAuthenticated = authenticated;
1255        }
1256    }
1257
1258    protected void callConnectionConnectedListener() {
1259        for (ConnectionListener listener : connectionListeners) {
1260            listener.connected(this);
1261        }
1262    }
1263
1264    protected void callConnectionAuthenticatedListener(boolean resumed) {
1265        for (ConnectionListener listener : connectionListeners) {
1266            try {
1267                listener.authenticated(this, resumed);
1268            } catch (Exception e) {
1269                // Catch and print any exception so we can recover
1270                // from a faulty listener and finish the shutdown process
1271                LOGGER.log(Level.SEVERE, "Exception in authenticated listener", e);
1272            }
1273        }
1274    }
1275
1276    void callConnectionClosedListener() {
1277        for (ConnectionListener listener : connectionListeners) {
1278            try {
1279                listener.connectionClosed();
1280            }
1281            catch (Exception e) {
1282                // Catch and print any exception so we can recover
1283                // from a faulty listener and finish the shutdown process
1284                LOGGER.log(Level.SEVERE, "Error in listener while closing connection", e);
1285            }
1286        }
1287    }
1288
1289    protected void callConnectionClosedOnErrorListener(Exception e) {
1290        boolean logWarning = true;
1291        if (e instanceof StreamErrorException) {
1292            StreamErrorException see = (StreamErrorException) e;
1293            if (see.getStreamError().getCondition() == StreamError.Condition.not_authorized
1294                            && wasAuthenticated) {
1295                logWarning = false;
1296                LOGGER.log(Level.FINE,
1297                                "Connection closed with not-authorized stream error after it was already authenticated. The account was likely deleted/unregistered on the server");
1298            }
1299        }
1300        if (logWarning) {
1301            LOGGER.log(Level.WARNING, "Connection " + this + " closed with error", e);
1302        }
1303        for (ConnectionListener listener : connectionListeners) {
1304            try {
1305                listener.connectionClosedOnError(e);
1306            }
1307            catch (Exception e2) {
1308                // Catch and print any exception so we can recover
1309                // from a faulty listener
1310                LOGGER.log(Level.SEVERE, "Error in listener while closing connection", e2);
1311            }
1312        }
1313    }
1314
1315    /**
1316     * A wrapper class to associate a stanza filter with a listener.
1317     */
1318    protected static class ListenerWrapper {
1319
1320        private final StanzaListener packetListener;
1321        private final StanzaFilter packetFilter;
1322
1323        /**
1324         * Create a class which associates a stanza filter with a listener.
1325         *
1326         * @param packetListener the stanza listener.
1327         * @param packetFilter the associated filter or null if it listen for all packets.
1328         */
1329        public ListenerWrapper(StanzaListener packetListener, StanzaFilter packetFilter) {
1330            this.packetListener = packetListener;
1331            this.packetFilter = packetFilter;
1332        }
1333
1334        public boolean filterMatches(Stanza packet) {
1335            return packetFilter == null || packetFilter.accept(packet);
1336        }
1337
1338        public StanzaListener getListener() {
1339            return packetListener;
1340        }
1341    }
1342
1343    /**
1344     * A wrapper class to associate a stanza filter with an interceptor.
1345     */
1346    protected static class InterceptorWrapper {
1347
1348        private final StanzaListener packetInterceptor;
1349        private final StanzaFilter packetFilter;
1350
1351        /**
1352         * Create a class which associates a stanza filter with an interceptor.
1353         *
1354         * @param packetInterceptor the interceptor.
1355         * @param packetFilter the associated filter or null if it intercepts all packets.
1356         */
1357        public InterceptorWrapper(StanzaListener packetInterceptor, StanzaFilter packetFilter) {
1358            this.packetInterceptor = packetInterceptor;
1359            this.packetFilter = packetFilter;
1360        }
1361
1362        public boolean filterMatches(Stanza packet) {
1363            return packetFilter == null || packetFilter.accept(packet);
1364        }
1365
1366        public StanzaListener getInterceptor() {
1367            return packetInterceptor;
1368        }
1369    }
1370
1371    @Override
1372    public int getConnectionCounter() {
1373        return connectionCounterValue;
1374    }
1375
1376    @Override
1377    public void setFromMode(FromMode fromMode) {
1378        this.fromMode = fromMode;
1379    }
1380
1381    @Override
1382    public FromMode getFromMode() {
1383        return this.fromMode;
1384    }
1385
1386    protected final void parseFeatures(XmlPullParser parser) throws Exception {
1387        streamFeatures.clear();
1388        final int initialDepth = parser.getDepth();
1389        while (true) {
1390            int eventType = parser.next();
1391
1392            if (eventType == XmlPullParser.START_TAG && parser.getDepth() == initialDepth + 1) {
1393                ExtensionElement streamFeature = null;
1394                String name = parser.getName();
1395                String namespace = parser.getNamespace();
1396                switch (name) {
1397                case StartTls.ELEMENT:
1398                    streamFeature = PacketParserUtils.parseStartTlsFeature(parser);
1399                    break;
1400                case Mechanisms.ELEMENT:
1401                    streamFeature = new Mechanisms(PacketParserUtils.parseMechanisms(parser));
1402                    break;
1403                case Bind.ELEMENT:
1404                    streamFeature = Bind.Feature.INSTANCE;
1405                    break;
1406                case Session.ELEMENT:
1407                    streamFeature = PacketParserUtils.parseSessionFeature(parser);
1408                    break;
1409                case Compress.Feature.ELEMENT:
1410                    streamFeature = PacketParserUtils.parseCompressionFeature(parser);
1411                    break;
1412                default:
1413                    ExtensionElementProvider<ExtensionElement> provider = ProviderManager.getStreamFeatureProvider(name, namespace);
1414                    if (provider != null) {
1415                        streamFeature = provider.parse(parser);
1416                    }
1417                    break;
1418                }
1419                if (streamFeature != null) {
1420                    addStreamFeature(streamFeature);
1421                }
1422            }
1423            else if (eventType == XmlPullParser.END_TAG && parser.getDepth() == initialDepth) {
1424                break;
1425            }
1426        }
1427
1428        if (hasFeature(Mechanisms.ELEMENT, Mechanisms.NAMESPACE)) {
1429            // Only proceed with SASL auth if TLS is disabled or if the server doesn't announce it
1430            if (!hasFeature(StartTls.ELEMENT, StartTls.NAMESPACE)
1431                            || config.getSecurityMode() == SecurityMode.disabled) {
1432                tlsHandled.reportSuccess();
1433                saslFeatureReceived.reportSuccess();
1434            }
1435        }
1436
1437        // If the server reported the bind feature then we are that that we did SASL and maybe
1438        // STARTTLS. We can then report that the last 'stream:features' have been parsed
1439        if (hasFeature(Bind.ELEMENT, Bind.NAMESPACE)) {
1440            if (!hasFeature(Compress.Feature.ELEMENT, Compress.NAMESPACE)
1441                            || !config.isCompressionEnabled()) {
1442                // This was was last features from the server is either it did not contain
1443                // compression or if we disabled it
1444                lastFeaturesReceived.reportSuccess();
1445            }
1446        }
1447        afterFeaturesReceived();
1448    }
1449
1450    @SuppressWarnings("unused")
1451    protected void afterFeaturesReceived() throws SecurityRequiredException, NotConnectedException, InterruptedException {
1452        // Default implementation does nothing
1453    }
1454
1455    @SuppressWarnings("unchecked")
1456    @Override
1457    public <F extends ExtensionElement> F getFeature(String element, String namespace) {
1458        return (F) streamFeatures.get(XmppStringUtils.generateKey(element, namespace));
1459    }
1460
1461    @Override
1462    public boolean hasFeature(String element, String namespace) {
1463        return getFeature(element, namespace) != null;
1464    }
1465
1466    protected void addStreamFeature(ExtensionElement feature) {
1467        String key = XmppStringUtils.generateKey(feature.getElementName(), feature.getNamespace());
1468        streamFeatures.put(key, feature);
1469    }
1470
1471    @SuppressWarnings("deprecation")
1472    @Override
1473    public void sendStanzaWithResponseCallback(Stanza stanza, StanzaFilter replyFilter,
1474                    StanzaListener callback) throws NotConnectedException, InterruptedException {
1475        sendStanzaWithResponseCallback(stanza, replyFilter, callback, null);
1476    }
1477
1478    @SuppressWarnings("deprecation")
1479    @Override
1480    public void sendStanzaWithResponseCallback(Stanza stanza, StanzaFilter replyFilter,
1481                    StanzaListener callback, ExceptionCallback exceptionCallback)
1482                    throws NotConnectedException, InterruptedException {
1483        sendStanzaWithResponseCallback(stanza, replyFilter, callback, exceptionCallback,
1484                        getReplyTimeout());
1485    }
1486
1487    @Override
1488    public SmackFuture<IQ, Exception> sendIqRequestAsync(IQ request) {
1489        return sendIqRequestAsync(request, getReplyTimeout());
1490    }
1491
1492    @Override
1493    public SmackFuture<IQ, Exception> sendIqRequestAsync(IQ request, long timeout) {
1494        StanzaFilter replyFilter = new IQReplyFilter(request, this);
1495        return sendAsync(request, replyFilter, timeout);
1496    }
1497
1498    @Override
1499    public <S extends Stanza> SmackFuture<S, Exception> sendAsync(S stanza, final StanzaFilter replyFilter) {
1500        return sendAsync(stanza, replyFilter, getReplyTimeout());
1501    }
1502
1503    @SuppressWarnings("FutureReturnValueIgnored")
1504    @Override
1505    public <S extends Stanza> SmackFuture<S, Exception> sendAsync(S stanza, final StanzaFilter replyFilter, long timeout) {
1506        Objects.requireNonNull(stanza, "stanza must not be null");
1507        // While Smack allows to add PacketListeners with a PacketFilter value of 'null', we
1508        // disallow it here in the async API as it makes no sense
1509        Objects.requireNonNull(replyFilter, "replyFilter must not be null");
1510
1511        final InternalSmackFuture<S, Exception> future = new InternalSmackFuture<>();
1512
1513        final StanzaListener stanzaListener = new StanzaListener() {
1514            @Override
1515            public void processStanza(Stanza stanza) throws NotConnectedException, InterruptedException {
1516                boolean removed = removeAsyncStanzaListener(this);
1517                if (!removed) {
1518                    // We lost a race against the "no response" handling runnable. Avoid calling the callback, as the
1519                    // exception callback will be invoked (if any).
1520                    return;
1521                }
1522                try {
1523                    XMPPErrorException.ifHasErrorThenThrow(stanza);
1524                    @SuppressWarnings("unchecked")
1525                    S s = (S) stanza;
1526                    future.setResult(s);
1527                }
1528                catch (XMPPErrorException exception) {
1529                    future.setException(exception);
1530                }
1531            }
1532        };
1533        schedule(new Runnable() {
1534            @Override
1535            public void run() {
1536                boolean removed = removeAsyncStanzaListener(stanzaListener);
1537                if (!removed) {
1538                    // We lost a race against the stanza listener, he already removed itself because he received a
1539                    // reply. There is nothing more to do here.
1540                    return;
1541                }
1542
1543                // If the packetListener got removed, then it was never run and
1544                // we never received a response, inform the exception callback
1545                Exception exception;
1546                if (!isConnected()) {
1547                    // If the connection is no longer connected, throw a not connected exception.
1548                    exception = new NotConnectedException(AbstractXMPPConnection.this, replyFilter);
1549                }
1550                else {
1551                    exception = NoResponseException.newWith(AbstractXMPPConnection.this, replyFilter);
1552                }
1553                future.setException(exception);
1554            }
1555        }, timeout, TimeUnit.MILLISECONDS);
1556
1557        addAsyncStanzaListener(stanzaListener, replyFilter);
1558        try {
1559            sendStanza(stanza);
1560        }
1561        catch (NotConnectedException | InterruptedException exception) {
1562            future.setException(exception);
1563        }
1564
1565        return future;
1566    }
1567
1568    @SuppressWarnings({ "FutureReturnValueIgnored", "deprecation" })
1569    @Override
1570    public void sendStanzaWithResponseCallback(Stanza stanza, final StanzaFilter replyFilter,
1571                    final StanzaListener callback, final ExceptionCallback exceptionCallback,
1572                    long timeout) throws NotConnectedException, InterruptedException {
1573        Objects.requireNonNull(stanza, "stanza must not be null");
1574        // While Smack allows to add PacketListeners with a PacketFilter value of 'null', we
1575        // disallow it here in the async API as it makes no sense
1576        Objects.requireNonNull(replyFilter, "replyFilter must not be null");
1577        Objects.requireNonNull(callback, "callback must not be null");
1578
1579        final StanzaListener packetListener = new StanzaListener() {
1580            @Override
1581            public void processStanza(Stanza packet) throws NotConnectedException, InterruptedException, NotLoggedInException {
1582                boolean removed = removeAsyncStanzaListener(this);
1583                if (!removed) {
1584                    // We lost a race against the "no response" handling runnable. Avoid calling the callback, as the
1585                    // exception callback will be invoked (if any).
1586                    return;
1587                }
1588                try {
1589                    XMPPErrorException.ifHasErrorThenThrow(packet);
1590                    callback.processStanza(packet);
1591                }
1592                catch (XMPPErrorException e) {
1593                    if (exceptionCallback != null) {
1594                        exceptionCallback.processException(e);
1595                    }
1596                }
1597            }
1598        };
1599        schedule(new Runnable() {
1600            @Override
1601            public void run() {
1602                boolean removed = removeAsyncStanzaListener(packetListener);
1603                // If the packetListener got removed, then it was never run and
1604                // we never received a response, inform the exception callback
1605                if (removed && exceptionCallback != null) {
1606                    Exception exception;
1607                    if (!isConnected()) {
1608                        // If the connection is no longer connected, throw a not connected exception.
1609                        exception = new NotConnectedException(AbstractXMPPConnection.this, replyFilter);
1610                    } else {
1611                        exception = NoResponseException.newWith(AbstractXMPPConnection.this, replyFilter);
1612                    }
1613                    final Exception exceptionToProcess = exception;
1614                    Async.go(new Runnable() {
1615                        @Override
1616                        public void run() {
1617                            exceptionCallback.processException(exceptionToProcess);
1618                        }
1619                    });
1620                }
1621            }
1622        }, timeout, TimeUnit.MILLISECONDS);
1623        addAsyncStanzaListener(packetListener, replyFilter);
1624        sendStanza(stanza);
1625    }
1626
1627    @SuppressWarnings("deprecation")
1628    @Override
1629    public void sendIqWithResponseCallback(IQ iqRequest, StanzaListener callback)
1630                    throws NotConnectedException, InterruptedException {
1631        sendIqWithResponseCallback(iqRequest, callback, null);
1632    }
1633
1634    @SuppressWarnings("deprecation")
1635    @Override
1636    public void sendIqWithResponseCallback(IQ iqRequest, StanzaListener callback,
1637                    ExceptionCallback exceptionCallback) throws NotConnectedException, InterruptedException {
1638        sendIqWithResponseCallback(iqRequest, callback, exceptionCallback, getReplyTimeout());
1639    }
1640
1641    @SuppressWarnings("deprecation")
1642    @Override
1643    public void sendIqWithResponseCallback(IQ iqRequest, final StanzaListener callback,
1644                    final ExceptionCallback exceptionCallback, long timeout)
1645                    throws NotConnectedException, InterruptedException {
1646        StanzaFilter replyFilter = new IQReplyFilter(iqRequest, this);
1647        sendStanzaWithResponseCallback(iqRequest, replyFilter, callback, exceptionCallback, timeout);
1648    }
1649
1650    @SuppressWarnings("FutureReturnValueIgnored")
1651    @Override
1652    public void addOneTimeSyncCallback(final StanzaListener callback, final StanzaFilter packetFilter) {
1653        final StanzaListener packetListener = new StanzaListener() {
1654            @Override
1655            public void processStanza(Stanza packet) throws NotConnectedException, InterruptedException, NotLoggedInException {
1656                try {
1657                    callback.processStanza(packet);
1658                } finally {
1659                    removeSyncStanzaListener(this);
1660                }
1661            }
1662        };
1663        addSyncStanzaListener(packetListener, packetFilter);
1664        schedule(new Runnable() {
1665            @Override
1666            public void run() {
1667                removeSyncStanzaListener(packetListener);
1668            }
1669        }, getReplyTimeout(), TimeUnit.MILLISECONDS);
1670    }
1671
1672    @Override
1673    public IQRequestHandler registerIQRequestHandler(final IQRequestHandler iqRequestHandler) {
1674        final String key = XmppStringUtils.generateKey(iqRequestHandler.getElement(), iqRequestHandler.getNamespace());
1675        switch (iqRequestHandler.getType()) {
1676        case set:
1677            synchronized (setIqRequestHandler) {
1678                return setIqRequestHandler.put(key, iqRequestHandler);
1679            }
1680        case get:
1681            synchronized (getIqRequestHandler) {
1682                return getIqRequestHandler.put(key, iqRequestHandler);
1683            }
1684        default:
1685            throw new IllegalArgumentException("Only IQ type of 'get' and 'set' allowed");
1686        }
1687    }
1688
1689    @Override
1690    public final IQRequestHandler unregisterIQRequestHandler(IQRequestHandler iqRequestHandler) {
1691        return unregisterIQRequestHandler(iqRequestHandler.getElement(), iqRequestHandler.getNamespace(),
1692                        iqRequestHandler.getType());
1693    }
1694
1695    @Override
1696    public IQRequestHandler unregisterIQRequestHandler(String element, String namespace, IQ.Type type) {
1697        final String key = XmppStringUtils.generateKey(element, namespace);
1698        switch (type) {
1699        case set:
1700            synchronized (setIqRequestHandler) {
1701                return setIqRequestHandler.remove(key);
1702            }
1703        case get:
1704            synchronized (getIqRequestHandler) {
1705                return getIqRequestHandler.remove(key);
1706            }
1707        default:
1708            throw new IllegalArgumentException("Only IQ type of 'get' and 'set' allowed");
1709        }
1710    }
1711
1712    private long lastStanzaReceived;
1713
1714    @Override
1715    public long getLastStanzaReceived() {
1716        return lastStanzaReceived;
1717    }
1718
1719    /**
1720     * Install a parsing exception callback, which will be invoked once an exception is encountered while parsing a
1721     * stanza.
1722     *
1723     * @param callback the callback to install
1724     */
1725    public void setParsingExceptionCallback(ParsingExceptionCallback callback) {
1726        parsingExceptionCallback = callback;
1727    }
1728
1729    /**
1730     * Get the current active parsing exception callback.
1731     *
1732     * @return the active exception callback or null if there is none
1733     */
1734    public ParsingExceptionCallback getParsingExceptionCallback() {
1735        return parsingExceptionCallback;
1736    }
1737
1738    @Override
1739    public final String toString() {
1740        EntityFullJid localEndpoint = getUser();
1741        String localEndpointString = (localEndpoint == null ?  "not-authenticated" : localEndpoint.toString());
1742        return getClass().getSimpleName() + '[' + localEndpointString + "] (" + getConnectionCounter() + ')';
1743    }
1744
1745    protected static void asyncGo(Runnable runnable) {
1746        CACHED_EXECUTOR_SERVICE.execute(runnable);
1747    }
1748
1749    protected static ScheduledFuture<?> schedule(Runnable runnable, long delay, TimeUnit unit) {
1750        return SCHEDULED_EXECUTOR_SERVICE.schedule(runnable, delay, unit);
1751    }
1752}