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