Streaming HTTP for Ruby

Ruby has several existing HTTP clients and servers, but none have support for both HTTP/1 and HTTP/2 streaming of request and response bodies. Native support for HTTP/2 is important especially when interacting with remote systems, as the per-request overhead is greatly reduced and it can be used for long-running connections. We present async-http, a modern, streaming HTTP client and server supporting HTTP/1, HTTP/2 and SSL out of the box.

Requests and Responses

Async::HTTP provides abstractions over HTTP/1 and HTTP/2 to minimise the complexity of your code. There is a beautiful symmetry: clients send requests and receive responses. Servers receive requests and send responses. This is reflected in the code.

#!/usr/bin/env ruby

require 'async'
require 'async/http/url_endpoint'
require 'async/http/server'
require 'async/http/client'

Async do
	endpoint = Async::HTTP::URLEndpoint.parse("http://localhost:8080")
	
	# Run the server in a separate fiber:
	server_task = Async do
		server = Async::HTTP::Server.for(endpoint) do |request|
			# Receive the request:
			Async::HTTP::Response[200,
				{'content-type' => 'text/plain'},
				["Hello World!"]
			]
		end
		
		server.run
	end
	
	# Create a client:
	client = Async::HTTP::Client.new(endpoint)
	
	# Make the request (it gets wrapped into Async::HTTP::Request):
	response = client.get("/")
	
	# Print out the response:
	puts "Status: #{response.status} (#{response.success?})"
	puts "Headers: #{response.headers.to_h}"
	puts "Body: #{response.read.inspect}"
ensure
	# Stop the server:
	server_task.stop if server_task
	
	# Close any persistent connections:
	client.close if client
end

Of particular interest is request and response bodies. These are typically an array of strings (chunks). To enable streaming you must uses a queue of chunks.

Streaming

HTTP allows streaming of both request and response bodies, even at the same time. This allows you to incrementally generate and consume information, which minimises latency and memory usage. Async::HTTP::Body contains the implementations required to stream both request and response bodies, and its usage is very straight forward.

Streaming Response Body

Perhaps the most basic use case, incrementally generating information and streaming it back to the client. Incrementally streaming events or records, formatted as newline delimited JSON, is a typical use case.

#!/usr/bin/env ruby

require 'async'
require 'async/http/url_endpoint'
require 'async/http/server'
require 'async/http/client'

# Show what's going on under the hood:
# Async.logger.debug!

require 'json'
APPLICATION_JSON_STREAM = "application/json; boundary=NL".freeze

Async do
	endpoint = Async::HTTP::URLEndpoint.parse("http://localhost:8080", timeout: 5)
	
	# Run the server in a separate fiber:
	server_task = Async do
		server = Async::HTTP::Server.for(endpoint) do |request|
			body = Async::HTTP::Body::Writable.new
			
			Async do |task|
				while true
					duration = rand
					task.sleep duration
					body.write(JSON.dump(["Hello World!", "I slept for #{duration}"]) + "\n")
				end
			end
			
			# Receive the request and respond with the dynamic body:
			Async::HTTP::Response[200,
				{'content-type' => APPLICATION_JSON_STREAM},
				# We echo the body back to the client:
				body
			]
		end
		
		server.run
	end
	
	# Create a client:
	client = Async::HTTP::Client.new(endpoint)
	
	response = client.get("/")
	
	response.each do |chunk|
		# As the server generates chunks we will receive them.
		# While Async::HTTP guarantees to preserve the chunk boundaries with HTTP/1.1 and HTTP/2, you might prefer to do additional buffering here to improve robustness.
		puts JSON.parse(chunk).inspect
	end
ensure
	# Stop the server:
	server_task.stop if server_task
	
	# Close any persistent connections:
	client.close if client
end

Streaming Request Bodies

Some services accept streaming bodies, e.g. upload services, incremental processing (voice recognition, video processing), and so on. While we are uploading the content, it's possible the remote server would send us back information, e.g. the current transcript, objects recognised, etc. Here is a simple streaming upper case service:

#!/usr/bin/env ruby

require 'async'
require 'async/http/url_endpoint'
require 'async/http/server'
require 'async/http/client'

# Show what's going on under the hood:
# Async.logger.debug!

require 'json'
APPLICATION_JSON_STREAM = "application/json; boundary=NL".freeze

Async do
	endpoint = Async::HTTP::URLEndpoint.parse("http://localhost:8080", timeout: 5)
	
	# Run the server in a separate fiber:
	server_task = Async do
		server = Async::HTTP::Server.for(endpoint) do |request|
			body = Async::HTTP::Body::Writable.new
			
			Async do
				request.each do |chunk|
					body.write chunk.upcase
				end
				
			ensure
				body.close
			end
			
			# Receive the request:
			Async::HTTP::Response[200,
				{'content-type' => request.headers['content-type']},
				# We echo the body back to the client:
				body
			]
		end
		
		server.run
	end
	
	# Create a client:
	client = Async::HTTP::Client.new(endpoint)
	
	# Make the request (it gets wrapped into Async::HTTP::Request):
	body = Async::HTTP::Body::Writable.new
	
	response = client.post("/", {'content-type' => 'text/plain'}, body)
	
	5.times do
		body.write("ping")
		puts response.body.read.inspect
	end
	
	body.close
	response.finish
ensure
	# Stop the server:
	server_task.stop if server_task
	
	# Close any persistent connections:
	client.close if client
end

Echo Server

Because of the symmetry of the implementation, fun services like this become possible:

#!/usr/bin/env ruby

require 'async'
require 'async/http/url_endpoint'
require 'async/http/server'
require 'async/http/client'

# Show what's going on under the hood:
# Async.logger.debug!

Async do
	endpoint = Async::HTTP::URLEndpoint.parse("http://localhost:8080", timeout: 5)
	
	# Run the server in a separate fiber:
	server_task = Async do
		server = Async::HTTP::Server.for(endpoint) do |request|
			# Receive the request:
			Async::HTTP::Response[200,
				{'content-type' => 'text/plain'},
				# We echo the body back to the client:
				request.body
			]
		end
		
		server.run
	end
	
	# Create a client:
	client = Async::HTTP::Client.new(endpoint)
	
	# Make the request (it gets wrapped into Async::HTTP::Request):
	body = Async::HTTP::Body::Writable.new
	
	response = client.post("/", {}, body)
	
	5.times do
		body.write(`fortune`)
		puts "*~" * 40
		puts response.body.read
	end
	
	body.close
	response.finish
ensure
	# Stop the server:
	server_task.stop if server_task
	
	# Close any persistent connections:
	client.close if client
end

99 bottles of beer on the wall

A reasonable question would be - does this work in production? If you use falcon, the answer is YES. The implementation even works with Rack.

Conclusion

Streaming request and response bodies minimise latency and open up new opportunities for interactivity. We have shown that async-http has a great model for both uni-directional and bi-directional streaming which works on both the client and server side. Such systems can be deployed into production using falcon.