From 89f948aeaef34114ec61291ff8ffbebcf85a748e Mon Sep 17 00:00:00 2001
From: Eric Wong <normalperson@yhbt.net>
Date: Tue, 11 Jan 2011 16:01:20 -0800
Subject: event_machine: buffer reads when waiting for async.callback

We cannot trigger on_read events and invoke the HTTP parser and
modify @env while we're waiting for an application to run
async.callback.  We also need to clear (and *maybe* re-set)
@deferred if we're writing from async.callback
---
 lib/rainbows/coolio/client.rb        |  7 +------
 lib/rainbows/ev_core.rb              |  1 +
 lib/rainbows/event_machine/client.rb |  2 +-
 t/t0402-async-keepalive.sh           | 22 +++++++++++++++++++++-
 4 files changed, 24 insertions(+), 8 deletions(-)

diff --git a/lib/rainbows/coolio/client.rb b/lib/rainbows/coolio/client.rb
index 6264df7..f44d2aa 100644
--- a/lib/rainbows/coolio/client.rb
+++ b/lib/rainbows/coolio/client.rb
@@ -105,17 +105,12 @@ class Rainbows::Coolio::Client < Coolio::IO
     @state = :headers
   end
 
-  def coolio_write_async_response(response)
-    write_async_response(response)
-    @deferred = nil
-  end
-
   def app_call
     KATO.delete(self)
     disable if enabled?
     @env[RACK_INPUT] = @input
     @env[REMOTE_ADDR] = @_io.kgio_addr
-    @env[ASYNC_CALLBACK] = method(:coolio_write_async_response)
+    @env[ASYNC_CALLBACK] = method(:write_async_response)
     status, headers, body = catch(:async) {
       APP.call(@env.merge!(RACK_DEFAULTS))
     }
diff --git a/lib/rainbows/ev_core.rb b/lib/rainbows/ev_core.rb
index 0bdaab3..a3bf50f 100644
--- a/lib/rainbows/ev_core.rb
+++ b/lib/rainbows/ev_core.rb
@@ -22,6 +22,7 @@ module Rainbows::EvCore
       alive = headers.include?(Content_Length) ||
               !!(%r{\Achunked\z}i =~ headers[Transfer_Encoding])
     end
+    @deferred = nil
     ev_write_response(status, headers, body, alive)
   end
 
diff --git a/lib/rainbows/event_machine/client.rb b/lib/rainbows/event_machine/client.rb
index c9f63a4..b75d7ee 100644
--- a/lib/rainbows/event_machine/client.rb
+++ b/lib/rainbows/event_machine/client.rb
@@ -40,7 +40,7 @@ class Rainbows::EventMachine::Client < EM::Connection
       APP.call(@env.merge!(RACK_DEFAULTS))
     }
 
-    (nil == status || -1 == status) or
+    (nil == status || -1 == status) ? @deferred = true :
       ev_write_response(status, headers, body, @hp.next?)
   end
 
diff --git a/t/t0402-async-keepalive.sh b/t/t0402-async-keepalive.sh
index fdf4cbc..7b7b2e7 100644
--- a/t/t0402-async-keepalive.sh
+++ b/t/t0402-async-keepalive.sh
@@ -9,7 +9,7 @@ Coolio|NeverBlock|EventMachine) ;;
 	;;
 esac
 
-t_plan 11 "async_chunk_app test for test for $model"
+t_plan 12 "async_chunk_app test for test for $model"
 
 CONFIG_RU=async_chunk_app.ru
 
@@ -44,6 +44,26 @@ t_begin "async.callback supports pipelining" && {
 	test 3 -eq "$(grep 'World ' $tmp | wc -l)"
 }
 
+t_begin "async.callback supports delayed pipelining" && {
+	rm -f $tmp
+	t0=$(date +%s)
+	(
+		cat $fifo > $tmp &
+		printf 'GET /0 HTTP/1.1\r\nHost: example.com\r\n\r\n'
+		sleep 1
+		printf 'GET /1 HTTP/1.1\r\nHost: example.com\r\n\r\n'
+		sleep 1
+		printf 'GET /2 HTTP/1.0\r\nHost: example.com\r\n\r\n'
+		wait
+	) | socat - TCP:$listen > $fifo
+	t1=$(date +%s)
+	elapsed=$(( $t1 - $t0 ))
+	t_info "elapsed=$elapsed $model.$0 ($t_current)"
+	test 3 -eq "$(fgrep 'HTTP/1.1 200 OK' $tmp | wc -l)"
+	test 3 -eq "$(grep '^Hello ' $tmp | wc -l)"
+	test 3 -eq "$(grep 'World ' $tmp | wc -l)"
+}
+
 t_begin "async.callback supports pipelining with delay $DELAY" && {
 	rm -f $tmp
 	t0=$(date +%s)
-- 
cgit v1.2.3-24-ge0c7