Class: Protocol::HTTP::Executor::Execution

Inherits:
Object
  • Object
show all
Defined in:
lib/protocol/http/executor/execution.rb

Overview

Manages one request and its isolated worker transport.

Instance Method Summary collapse

Constructor Details

#initialize(endpoint, backend, request, parent) ⇒ Execution

Initialize an execution.



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

def initialize(endpoint, backend, request, parent)
	@endpoint = endpoint
	@backend = backend
	@request = request
	@parent = parent
	@input_task = nil
	@input_close_task = nil
	@finished = false
	@mutex = Mutex.new
end

Instance Method Details

#body ⇒ Object 



37
38
39
 
# File 'lib/protocol/http/executor/execution.rb', line 37

def body
	@endpoint.body
end

#call ⇒ Object

Start the worker request and wait for the response head.



44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
 
# File 'lib/protocol/http/executor/execution.rb', line 44

def call
	description = request_description(@request)
	control.write(:request, description)
	
	if description[:body]
		@input_task = @parent.async do
			forward_request_body(@request.body)
		end
	end
	
	wait_for_response
rescue
	cancel($!)
	raise
end

#cancel(error = nil) ⇒ Object

Cancel the execution and release its transport.



104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
 
# File 'lib/protocol/http/executor/execution.rb', line 104

def cancel(error = nil)
	return unless transition_to_finished
	
	begin
		control.write(:cancel, error && RemoteError.dump(error))
	rescue ClosedError
		# The worker has already finished:
	end
	
	@input_task&.cancel
	@input_close_task&.cancel
	@request.close(error)
	@endpoint.close
	return nil
end

#close_input ⇒ Object

Finish the request direction once the initial request body has been forwarded.



83
84
85
86
87
88
 
# File 'lib/protocol/http/executor/execution.rb', line 83

def close_input
	@input_close_task ||= @parent.async do
		@input_task&.wait
		body.close_write
	end
end

#control ⇒ Object 



32
33
34
 
# File 'lib/protocol/http/executor/execution.rb', line 32

def control
	@endpoint.control
end

#finish ⇒ Object

Finish the execution and release its transport.



91
92
93
94
95
96
97
98
99
 
# File 'lib/protocol/http/executor/execution.rb', line 91

def finish
	return unless transition_to_finished
	
	@input_task&.cancel
	@input_close_task&.cancel
	@endpoint.close
	@backend.join
	return nil
end

#stream_input(stream) ⇒ Object

Forward upgraded-stream input after the request body phase.



64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
 
# File 'lib/protocol/http/executor/execution.rb', line 64

def stream_input(stream)
	@parent.async do
		@input_task&.wait
		
		while chunk = read_stream_chunk(stream)
			body.write(:stream_chunk, chunk)
		end
	rescue => error
		begin
			body.write(:stream_error, RemoteError.dump(error))
		rescue ClosedError
			# The worker has already finished:
		end
	ensure
		body.close_write
	end
end

#The bidirectional body channel.=(bidirectionalbodychannel. = (value)) ⇒ Object 



37
38
39
 
# File 'lib/protocol/http/executor/execution.rb', line 37

def body
	@endpoint.body
end

#The control channel.=(controlchannel. = (value)) ⇒ Object 



32
33
34
 
# File 'lib/protocol/http/executor/execution.rb', line 32

def control
	@endpoint.control
end