Module: Protocol::HTTP::Executor::Worker

Defined in:
lib/protocol/http/executor/worker.rb

Overview

Executes one transported request in the isolated execution context.

Class Method Summary collapse

Class Method Details

.body_metadata(body) ⇒ Object

Describe a non-empty body for transport.



124
125
126
127
128
129
130
131
132
 
# File 'lib/protocol/http/executor/worker.rb', line 124

def self.body_metadata(body)
	return unless body
	return if body.empty?
	
	return {
		length: body.length,
		stream: body.stream?,
	}
end

.consume(response, request, endpoint) ⇒ Object

Consume the response body according to the caller's selected mode.



91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
 
# File 'lib/protocol/http/executor/worker.rb', line 91

def self.consume(response, request, endpoint)
	message = endpoint.control.read
	raise ClosedError unless message
	
	type, mode = message
	return if type == :cancel
	raise ClosedError, "Expected a consumption mode, but received #{type.inspect}!" unless type == :consume
	
	case mode
	when :read
		response.body.each{|chunk| endpoint.body.write(:chunk, chunk)}
	when :stream
		input = Body::StreamInput.new(endpoint.body, request.headers)
		output = Body::StreamOutput.new(endpoint.body)
		stream = ::Protocol::HTTP::Body::Stream.new(input, output)
		response.body.call(stream)
	else
		raise ArgumentError, "Unknown response consumption mode: #{mode.inspect}!"
	end
	
	trailers = response.headers.trailer.to_a
	endpoint.body.write(:trailers, trailers) unless trailers.empty?
	endpoint.body.close_write
rescue => error
	write_error(endpoint.body, error)
ensure
	response.close(error)
end

.execute(application, endpoint) ⇒ Object

Execute a request received on the endpoint.



34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/protocol/http/executor/worker.rb', line 34

def self.execute(application, endpoint)
	response_started = false
	message = endpoint.control.read
	raise ClosedError unless message
	
	type, description = message
	raise ClosedError, "Expected a request, but received #{type.inspect}!" unless type == :request
	
	request = Request.new(description, endpoint)
	response = application.call(request)
	response ||= ::Protocol::HTTP::Response[500]
	
	body = response.body
	body_metadata = body_metadata(body)
	trailers = nil
	
	unless body_metadata
		trailers = response.headers.trailer.to_a
		trailers = nil if trailers.empty?
	end
	
	endpoint.control.write(:response, {
		version: response.version,
		status: response.status,
		headers: response.headers.header.to_a,
		trailers: trailers,
		body: body_metadata,
		protocol: response.protocol,
	})
	response_started = true
	
	if body_metadata
		consume(response, request, endpoint)
	else
		response.close
	end
rescue => error
	if response_started
		write_error(endpoint.body, error)
	else
		write_error(endpoint.control, error)
	end
	
	begin
		response&.close(error)
	rescue
		# The original error has already been reported:
	end
ensure
	request&.close
end

.run(application, streams) ⇒ Object

Run the application using the given transport streams.



20
21
22
23
24
25
26
27
28
 
# File 'lib/protocol/http/executor/worker.rb', line 20

def self.run(application, streams)
	endpoint = Transport::Endpoint.new(*streams)
	
	Sync do
		execute(application, endpoint)
	end
ensure
	endpoint&.close
end

.write_error(channel, error) ⇒ Object

Write an error unless the transport has already closed.



138
139
140
141
142
 
# File 'lib/protocol/http/executor/worker.rb', line 138

def self.write_error(channel, error)
	channel.write(:error, RemoteError.dump(error))
rescue ClosedError
	# The caller has already abandoned the execution:
end