event_machine: refactor async.callback for keepalive

async.callback will be useful with Coolio (and more!) soon, so
ensure it works as well as the rest of Rainbows!

2 files changed, 32 insertions, 25 deletions

diff --git a/lib/rainbows/event_machine/client.rb b/lib/rainbows/event_machine/client.rb
index 5abdc3b..22e5360 100644
--- a/lib/rainbows/event_machine/client.rb
+++ b/lib/rainbows/event_machine/client.rb
@@ -48,18 +48,7 @@ class Rainbows::EventMachine::Client < EM::Connection
     # second (pipelined) request could be a stuck behind a
     # long-running async response
     (status.nil? || -1 == status) and return @state = :close
-
-    if @hp.next?
-      @state = :headers
-      write_response(status, headers, body, true)
-      if @buf.empty?
-        set_comm_inactivity_timeout(Rainbows.keepalive_timeout)
-      elsif @body.nil?
-        EM.next_tick { receive_data(nil) }
-      end
-    else
-      write_response(status, headers, body, false)
-    end
+    write_response(status, headers, body, @hp.next?)
   end
 
   def next!
diff --git a/lib/rainbows/event_machine/response.rb b/lib/rainbows/event_machine/response.rb
index 49bcbd5..7b88261 100644
--- a/lib/rainbows/event_machine/response.rb
+++ b/lib/rainbows/event_machine/response.rb
@@ -1,27 +1,35 @@
 # -*- encoding: binary -*-
 # :enddoc:
 module Rainbows::EventMachine::Response
+  def deferred_errback(orig_body)
+    @body.errback do
+      orig_body.close if orig_body.respond_to?(:close)
+      quit
+    end
+  end
+
+  def deferred_callback(orig_body, alive)
+    @body.callback do
+      orig_body.close if orig_body.respond_to?(:close)
+      @body = nil
+      alive ? receive_data(nil) : quit
+    end
+  end
+
   def write_response(status, headers, body, alive)
+    @state = :headers if alive
     if body.respond_to?(:errback) && body.respond_to?(:callback)
       @body = body
-      body.callback { quit }
-      body.errback { quit }
-      alive = true
+      deferred_errback(body)
+      deferred_callback(body, alive)
     elsif body.respond_to?(:to_path)
       st = File.stat(path = body.to_path)
 
       if st.file?
         write_headers(status, headers, alive)
         @body = stream_file_data(path)
-        @body.errback do
-          body.close if body.respond_to?(:close)
-          quit
-        end
-        @body.callback do
-          body.close if body.respond_to?(:close)
-          @body = nil
-          alive ? receive_data(nil) : quit
-        end
+        deferred_errback(body)
+        deferred_callback(body, alive)
         return
       elsif st.socket? || st.pipe?
         io = body_to_io(@body = body)
@@ -33,6 +41,16 @@ module Rainbows::EventMachine::Response
       # char or block device... WTF? fall through to body.each
     end
     super(status, headers, body, alive)
-    quit unless alive
+    if alive
+      if @body.nil?
+        if @buf.empty?
+          set_comm_inactivity_timeout(Rainbows.keepalive_timeout)
+        else
+          EM.next_tick { receive_data(nil) }
+        end
+      end
+    else
+      quit unless @body
+    end
   end
 end