lxy/redkale
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复AioAsyncConnection关闭时write队列还有数据没写完的bug

Browse Source
This commit is contained in:
Redkale
2019-01-18 15:50:23 +08:00
parent 204514cb08
commit cf2ab617c2
1 changed files with 17 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files

30
src/org/redkale/net/TcpAioAsyncConnection.java
View File

@@ -110,18 +110,22 @@ public class TcpAioAsyncConnection extends AsyncConnection {
channel.read(dst, timeout < 0 ? 0 : timeout, unit, dst, handler); channel.read(dst, timeout < 0 ? 0 : timeout, unit, dst, handler);
} }
private <A> void nextWrite(A attachment) { private <A> void nextWrite(Throwable exc, A attachment) {
BlockingQueue<WriteEntry> queue = this.writeQueue; BlockingQueue<WriteEntry> queue = this.writeQueue;
WriteEntry entry = queue == null ? null : queue.poll(); WriteEntry entry = queue == null ? null : queue.poll();
if (entry != null) { if (entry != null) {
try { if (exc == null) {
if (entry.writeOneBuffer == null) { try {
write(false, entry.writeBuffers, entry.writeOffset, entry.writeLength, entry.writeAttachment, entry.writeHandler); if (entry.writeOneBuffer == null) {
} else { write(false, entry.writeBuffers, entry.writeOffset, entry.writeLength, entry.writeAttachment, entry.writeHandler);
write(false, entry.writeOneBuffer, entry.writeAttachment, entry.writeHandler); } else {
write(false, entry.writeOneBuffer, entry.writeAttachment, entry.writeHandler);
}
} catch (Exception e) {
entry.writeHandler.failed(e, entry.writeAttachment);
} }
} catch (Exception e) { } else { //当连接已经关掉了，不需要调用write方法，直接报异常
entry.writeHandler.failed(e, entry.writeAttachment); entry.writeHandler.failed(exc, entry.writeAttachment);
} }
} else { } else {
semaphore.release(); semaphore.release();
@@ -312,17 +316,17 @@ public class TcpAioAsyncConnection extends AsyncConnection {
failed(e, attachment); failed(e, attachment);
return; return;
} }
nextWrite(attachment); nextWrite(null, attachment);
writeHandler.completed(writeCount, attachment); writeHandler.completed(writeCount, attachment);
} else { } else {
nextWrite(attachment); nextWrite(null, attachment);
writeHandler.completed(result.intValue(), attachment); writeHandler.completed(result.intValue(), attachment);
} }
} }
@Override @Override
public void failed(Throwable exc, A attachment) { public void failed(Throwable exc, A attachment) {
nextWrite(attachment); nextWrite(isOpen() ? null : exc, attachment);
writeHandler.failed(exc, attachment); writeHandler.failed(exc, attachment);
} }
@@ -350,13 +354,13 @@ public class TcpAioAsyncConnection extends AsyncConnection {
failed(e, attachment); failed(e, attachment);
return; return;
} }
nextWrite(attachment); nextWrite(null, attachment);
writeHandler.completed(result, attachment); writeHandler.completed(result, attachment);
} }
@Override @Override
public void failed(Throwable exc, A attachment) { public void failed(Throwable exc, A attachment) {
nextWrite(attachment); nextWrite(isOpen() ? null : exc, attachment);
writeHandler.failed(exc, attachment); writeHandler.failed(exc, attachment);
} }