Spring/spring-integration
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

INT-1367 Fix Deadlock on NIO Reading Thread

Browse Source
This commit is contained in:
Gary Russell
2010-08-23 18:40:25 +00:00
parent 971e5dc465
commit 1d5ccb7a0e
1 changed files with 21 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
spring-integration-ip/src/main/java/org/springframework/integration/ip/tcp/connection/TcpNioConnection.java
View File

@@ -63,6 +63,8 @@ public class TcpNioConnection extends AbstractTcpConnection {
private long lastRead;
private AtomicInteger executionControl = new AtomicInteger();
private boolean writingToPipe;
/**
* Constructs a TcpNetConnection for the SocketChannel.
@@ -156,8 +158,7 @@ public class TcpNioConnection extends AbstractTcpConnection {
}
while (active) {
try {
while (this.socketChannel.isOpen() &&
this.pipedInputStream.available() > 0) {
while (dataAvailable()) {
convertAndSend();
}
} catch (IOException e) {
@@ -174,8 +175,13 @@ public class TcpNioConnection extends AbstractTcpConnection {
}
}
private boolean dataAvailable() throws IOException {
return this.socketChannel.isOpen() &&
(this.pipedInputStream.available() > 0 || writingToPipe);
}
private synchronized void convertAndSend() throws IOException {
if (!this.socketChannel.isOpen() || this.pipedInputStream.available() <= 0) {
if (!dataAvailable()) {
return;
}
Message<?> message = null;
@@ -231,6 +237,16 @@ public class TcpNioConnection extends AbstractTcpConnection {
if (rawBuffer == null) {
rawBuffer = allocate(maxMessageSize);
}
writingToPipe = true;
if (this.taskExecutor == null) {
this.taskExecutor = Executors.newSingleThreadExecutor();
}
if (this.executionControl.incrementAndGet() <= 1) {
// only execute run() if we don't already have one running
this.executionControl.set(1);
this.taskExecutor.execute(this);
}
rawBuffer.clear();
int len = socketChannel.read(rawBuffer);
if (len < 0) {
@@ -243,15 +259,8 @@ public class TcpNioConnection extends AbstractTcpConnection {
}
pipedOutputStream.write(rawBuffer.array(), 0, rawBuffer.limit());
pipedOutputStream.flush();
if (this.taskExecutor == null) {
this.taskExecutor = Executors.newSingleThreadExecutor();
}
if (this.executionControl.incrementAndGet() <= 1) {
// only execute run() if we don't already have one running
this.executionControl.set(1);
this.taskExecutor.execute(this);
}
writingToPipe = false;
}
/**