From c69c1bb13416068597fe5204190a5c7494746e9c Mon Sep 17 00:00:00 2001
From: Redkale <8730487+redkale@users.noreply.github.com>
Date: Sat, 4 Aug 2018 15:12:22 +0800
Subject: [PATCH] ppl

---
 src/org/redkale/net/PrepareServlet.java | 12 ++++++++++--
 src/org/redkale/net/Request.java        | 16 ++++++++++++++++
 src/org/redkale/net/Response.java       | 23 +++++++++++++----------
 3 files changed, 39 insertions(+), 12 deletions(-)

diff --git a/src/org/redkale/net/PrepareServlet.java b/src/org/redkale/net/PrepareServlet.java
index b6c416327..ec5df1be5 100644
--- a/src/org/redkale/net/PrepareServlet.java
+++ b/src/org/redkale/net/PrepareServlet.java
@@ -218,7 +218,11 @@ public abstract class PrepareServlet<K extends Serializable, C extends Context,
             if (rs != Integer.MIN_VALUE) illRequestCounter.incrementAndGet();
             response.finish(true);
         } else if (rs == 0) {
-            request.offerReadBuffer(buffer);
+            if (buffer.hasRemaining()) {
+                request.setMoredata(buffer);
+            } else {
+                request.offerReadBuffer(buffer);
+            }
             request.prepare();
             response.filter = this.headFilter;
             response.servlet = this;
@@ -236,7 +240,11 @@ public abstract class PrepareServlet<K extends Serializable, C extends Context,
                         buffer.clear();
                         request.channel.read(buffer, buffer, this);
                     } else {
-                        request.offerReadBuffer(buffer);
+                        if (buffer.hasRemaining()) {
+                            request.setMoredata(buffer);
+                        } else {
+                            request.offerReadBuffer(buffer);
+                        }
                         request.prepare();
                         try {
                             response.filter = PrepareServlet.this.headFilter;
diff --git a/src/org/redkale/net/Request.java b/src/org/redkale/net/Request.java
index 5ede7d4a1..3f9ce8a23 100644
--- a/src/org/redkale/net/Request.java
+++ b/src/org/redkale/net/Request.java
@@ -31,6 +31,10 @@ public abstract class Request<C extends Context> {
 
     protected boolean keepAlive;
 
+    protected boolean more; //pipeline模式
+
+    protected ByteBuffer moredata; //pipeline模式
+
     protected AsyncConnection channel;
 
     protected ByteBuffer readBuffer;
@@ -50,6 +54,16 @@ public abstract class Request<C extends Context> {
         this.jsonConvert = context.getJsonConvert();
     }
 
+    protected void setMoredata(ByteBuffer buffer) {
+        this.moredata = buffer;
+    }
+
+    protected ByteBuffer removeMoredata() {
+        ByteBuffer rs = this.moredata;
+        this.moredata = null;
+        return rs;
+    }
+
     protected ByteBuffer pollReadBuffer() {
         ByteBuffer buffer = this.readBuffer;
         this.readBuffer = null;
@@ -90,6 +104,8 @@ public abstract class Request<C extends Context> {
     protected void recycle() {
         createtime = 0;
         keepAlive = false;
+        more = false;
+        moredata = null;
         attributes.clear();
         channel = null; // close it by response
     }
diff --git a/src/org/redkale/net/Response.java b/src/org/redkale/net/Response.java
index 74ab32dc0..a64fde995 100644
--- a/src/org/redkale/net/Response.java
+++ b/src/org/redkale/net/Response.java
@@ -30,8 +30,6 @@ public abstract class Response<C extends Context, R extends Request<C>> {
 
     protected AsyncConnection channel;
 
-    protected ByteBuffer moredata; //pipeline模式
-
     protected ByteBuffer writeHeadBuffer;
 
     protected ByteBuffer writeBodyBuffer;
@@ -169,12 +167,6 @@ public abstract class Response<C extends Context, R extends Request<C>> {
         return ch;
     }
 
-    protected ByteBuffer removeMoredata() {
-        ByteBuffer rs = this.moredata;
-        this.moredata = null;
-        return rs;
-    }
-
     protected void prepare() {
         inited = true;
     }
@@ -184,7 +176,6 @@ public abstract class Response<C extends Context, R extends Request<C>> {
         this.output = null;
         this.filter = null;
         this.servlet = null;
-        this.moredata = null;
         request.recycle();
         if (channel != null) {
             channel.dispose();
@@ -260,7 +251,7 @@ public abstract class Response<C extends Context, R extends Request<C>> {
             }
             this.recycleListener = null;
         }
-        if (request.keepAlive && channel != null) {
+        if (request.keepAlive && !request.more && channel != null) {
             if (channel.isOpen()) {
                 AsyncConnection conn = removeChannel();
                 this.recycle();
@@ -288,24 +279,36 @@ public abstract class Response<C extends Context, R extends Request<C>> {
 
     public void finish(ByteBuffer buffer) {
         if (!this.inited) return; //避免重复关闭
+        ByteBuffer data = this.request.removeMoredata();
+        this.request.more = data != null && this.request.keepAlive;
         this.channel.write(buffer, buffer, finishHandler);
+        if (this.request.more) new PrepareRunner(this.context, this.channel, data, null).run();
     }
 
     public void finish(boolean kill, ByteBuffer buffer) {
         if (!this.inited) return; //避免重复关闭
         if (kill) refuseAlive();
+        ByteBuffer data = this.request.removeMoredata();
+        this.request.more = data != null && this.request.keepAlive;
         this.channel.write(buffer, buffer, finishHandler);
+        if (this.request.more) new PrepareRunner(this.context, this.channel, data, null).run();
     }
 
     public void finish(ByteBuffer... buffers) {
         if (!this.inited) return; //避免重复关闭
+        ByteBuffer data = this.request.removeMoredata();
+        this.request.more = data != null && this.request.keepAlive;
         this.channel.write(buffers, buffers, finishHandler2);
+        if (this.request.more) new PrepareRunner(this.context, this.channel, data, null).run();
     }
 
     public void finish(boolean kill, ByteBuffer... buffers) {
         if (!this.inited) return; //避免重复关闭
         if (kill) refuseAlive();
+        ByteBuffer data = this.request.removeMoredata();
+        this.request.more = data != null && this.request.keepAlive;
         this.channel.write(buffers, buffers, finishHandler2);
+        if (this.request.more) new PrepareRunner(this.context, this.channel, data, null).run();
     }
 
     protected <A> void send(final ByteBuffer buffer, final A attachment, final CompletionHandler<Integer, A> handler) {