streaming responses

[matti]

I tried to adapt code from beer.rb and run falcon from my ruby and not with falcon - can't get streaming responses to work, the following outputs just one "hello" before connection is closed:

class Chunk
  def call(env)
    task = Async::Task.current
    body = Async::HTTP::Body::Writable.new
    task.async do |task|
      10.times do
        body.write "hello"
        task.sleep 0.5
      end
    rescue Exception => ex
      p ex
    ensure
      body.close
    end

    [200 {'content-type' => 'text/html; charset=utf-8'}, body]
  end
end

rack_app = Rack::Builder.new do
  use Rack::CommonLogger

  map "/chunk" do
    run Chunk.new
  end
end

app = Falcon::Server.middleware rack_app, verbose: true

endpoint = Async::HTTP::URLEndpoint.parse "http://0.0.0.0:1234"
bound_endpoint = Async::Reactor.run do
  Async::IO::SharedEndpoint.bound(endpoint)
end.result

server = Falcon::Server.new(app, bound_endpoint, endpoint.protocol, endpoint.scheme)

Async::Reactor.run do
  server.run
  puts "serving..."
end

[ioquatix]

use Rack::CommonLogger is not compatible with streaming body it seems.

[ioquatix]

Here is implementation:

https://github.com/rack/rack/blob/master/lib/rack/common_logger.rb

It's using BodyProxy, so the code here is not triggered:

falcon/lib/falcon/adapters/output.rb

Lines 39 to 40 in 68b274b

	if body.is_a?(Async::HTTP::Body::Readable)
	return body

Instead it goes through this code path:

falcon/lib/falcon/adapters/output.rb

Line 45 in 68b274b

return self.new(headers, body, content_length)

This makes an enumerator over the response body:

falcon/lib/falcon/adapters/output.rb

Line 54 in 68b274b

@chunks = body.to_enum(:each)

But this is not well supported in Ruby at the moment for async tasks. Therefore, it fails in a strange way. Unfortunately the best solution right now is to avoid using Rack::CommonLogger or for us to some how special case this.

[ioquatix]

Here is a working example:

#!/usr/bin/env ruby

require 'async'
require 'async/http/client'
require 'async/http/url_endpoint'
require 'async/io/shared_endpoint'
require 'falcon/server'

# Async.logger.level = Logger::DEBUG

class Chunk
	def call(env)
		task = Async::Task.current
		body = Async::HTTP::Body::Writable.new
		task.async do |task|
			10.times do
				body.write "hello"
				task.sleep 0.5
			end
		rescue Exception => ex
			p ex
		ensure
			body.close
		end
		
		[200, {'content-type' => 'text/html; charset=utf-8'}, body]
	end
end
rack_app = Rack::Builder.new do
	#use Rack::CommonLogger

	map "/chunk" do
		run Chunk.new
	end
end

app = Falcon::Server.middleware rack_app, verbose: true

endpoint = Async::HTTP::URLEndpoint.parse "http://0.0.0.0:12345"

bound_endpoint = Async::Reactor.run do
	Async::IO::SharedEndpoint.bound(endpoint)
end.result

server = Falcon::Server.new(app, bound_endpoint, endpoint.protocol, endpoint.scheme)

Async.run do |task|
	server_task = task.async do
		puts "Starting server..."
		server.run
	end
	
	client = Async::HTTP::Client.new(endpoint)
	
	puts "Fetching /chunk"
	response = client.get("/chunk")
	
	response.body.each do |chunk|
		puts chunk
	end
end

[matti]

@ioquatix thank you again!

[ioquatix]

The only real solution I can see here is to fix enumerators: ruby/ruby#2002

[matti]

I think issue can be closed - the example works great.