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

Browse Source
This commit is contained in:
Redkale
2018-04-30 14:27:15 +08:00
parent a416ae564a
commit c523a761c5
3 changed files with 21 additions and 17 deletions
Download Patch File Download Diff File Expand all files Collapse all files

14
src/org/redkale/net/PrepareRunner.java
View File

@@ -26,23 +26,24 @@ public final class PrepareRunner implements Runnable {
private final Context context; private final Context context;
private final boolean keepAlive;
private ByteBuffer data; private ByteBuffer data;
public PrepareRunner(Context context, AsyncConnection channel, ByteBuffer data, boolean keepAlive) { private Response response;
public PrepareRunner(Context context, AsyncConnection channel, ByteBuffer data, Response response) {
this.context = context; this.context = context;
this.channel = channel; this.channel = channel;
this.data = data; this.data = data;
this.keepAlive = keepAlive; this.response = response;
} }
@Override @Override
public void run() { public void run() {
final boolean keepalive = response != null;
final PrepareServlet prepare = context.prepare; final PrepareServlet prepare = context.prepare;
final ObjectPool<? extends Response> responsePool = context.responsePool; final ObjectPool<? extends Response> responsePool = context.responsePool;
final Response response = responsePool.get();
if (data != null) { //BIO模式的UDP连接创建AsyncConnection时已经获取到ByteBuffer数据了 if (data != null) { //BIO模式的UDP连接创建AsyncConnection时已经获取到ByteBuffer数据了
if (response == null) response = responsePool.get();
try { try {
response.init(channel); response.init(channel);
prepare.prepare(data, response.request, response); prepare.prepare(data, response.request, response);
@@ -52,9 +53,10 @@ public final class PrepareRunner implements Runnable {
} }
return; return;
} }
if (response == null) response = responsePool.get();
final ByteBuffer buffer = response.request.pollReadBuffer(); final ByteBuffer buffer = response.request.pollReadBuffer();
try { try {
channel.read(buffer, keepAlive ? context.getAliveTimeoutSeconds() : 0, TimeUnit.SECONDS, null, channel.read(buffer, keepalive ? context.getAliveTimeoutSeconds() : 0, TimeUnit.SECONDS, null,
new CompletionHandler<Integer, Void>() { new CompletionHandler<Integer, Void>() {
@Override @Override
public void completed(Integer count, Void attachment1) { public void completed(Integer count, Void attachment1) {

4
src/org/redkale/net/ProtocolServer.java
View File

@@ -150,7 +150,7 @@ public abstract class ProtocolServer {
SocketAddress address = serchannel.receive(buffer); SocketAddress address = serchannel.receive(buffer);
buffer.flip(); buffer.flip();
AsyncConnection conn = AsyncConnection.create(serchannel, address, false, readTimeoutSeconds, writeTimeoutSeconds); AsyncConnection conn = AsyncConnection.create(serchannel, address, false, readTimeoutSeconds, writeTimeoutSeconds);
context.runAsync(new PrepareRunner(context, conn, buffer, false)); context.runAsync(new PrepareRunner(context, conn, buffer, null));
} catch (Exception e) { } catch (Exception e) {
context.offerBuffer(buffer); context.offerBuffer(buffer);
} }
@@ -244,7 +244,7 @@ public abstract class ProtocolServer {
AsyncConnection conn = AsyncConnection.create(channel, null, context); AsyncConnection conn = AsyncConnection.create(channel, null, context);
conn.livingCounter = livingCounter; conn.livingCounter = livingCounter;
conn.closedCounter = closedCounter; conn.closedCounter = closedCounter;
context.runAsync(new PrepareRunner(context, conn, null, false)); context.runAsync(new PrepareRunner(context, conn, null, null));
} }
@Override @Override

20
src/org/redkale/net/Response.java
View File

@@ -169,19 +169,14 @@ public abstract class Response<C extends Context, R extends Request<C>> {
protected boolean recycle() { protected boolean recycle() {
if (!inited) return false; if (!inited) return false;
boolean keepAlive = request.keepAlive;
this.output = null; this.output = null;
this.filter = null; this.filter = null;
this.servlet = null; this.servlet = null;
request.recycle(); request.recycle();
if (channel != null) { if (channel != null) {
if (keepAlive) { try {
this.context.runAsync(new PrepareRunner(context, channel, null, keepAlive)); if (channel.isOpen()) channel.close();
} else { } catch (Exception e) {
try {
if (channel.isOpen()) channel.close();
} catch (Exception e) {
}
} }
channel = null; channel = null;
} }
@@ -255,7 +250,14 @@ public abstract class Response<C extends Context, R extends Request<C>> {
} }
this.recycleListener = null; this.recycleListener = null;
} }
this.context.responsePool.accept(this); if (request.keepAlive) {
AsyncConnection conn = removeChannel();
this.recycle();
this.prepare();
new PrepareRunner(context, conn, null, this).run();
} else {
this.context.responsePool.accept(this);
}
} }
public void finish(final byte[] bs) { public void finish(final byte[] bs) {