mirror of
https://codeberg.org/Mercury-IM/Smack
synced 2024-12-22 20:47:57 +01:00
Typo in XMPPTCPConnection s/ResumptiodDefault/ResumptionDefault/
This commit is contained in:
Florian Schmaus
parent
15c1c8ad19
commit
7ceb5f09df
1 changed files with 179 additions and 160 deletions
|
|
@ -20,7 +20,6 @@ import org.jivesoftware.smack.AbstractConnectionListener;
|
|||
import org.jivesoftware.smack.AbstractXMPPConnection;
|
||||
import org.jivesoftware.smack.ConnectionConfiguration;
|
||||
import org.jivesoftware.smack.ConnectionConfiguration.SecurityMode;
|
||||
import org.jivesoftware.smack.ConnectionCreationListener;
|
||||
import org.jivesoftware.smack.StanzaListener;
|
||||
import org.jivesoftware.smack.SmackConfiguration;
|
||||
import org.jivesoftware.smack.SmackException;
|
||||
|
|
@ -68,14 +67,19 @@ import org.jivesoftware.smack.sm.packet.StreamManagement.Resumed;
|
|||
import org.jivesoftware.smack.sm.packet.StreamManagement.StreamManagementFeature;
|
||||
import org.jivesoftware.smack.sm.predicates.Predicate;
|
||||
import org.jivesoftware.smack.sm.provider.ParseStreamManagement;
|
||||
import org.jivesoftware.smack.packet.PlainStreamElement;
|
||||
import org.jivesoftware.smack.packet.Nonza;
|
||||
import org.jivesoftware.smack.packet.XMPPError;
|
||||
import org.jivesoftware.smack.proxy.ProxyInfo;
|
||||
import org.jivesoftware.smack.util.ArrayBlockingQueueWithShutdown;
|
||||
import org.jivesoftware.smack.util.Async;
|
||||
import org.jivesoftware.smack.util.PacketParserUtils;
|
||||
import org.jivesoftware.smack.util.StringUtils;
|
||||
import org.jivesoftware.smack.util.TLSUtils;
|
||||
import org.jivesoftware.smack.util.XmlStringBuilder;
|
||||
import org.jivesoftware.smack.util.dns.HostAddress;
|
||||
import org.jxmpp.jid.impl.JidCreate;
|
||||
import org.jxmpp.jid.parts.Resourcepart;
|
||||
import org.jxmpp.stringprep.XmppStringprepException;
|
||||
import org.jxmpp.util.XmppStringUtils;
|
||||
import org.xmlpull.v1.XmlPullParser;
|
||||
import org.xmlpull.v1.XmlPullParserException;
|
||||
|
|
@ -152,14 +156,6 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
*/
|
||||
private boolean disconnectedButResumeable = false;
|
||||
|
||||
/**
|
||||
* Flag to indicate if the socket was closed intentionally by Smack.
|
||||
* <p>
|
||||
* This boolean flag is used concurrently, therefore it is marked volatile.
|
||||
* </p>
|
||||
*/
|
||||
private volatile boolean socketClosed = false;
|
||||
|
||||
private boolean usingTLS = false;
|
||||
|
||||
/**
|
||||
|
|
@ -172,19 +168,27 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
*/
|
||||
protected PacketReader packetReader;
|
||||
|
||||
private final SynchronizationPoint<Exception> initalOpenStreamSend = new SynchronizationPoint<Exception>(this);
|
||||
private final SynchronizationPoint<Exception> initalOpenStreamSend = new SynchronizationPoint<>(
|
||||
this, "initial open stream element send to server");
|
||||
|
||||
/**
|
||||
*
|
||||
*/
|
||||
private final SynchronizationPoint<XMPPException> maybeCompressFeaturesReceived = new SynchronizationPoint<XMPPException>(
|
||||
this);
|
||||
this, "stream compression feature");
|
||||
|
||||
/**
|
||||
*
|
||||
*/
|
||||
private final SynchronizationPoint<XMPPException> compressSyncPoint = new SynchronizationPoint<XMPPException>(
|
||||
this);
|
||||
this, "stream compression");
|
||||
|
||||
/**
|
||||
* A synchronization point which is successful if this connection has received the closing
|
||||
* stream element from the remote end-point, i.e. the server.
|
||||
*/
|
||||
private final SynchronizationPoint<Exception> closingStreamReceived = new SynchronizationPoint<>(
|
||||
this, "stream closing element received");
|
||||
|
||||
/**
|
||||
* The default bundle and defer callback, used for new connections.
|
||||
|
|
@ -213,10 +217,10 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
private String smSessionId;
|
||||
|
||||
private final SynchronizationPoint<XMPPException> smResumedSyncPoint = new SynchronizationPoint<XMPPException>(
|
||||
this);
|
||||
this, "stream resumed element");
|
||||
|
||||
private final SynchronizationPoint<XMPPException> smEnabledSyncPoint = new SynchronizationPoint<XMPPException>(
|
||||
this);
|
||||
this, "stream enabled element");
|
||||
|
||||
/**
|
||||
* The client's preferred maximum resumption time in seconds.
|
||||
|
|
@ -316,8 +320,9 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
*
|
||||
* @param jid the bare JID used by the client.
|
||||
* @param password the password or authentication token.
|
||||
* @throws XmppStringprepException
|
||||
*/
|
||||
public XMPPTCPConnection(CharSequence jid, String password) {
|
||||
public XMPPTCPConnection(CharSequence jid, String password) throws XmppStringprepException {
|
||||
this(XmppStringUtils.parseLocalpart(jid.toString()), password, XmppStringUtils.parseDomain(jid.toString()));
|
||||
}
|
||||
|
||||
|
|
@ -331,10 +336,11 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
* @param username
|
||||
* @param password
|
||||
* @param serviceName
|
||||
* @throws XmppStringprepException
|
||||
*/
|
||||
public XMPPTCPConnection(CharSequence username, String password, String serviceName) {
|
||||
this(XMPPTCPConnectionConfiguration.builder().setUsernameAndPassword(username, password).setServiceName(
|
||||
serviceName).build());
|
||||
public XMPPTCPConnection(CharSequence username, String password, String serviceName) throws XmppStringprepException {
|
||||
this(XMPPTCPConnectionConfiguration.builder().setUsernameAndPassword(username, password).setXmppDomain(
|
||||
JidCreate.domainBareFrom(serviceName)).build());
|
||||
}
|
||||
|
||||
@Override
|
||||
|
|
@ -360,31 +366,21 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
}
|
||||
|
||||
@Override
|
||||
protected void afterSuccessfulLogin(final boolean resumed) throws NotConnectedException {
|
||||
protected void afterSuccessfulLogin(final boolean resumed) throws NotConnectedException, InterruptedException {
|
||||
// Reset the flag in case it was set
|
||||
disconnectedButResumeable = false;
|
||||
super.afterSuccessfulLogin(resumed);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected synchronized void loginNonAnonymously(String username, String password, String resource) throws XMPPException, SmackException, IOException {
|
||||
if (saslAuthentication.hasNonAnonymousAuthentication()) {
|
||||
// Authenticate using SASL
|
||||
if (password != null) {
|
||||
saslAuthentication.authenticate(username, password, resource);
|
||||
}
|
||||
else {
|
||||
saslAuthentication.authenticate(resource, config.getCallbackHandler());
|
||||
}
|
||||
} else {
|
||||
throw new SmackException("No non-anonymous SASL authentication mechanism available");
|
||||
}
|
||||
protected synchronized void loginInternal(String username, String password, Resourcepart resource) throws XMPPException,
|
||||
SmackException, IOException, InterruptedException {
|
||||
// Authenticate using SASL
|
||||
saslAuthentication.authenticate(username, password, config.getAuthzid());
|
||||
|
||||
// If compression is enabled then request the server to use stream compression. XEP-170
|
||||
// recommends to perform stream compression before resource binding.
|
||||
if (config.isCompressionEnabled()) {
|
||||
useCompression();
|
||||
}
|
||||
maybeEnableCompression();
|
||||
|
||||
if (isSmResumptionPossible()) {
|
||||
smResumedSyncPoint.sendAndWaitForResponse(new Resume(clientHandledStanzasCount, smSessionId));
|
||||
|
|
@ -436,37 +432,11 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
afterSuccessfulLogin(false);
|
||||
}
|
||||
|
||||
@Override
|
||||
public synchronized void loginAnonymously() throws XMPPException, SmackException, IOException {
|
||||
// Wait with SASL auth until the SASL mechanisms have been received
|
||||
saslFeatureReceived.checkIfSuccessOrWaitOrThrow();
|
||||
|
||||
if (saslAuthentication.hasAnonymousAuthentication()) {
|
||||
saslAuthentication.authenticateAnonymously();
|
||||
}
|
||||
else {
|
||||
throw new SmackException("No anonymous SASL authentication mechanism available");
|
||||
}
|
||||
|
||||
// If compression is enabled then request the server to use stream compression
|
||||
if (config.isCompressionEnabled()) {
|
||||
useCompression();
|
||||
}
|
||||
|
||||
bindResourceAndEstablishSession(null);
|
||||
|
||||
afterSuccessfulLogin(false);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isSecureConnection() {
|
||||
return usingTLS;
|
||||
}
|
||||
|
||||
public boolean isSocketClosed() {
|
||||
return socketClosed;
|
||||
}
|
||||
|
||||
/**
|
||||
* Shuts the current connection down. After this method returns, the connection must be ready
|
||||
* for re-use by connect.
|
||||
|
|
@ -478,7 +448,7 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
// Try to send a last SM Acknowledgement. Most servers won't find this information helpful, as the SM
|
||||
// state is dropped after a clean disconnect anyways. OTOH it doesn't hurt much either.
|
||||
sendSmAcknowledgementInternal();
|
||||
} catch (NotConnectedException e) {
|
||||
} catch (InterruptedException | NotConnectedException e) {
|
||||
LOGGER.log(Level.FINE, "Can not send final SM ack as connection is not connected", e);
|
||||
}
|
||||
}
|
||||
|
|
@ -496,18 +466,27 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
if (disconnectedButResumeable) {
|
||||
return;
|
||||
}
|
||||
|
||||
// First shutdown the writer, this will result in a closing stream element getting send to
|
||||
// the server
|
||||
if (packetWriter != null) {
|
||||
packetWriter.shutdown(instant);
|
||||
}
|
||||
|
||||
try {
|
||||
// After we send the closing stream element, check if there was already a
|
||||
// closing stream element sent by the server or wait with a timeout for a
|
||||
// closing stream element to be received from the server.
|
||||
Exception res = closingStreamReceived.checkIfSuccessOrWait();
|
||||
LOGGER.info("closingstream " + res);
|
||||
} catch (InterruptedException | NoResponseException e) {
|
||||
LOGGER.log(Level.INFO, "Exception while waiting for closing stream element from the server " + this, e);
|
||||
}
|
||||
|
||||
if (packetReader != null) {
|
||||
packetReader.shutdown();
|
||||
}
|
||||
if (packetWriter != null) {
|
||||
packetWriter.shutdown(instant);
|
||||
}
|
||||
|
||||
// Set socketClosed to true. This will cause the PacketReader
|
||||
// and PacketWriter to ignore any Exceptions that are thrown
|
||||
// because of a read/write from/to a closed stream.
|
||||
// It is *important* that this is done before socket.close()!
|
||||
socketClosed = true;
|
||||
try {
|
||||
socket.close();
|
||||
} catch (Exception e) {
|
||||
|
|
@ -540,12 +519,12 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
}
|
||||
|
||||
@Override
|
||||
public void send(PlainStreamElement element) throws NotConnectedException {
|
||||
public void sendNonza(Nonza element) throws NotConnectedException, InterruptedException {
|
||||
packetWriter.sendStreamElement(element);
|
||||
}
|
||||
|
||||
@Override
|
||||
protected void sendStanzaInternal(Stanza packet) throws NotConnectedException {
|
||||
protected void sendStanzaInternal(Stanza packet) throws NotConnectedException, InterruptedException {
|
||||
packetWriter.sendStreamElement(packet);
|
||||
if (isSmEnabled()) {
|
||||
for (StanzaFilter requestAckPredicate : requestAckPredicates) {
|
||||
|
|
@ -560,6 +539,8 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
private void connectUsingConfiguration() throws IOException, ConnectionException {
|
||||
List<HostAddress> failedAddresses = populateHostAddresses();
|
||||
SocketFactory socketFactory = config.getSocketFactory();
|
||||
ProxyInfo proxyInfo = config.getProxyInfo();
|
||||
int timeout = config.getConnectTimeout();
|
||||
if (socketFactory == null) {
|
||||
socketFactory = SocketFactory.getDefault();
|
||||
}
|
||||
|
|
@ -579,7 +560,12 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
final String inetAddressAndPort = inetAddress + " at port " + port;
|
||||
LOGGER.finer("Trying to establish TCP connection to " + inetAddressAndPort);
|
||||
try {
|
||||
socket.connect(new InetSocketAddress(inetAddress, port), config.getConnectTimeout());
|
||||
if (proxyInfo == null) {
|
||||
socket.connect(new InetSocketAddress(inetAddress, port), timeout);
|
||||
}
|
||||
else {
|
||||
proxyInfo.getProxySocketConnection().connect(socket, inetAddress, port, timeout);
|
||||
}
|
||||
} catch (Exception e) {
|
||||
if (inetAddresses.hasNext()) {
|
||||
continue innerloop;
|
||||
|
|
@ -638,13 +624,6 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
// Start the packet reader. The startup() method will block until we
|
||||
// get an opening stream packet back from server
|
||||
packetReader.init();
|
||||
|
||||
if (isFirstInitialization) {
|
||||
// Notify listeners that a new connection has been established
|
||||
for (ConnectionCreationListener listener : getConnectionCreationListeners()) {
|
||||
listener.connectionCreated(this);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private void initReaderAndWriter() throws IOException {
|
||||
|
|
@ -682,14 +661,8 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
KeyManager[] kms = null;
|
||||
PasswordCallback pcb = null;
|
||||
|
||||
if(config.getCallbackHandler() == null) {
|
||||
ks = null;
|
||||
} else if (context == null) {
|
||||
if(config.getKeystoreType().equals("NONE")) {
|
||||
ks = null;
|
||||
pcb = null;
|
||||
}
|
||||
else if(config.getKeystoreType().equals("PKCS11")) {
|
||||
if (context == null) {
|
||||
if(config.getKeystoreType().equals("PKCS11")) {
|
||||
try {
|
||||
Constructor<?> c = Class.forName("sun.security.pkcs11.SunPKCS11").getConstructor(InputStream.class);
|
||||
String pkcs11Config = "name = SmartCard\nlibrary = "+config.getPKCS11Library();
|
||||
|
|
@ -759,8 +732,8 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
final HostnameVerifier verifier = getConfiguration().getHostnameVerifier();
|
||||
if (verifier == null) {
|
||||
throw new IllegalStateException("No HostnameVerifier set. Use connectionConfiguration.setHostnameVerifier() to configure.");
|
||||
} else if (!verifier.verify(getServiceName(), sslSocket.getSession())) {
|
||||
throw new CertificateException("Hostname verification of certificate failed. Certificate does not authenticate " + getServiceName());
|
||||
} else if (!verifier.verify(getXMPPServiceDomain().toString(), sslSocket.getSession())) {
|
||||
throw new CertificateException("Hostname verification of certificate failed. Certificate does not authenticate " + getXMPPServiceDomain());
|
||||
}
|
||||
|
||||
// Set that TLS was successful
|
||||
|
|
@ -773,12 +746,7 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
* @return a instance of XMPPInputOutputStream or null if no suitable instance was found
|
||||
*
|
||||
*/
|
||||
private XMPPInputOutputStream maybeGetCompressionHandler() {
|
||||
Compress.Feature compression = getFeature(Compress.Feature.ELEMENT, Compress.NAMESPACE);
|
||||
if (compression == null) {
|
||||
// Server does not support compression
|
||||
return null;
|
||||
}
|
||||
private static XMPPInputOutputStream maybeGetCompressionHandler(Compress.Feature compression) {
|
||||
for (XMPPInputOutputStream handler : SmackConfiguration.getCompresionHandlers()) {
|
||||
String method = handler.getCompressionMethod();
|
||||
if (compression.getMethods().contains(method))
|
||||
|
|
@ -808,12 +776,21 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
* @throws NotConnectedException
|
||||
* @throws XMPPException
|
||||
* @throws NoResponseException
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
private void useCompression() throws NotConnectedException, NoResponseException, XMPPException {
|
||||
private void maybeEnableCompression() throws NotConnectedException, NoResponseException, XMPPException, InterruptedException {
|
||||
if (!config.isCompressionEnabled()) {
|
||||
return;
|
||||
}
|
||||
maybeCompressFeaturesReceived.checkIfSuccessOrWait();
|
||||
Compress.Feature compression = getFeature(Compress.Feature.ELEMENT, Compress.NAMESPACE);
|
||||
if (compression == null) {
|
||||
// Server does not support compression
|
||||
return;
|
||||
}
|
||||
// If stream compression was offered by the server and we want to use
|
||||
// compression then send compression request to the server
|
||||
if ((compressionHandler = maybeGetCompressionHandler()) != null) {
|
||||
if ((compressionHandler = maybeGetCompressionHandler(compression)) != null) {
|
||||
compressSyncPoint.sendAndWaitForResponseOrThrow(new Compress(compressionHandler.getCompressionMethod()));
|
||||
} else {
|
||||
LOGGER.warning("Could not enable compression because no matching handler/method pair was found");
|
||||
|
|
@ -821,25 +798,26 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
}
|
||||
|
||||
/**
|
||||
* Establishes a connection to the XMPP server and performs an automatic login
|
||||
* only if the previous connection state was logged (authenticated). It basically
|
||||
* creates and maintains a socket connection to the server.<p>
|
||||
* <p/>
|
||||
* Establishes a connection to the XMPP server. It basically
|
||||
* creates and maintains a socket connection to the server.
|
||||
* <p>
|
||||
* Listeners will be preserved from a previous connection if the reconnection
|
||||
* occurs after an abrupt termination.
|
||||
* </p>
|
||||
*
|
||||
* @throws XMPPException if an error occurs while trying to establish the connection.
|
||||
* @throws SmackException
|
||||
* @throws IOException
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
@Override
|
||||
protected void connectInternal() throws SmackException, IOException, XMPPException {
|
||||
protected void connectInternal() throws SmackException, IOException, XMPPException, InterruptedException {
|
||||
closingStreamReceived.init();
|
||||
// Establishes the TCP connection to the server and does setup the reader and writer. Throws an exception if
|
||||
// there is an error establishing the connection
|
||||
connectUsingConfiguration();
|
||||
|
||||
// We connected successfully to the servers TCP port
|
||||
socketClosed = false;
|
||||
initConnection();
|
||||
|
||||
// Wait with SASL auth until the SASL mechanisms have been received
|
||||
|
|
@ -848,13 +826,6 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
// Make note of the fact that we're now connected.
|
||||
connected = true;
|
||||
callConnectionConnectedListener();
|
||||
|
||||
// Automatically makes the login if the user was previously connected successfully
|
||||
// to the server and the connection was terminated abruptly
|
||||
if (wasAuthenticated) {
|
||||
login();
|
||||
notifyReconnection();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -885,7 +856,7 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
}
|
||||
|
||||
@Override
|
||||
protected void afterFeaturesReceived() throws SecurityRequiredException, NotConnectedException {
|
||||
protected void afterFeaturesReceived() throws SecurityRequiredException, NotConnectedException, InterruptedException {
|
||||
StartTls startTlsFeature = getFeature(StartTls.ELEMENT, StartTls.NAMESPACE);
|
||||
if (startTlsFeature != null) {
|
||||
if (startTlsFeature.required() && config.getSecurityMode() == SecurityMode.disabled) {
|
||||
|
|
@ -894,7 +865,7 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
}
|
||||
|
||||
if (config.getSecurityMode() != ConnectionConfiguration.SecurityMode.disabled) {
|
||||
send(new StartTls());
|
||||
sendNonza(new StartTls());
|
||||
}
|
||||
}
|
||||
// If TLS is required but the server doesn't offer it, disconnect
|
||||
|
|
@ -919,20 +890,21 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
* to be sent by the server.
|
||||
*
|
||||
* @throws SmackException if the parser could not be reset.
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
void openStream() throws SmackException {
|
||||
void openStream() throws SmackException, InterruptedException {
|
||||
// If possible, provide the receiving entity of the stream open tag, i.e. the server, as much information as
|
||||
// possible. The 'to' attribute is *always* available. The 'from' attribute if set by the user and no external
|
||||
// mechanism is used to determine the local entity (user). And the 'id' attribute is available after the first
|
||||
// response from the server (see e.g. RFC 6120 § 9.1.1 Step 2.)
|
||||
CharSequence to = getServiceName();
|
||||
CharSequence to = getXMPPServiceDomain();
|
||||
CharSequence from = null;
|
||||
CharSequence localpart = config.getUsername();
|
||||
if (localpart != null) {
|
||||
from = XmppStringUtils.completeJidFrom(localpart, to);
|
||||
}
|
||||
String id = getStreamId();
|
||||
send(new StreamOpen(to, from, id));
|
||||
sendNonza(new StreamOpen(to, from, id));
|
||||
try {
|
||||
packetReader.parser = PacketParserUtils.newXmppParser(reader);
|
||||
}
|
||||
|
|
@ -995,8 +967,8 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
// We found an opening stream.
|
||||
if ("jabber:client".equals(parser.getNamespace(null))) {
|
||||
streamId = parser.getAttributeValue("", "id");
|
||||
String reportedServiceName = parser.getAttributeValue("", "from");
|
||||
assert(reportedServiceName.equals(config.getServiceName()));
|
||||
String reportedServerDomain = parser.getAttributeValue("", "from");
|
||||
assert(config.getXMPPServiceDomain().equals(reportedServerDomain));
|
||||
}
|
||||
break;
|
||||
case "error":
|
||||
|
|
@ -1151,8 +1123,33 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
break;
|
||||
case XmlPullParser.END_TAG:
|
||||
if (parser.getName().equals("stream")) {
|
||||
// Disconnect the connection
|
||||
disconnect();
|
||||
if (!parser.getNamespace().equals("http://etherx.jabber.org/streams")) {
|
||||
LOGGER.warning(XMPPTCPConnection.this + " </stream> but different namespace " + parser.getNamespace());
|
||||
break;
|
||||
}
|
||||
|
||||
// Check if the queue was already shut down before reporting success on closing stream tag
|
||||
// received. This avoids a race if there is a disconnect(), followed by a connect(), which
|
||||
// did re-start the queue again, causing this writer to assume that the queue is not
|
||||
// shutdown, which results in a call to disconnect().
|
||||
final boolean queueWasShutdown = packetWriter.queue.isShutdown();
|
||||
closingStreamReceived.reportSuccess();
|
||||
|
||||
if (queueWasShutdown) {
|
||||
// We received a closing stream element *after* we initiated the
|
||||
// termination of the session by sending a closing stream element to
|
||||
// the server first
|
||||
return;
|
||||
} else {
|
||||
// We received a closing stream element from the server without us
|
||||
// sending a closing stream element first. This means that the
|
||||
// server wants to terminate the session, therefore disconnect
|
||||
// the connection
|
||||
LOGGER.info(XMPPTCPConnection.this
|
||||
+ " received closing </stream> element."
|
||||
+ " Server wants to terminate the connection, calling disconnect()");
|
||||
disconnect();
|
||||
}
|
||||
}
|
||||
break;
|
||||
case XmlPullParser.END_DOCUMENT:
|
||||
|
|
@ -1165,9 +1162,10 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
}
|
||||
}
|
||||
catch (Exception e) {
|
||||
closingStreamReceived.reportFailure(e);
|
||||
// The exception can be ignored if the the connection is 'done'
|
||||
// or if the it was caused because the socket got closed
|
||||
if (!(done || isSocketClosed())) {
|
||||
if (!(done || packetWriter.queue.isShutdown())) {
|
||||
// Close the connection and notify connection listeners of the
|
||||
// error.
|
||||
notifyConnectionError(e);
|
||||
|
|
@ -1186,7 +1184,7 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
* Needs to be protected for unit testing purposes.
|
||||
*/
|
||||
protected SynchronizationPoint<NoResponseException> shutdownDone = new SynchronizationPoint<NoResponseException>(
|
||||
XMPPTCPConnection.this);
|
||||
XMPPTCPConnection.this, "shutdown completed");
|
||||
|
||||
/**
|
||||
* If set, the stanza(/packet) writer is shut down
|
||||
|
|
@ -1234,9 +1232,14 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
}
|
||||
|
||||
protected void throwNotConnectedExceptionIfDoneAndResumptionNotPossible() throws NotConnectedException {
|
||||
if (done() && !isSmResumptionPossible()) {
|
||||
final boolean done = done();
|
||||
if (done) {
|
||||
final boolean smResumptionPossbile = isSmResumptionPossible();
|
||||
// Don't throw a NotConnectedException is there is an resumable stream available
|
||||
throw new NotConnectedException();
|
||||
if (!smResumptionPossbile) {
|
||||
throw new NotConnectedException(XMPPTCPConnection.this, "done=" + done
|
||||
+ " smResumptionPossible=" + smResumptionPossbile);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -1245,39 +1248,37 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
*
|
||||
* @param element the element to send.
|
||||
* @throws NotConnectedException
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
protected void sendStreamElement(Element element) throws NotConnectedException {
|
||||
protected void sendStreamElement(Element element) throws NotConnectedException, InterruptedException {
|
||||
throwNotConnectedExceptionIfDoneAndResumptionNotPossible();
|
||||
|
||||
boolean enqueued = false;
|
||||
while (!enqueued) {
|
||||
try {
|
||||
queue.put(element);
|
||||
enqueued = true;
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
throwNotConnectedExceptionIfDoneAndResumptionNotPossible();
|
||||
// If the method above did not throw, then the sending thread was interrupted
|
||||
// TODO in a later version of Smack the InterruptedException should be thrown to
|
||||
// allow users to interrupt a sending thread that is currently blocking because
|
||||
// the queue is full.
|
||||
LOGGER.log(Level.WARNING, "Sending thread was interrupted", e);
|
||||
}
|
||||
try {
|
||||
queue.put(element);
|
||||
}
|
||||
catch (InterruptedException e) {
|
||||
// put() may throw an InterruptedException for two reasons:
|
||||
// 1. If the queue was shut down
|
||||
// 2. If the thread was interrupted
|
||||
// so we have to check which is the case
|
||||
throwNotConnectedExceptionIfDoneAndResumptionNotPossible();
|
||||
// If the method above did not throw, then the sending thread was interrupted
|
||||
throw e;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Shuts down the stanza(/packet) writer. Once this method has been called, no further
|
||||
* packets will be written to the server.
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
void shutdown(boolean instant) {
|
||||
instantShutdown = instant;
|
||||
shutdownTimestamp = System.currentTimeMillis();
|
||||
queue.shutdown();
|
||||
shutdownTimestamp = System.currentTimeMillis();
|
||||
try {
|
||||
shutdownDone.checkIfSuccessOrWait();
|
||||
}
|
||||
catch (NoResponseException e) {
|
||||
catch (NoResponseException | InterruptedException e) {
|
||||
LOGGER.log(Level.WARNING, "shutdownDone was not marked as successful by the writer thread", e);
|
||||
}
|
||||
}
|
||||
|
|
@ -1374,7 +1375,15 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
throw new IllegalStateException(e);
|
||||
}
|
||||
}
|
||||
writer.write(element.toXML().toString());
|
||||
|
||||
CharSequence elementXml = element.toXML();
|
||||
if (elementXml instanceof XmlStringBuilder) {
|
||||
((XmlStringBuilder) elementXml).write(writer);
|
||||
}
|
||||
else {
|
||||
writer.write(elementXml.toString());
|
||||
}
|
||||
|
||||
if (queue.isEmpty()) {
|
||||
writer.flush();
|
||||
}
|
||||
|
|
@ -1405,6 +1414,7 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
catch (Exception e) {
|
||||
LOGGER.log(Level.WARNING, "Exception writing closing stream element", e);
|
||||
}
|
||||
|
||||
// Delete the queue contents (hopefully nothing is left).
|
||||
queue.clear();
|
||||
} else if (instantShutdown && isSmEnabled()) {
|
||||
|
|
@ -1412,19 +1422,15 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
// into the unacknowledgedStanzas queue
|
||||
drainWriterQueueToUnacknowledgedStanzas();
|
||||
}
|
||||
|
||||
try {
|
||||
writer.close();
|
||||
}
|
||||
catch (Exception e) {
|
||||
// Do nothing
|
||||
}
|
||||
|
||||
// Do *not* close the writer here, as it will cause the socket
|
||||
// to get closed. But we may want to receive further stanzas
|
||||
// until the closing stream tag is received. The socket will be
|
||||
// closed in shutdown().
|
||||
}
|
||||
catch (Exception e) {
|
||||
// The exception can be ignored if the the connection is 'done'
|
||||
// or if the it was caused because the socket got closed
|
||||
if (!(done() || isSocketClosed())) {
|
||||
if (!(done() || queue.isShutdown())) {
|
||||
notifyConnectionError(e);
|
||||
} else {
|
||||
LOGGER.log(Level.FINE, "Ignoring Exception in writePackets()", e);
|
||||
|
|
@ -1459,8 +1465,19 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
* Set if Stream Management resumption should be used by default for new connections.
|
||||
*
|
||||
* @param useSmResumptionDefault true to use Stream Management resumption for new connections.
|
||||
* @deprecated use {@link #setUseStreamManagementResumptionDefault(boolean)} instead.
|
||||
*/
|
||||
@Deprecated
|
||||
public static void setUseStreamManagementResumptiodDefault(boolean useSmResumptionDefault) {
|
||||
setUseStreamManagementDefault(useSmResumptionDefault);
|
||||
}
|
||||
|
||||
/**
|
||||
* Set if Stream Management resumption should be used by default for new connections.
|
||||
*
|
||||
* @param useSmResumptionDefault true to use Stream Management resumption for new connections.
|
||||
*/
|
||||
public static void setUseStreamManagementResumptionDefault(boolean useSmResumptionDefault) {
|
||||
if (useSmResumptionDefault) {
|
||||
// Also enable SM is resumption is enabled
|
||||
setUseStreamManagementDefault(useSmResumptionDefault);
|
||||
|
|
@ -1542,15 +1559,16 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
*
|
||||
* @throws StreamManagementNotEnabledException if Stream Mangement is not enabled.
|
||||
* @throws NotConnectedException if the connection is not connected.
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
public void requestSmAcknowledgement() throws StreamManagementNotEnabledException, NotConnectedException {
|
||||
public void requestSmAcknowledgement() throws StreamManagementNotEnabledException, NotConnectedException, InterruptedException {
|
||||
if (!isSmEnabled()) {
|
||||
throw new StreamManagementException.StreamManagementNotEnabledException();
|
||||
}
|
||||
requestSmAcknowledgementInternal();
|
||||
}
|
||||
|
||||
private void requestSmAcknowledgementInternal() throws NotConnectedException {
|
||||
private void requestSmAcknowledgementInternal() throws NotConnectedException, InterruptedException {
|
||||
packetWriter.sendStreamElement(AckRequest.INSTANCE);
|
||||
}
|
||||
|
||||
|
|
@ -1564,15 +1582,16 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
*
|
||||
* @throws StreamManagementNotEnabledException if Stream Management is not enabled.
|
||||
* @throws NotConnectedException if the connection is not connected.
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
public void sendSmAcknowledgement() throws StreamManagementNotEnabledException, NotConnectedException {
|
||||
public void sendSmAcknowledgement() throws StreamManagementNotEnabledException, NotConnectedException, InterruptedException {
|
||||
if (!isSmEnabled()) {
|
||||
throw new StreamManagementException.StreamManagementNotEnabledException();
|
||||
}
|
||||
sendSmAcknowledgementInternal();
|
||||
}
|
||||
|
||||
private void sendSmAcknowledgementInternal() throws NotConnectedException {
|
||||
private void sendSmAcknowledgementInternal() throws NotConnectedException, InterruptedException {
|
||||
packetWriter.sendStreamElement(new AckAnswer(clientHandledStanzasCount));
|
||||
}
|
||||
|
||||
|
|
@ -1785,8 +1804,8 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
try {
|
||||
listener.processPacket(ackedStanza);
|
||||
}
|
||||
catch (NotConnectedException e) {
|
||||
LOGGER.log(Level.FINER, "Received not connected exception", e);
|
||||
catch (InterruptedException | NotConnectedException e) {
|
||||
LOGGER.log(Level.FINER, "Received exception", e);
|
||||
}
|
||||
}
|
||||
String id = ackedStanza.getStanzaId();
|
||||
|
|
@ -1798,8 +1817,8 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
|
|||
try {
|
||||
listener.processPacket(ackedStanza);
|
||||
}
|
||||
catch (NotConnectedException e) {
|
||||
LOGGER.log(Level.FINER, "Received not connected exception", e);
|
||||
catch (InterruptedException | NotConnectedException e) {
|
||||
LOGGER.log(Level.FINER, "Received exception", e);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
|
|||
Loading…
Reference in a new issue