vanitasvitae/Smack
1
0
Fork 0
mirror of https://github.com/vanitasvitae/Smack.git synced 2024-11-25 13:32:07 +01:00
Code Issues Releases Wiki Activity

XMPPTCPConnection: Ensure both writer/reader threads are terminated

Browse source 
This should fix SMACK-855.
This commit is contained in:
Florian Schmaus 2019-02-06 21:49:58 +01:00
parent 5273402b87
commit 62cba0d96f
1 changed files with 17 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

16
smack-tcp/src/main/java/org/jivesoftware/smack/tcp/XMPPTCPConnection.java
View file

@ -51,6 +51,7 @@ import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue; import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedQueue; import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicBoolean;
import java.util.logging.Level; import java.util.logging.Level;
@ -166,6 +167,8 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
private SSLSocket secureSocket; private SSLSocket secureSocket;
private final Semaphore readerWriterSemaphore = new Semaphore(2);
/** /**
* Protected access level because of unit test purposes * Protected access level because of unit test purposes
*/ */
@ -635,8 +638,9 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
* @throws XMPPException if establishing a connection to the server fails. * @throws XMPPException if establishing a connection to the server fails.
* @throws SmackException if the server fails to respond back or if there is anther error. * @throws SmackException if the server fails to respond back or if there is anther error.
* @throws IOException * @throws IOException
* @throws InterruptedException
*/ */
private void initConnection() throws IOException { private void initConnection() throws IOException, InterruptedException {
boolean isFirstInitialization = packetReader == null || packetWriter == null; boolean isFirstInitialization = packetReader == null || packetWriter == null;
compressionHandler = null; compressionHandler = null;
@ -647,6 +651,8 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
packetWriter = new PacketWriter(); packetWriter = new PacketWriter();
packetReader = new PacketReader(); packetReader = new PacketReader();
} }
readerWriterSemaphore.acquire(2);
// Start the writer thread. This will open an XMPP stream to the server // Start the writer thread. This will open an XMPP stream to the server
packetWriter.init(); packetWriter.init();
// Start the reader thread. The startup() method will block until we // Start the reader thread. The startup() method will block until we
@ -1014,7 +1020,11 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
Async.go(new Runnable() { Async.go(new Runnable() {
@Override @Override
public void run() { public void run() {
try {
parsePackets(); parsePackets();
} finally {
XMPPTCPConnection.this.readerWriterSemaphore.release();
}
} }
}, "Smack Reader (" + getConnectionCounter() + ")"); }, "Smack Reader (" + getConnectionCounter() + ")");
} }
@ -1310,7 +1320,11 @@ public class XMPPTCPConnection extends AbstractXMPPConnection {
Async.go(new Runnable() { Async.go(new Runnable() {
@Override @Override
public void run() { public void run() {
try {
writePackets(); writePackets();
} finally {
XMPPTCPConnection.this.readerWriterSemaphore.release();
}
} }
}, "Smack Writer (" + getConnectionCounter() + ")"); }, "Smack Writer (" + getConnectionCounter() + ")");
} }