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