Class: Async::Promise

Inherits:
Object
  • Object
show all
Defined in:
lib/async/promise.rb

Overview

A promise represents a value that will be available in the future. Unlike Condition, once resolved (or rejected), all future waits return immediately with the stored value or raise the stored exception.

This is thread-safe and integrates with the fiber scheduler.

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Promise

Create a new promise.



22
23
24
25
26
27
28
29
30
31
32
33
34
 
# File 'lib/async/promise.rb', line 22

def initialize
	# nil = pending, :completed = success, :failed = failure, :cancelled = cancelled:
	@resolved = nil
	
	# Stores either the result value or the exception:
	@value = nil
	
	# Track how many fibers are currently waiting:
	@waiting = 0
	
	@mutex = Mutex.new
	@condition = ConditionVariable.new
end

Class Method Details

.fulfill(promise, &block) ⇒ Object

If a promise is given, fulfill it with the result of the block. If no promise is given, simply yield to the block. This is useful for methods that may optionally take a promise to fulfill.



258
259
260
261
262
263
264
 
# File 'lib/async/promise.rb', line 258

def self.fulfill(promise, &block)
	if promise
		return promise.fulfill(&block)
	else
		return yield
	end
end

Instance Method Details

#cancel(exception = Cancel.new("Promise cancelled!")) ⇒ Object

Cancel the promise, indicating cancellation. All current and future waiters will receive nil. Can only be called on pending promises - no-op if already resolved.



214
215
216
217
218
219
220
221
222
223
224
225
226
227
 
# File 'lib/async/promise.rb', line 214

def cancel(exception = Cancel.new("Promise cancelled!"))
	@mutex.synchronize do
		# No-op if already in any final state
		return if @resolved
		
		@resolved = :cancelled
		@value = exception
		
		# Wake up all waiting fibers:
		@condition.broadcast
	end
	
	return nil
end

#cancelled? ⇒ Boolean

Returns:

  • (Boolean) 


48
49
50
 
# File 'lib/async/promise.rb', line 48

def cancelled?
	@mutex.synchronize{@resolved == :cancelled}
end

#completed? ⇒ Boolean

Returns:

  • (Boolean) 


58
59
60
 
# File 'lib/async/promise.rb', line 58

def completed?
	@mutex.synchronize{@resolved == :completed}
end

#failed? ⇒ Boolean

Returns:

  • (Boolean) 


53
54
55
 
# File 'lib/async/promise.rb', line 53

def failed?
	@mutex.synchronize{@resolved == :failed}
end

#fulfill(&block) ⇒ Object

Resolve the promise with the result of the block. If the block raises an exception, the promise will be rejected. If the promise was already resolved, the block will not be called.



234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
 
# File 'lib/async/promise.rb', line 234

def fulfill(&block)
	raise "Promise already resolved!" if @resolved
	
	begin
		return self.resolve(yield)
	rescue Cancel => exception
		return self.cancel(exception)
	rescue => error
		return self.reject(error)
	rescue Exception => exception
		self.reject(exception)
		raise
	ensure
		# Handle non-local exits (throw, etc.) that bypass normal flow:
		self.resolve(nil) unless @resolved
	end
end

#reject(exception) ⇒ Object

Reject the promise with an exception. All current and future waiters will receive this exception. Can only be called once - subsequent calls are ignored.



197
198
199
200
201
202
203
204
205
206
207
208
209
 
# File 'lib/async/promise.rb', line 197

def reject(exception)
	@mutex.synchronize do
		return if @resolved
		
		@resolved = :failed
		@value = exception
		
		# Wake up all waiting fibers:
		@condition.broadcast
	end
	
	return nil
end

#resolve(value) ⇒ Object

Resolve the promise with a value. All current and future waiters will receive this value. Can only be called once - subsequent calls are ignored.



178
179
180
181
182
183
184
185
186
187
188
189
190
 
# File 'lib/async/promise.rb', line 178

def resolve(value)
	@mutex.synchronize do
		return if @resolved
		
		@resolved = :completed
		@value = value
		
		# Wake up all waiting fibers:
		@condition.broadcast
	end
	
	return value
end

#resolved ⇒ Object 



43
44
45
 
# File 'lib/async/promise.rb', line 43

def resolved
	@mutex.synchronize{@resolved}
end

#resolved? ⇒ Boolean

Returns:

  • (Boolean) 


37
38
39
 
# File 'lib/async/promise.rb', line 37

def resolved?
	@mutex.synchronize{!!@resolved}
end

#suppress_warnings! ⇒ Object

Artificially mark that someone is waiting (useful for suppressing warnings).



69
70
71
 
# File 'lib/async/promise.rb', line 69

def suppress_warnings!
	@mutex.synchronize{@waiting += 1}
end

#value ⇒ Object

Non-blocking access to the current value. Returns nil if not yet resolved. Does not raise exceptions even if the promise was rejected or cancelled. For resolved promises, returns the raw stored value (result, exception, or cancel exception).



78
79
80
 
# File 'lib/async/promise.rb', line 78

def value
	@mutex.synchronize{@resolved ? @value : nil}
end

#wait ⇒ Object

Wait for the promise to be resolved and return the value.

If already resolved, returns immediately. If rejected, raises the stored exception.



160
161
162
163
164
165
166
167
168
169
170
171
 
# File 'lib/async/promise.rb', line 160

def wait(...)
	resolved = wait?(...)
	
	if resolved.nil?
		raise TimeoutError, "Timeout while waiting for promise!"
	elsif resolved == :completed
		return @value
	elsif @value
		# If we aren't completed, we should have an exception or cancel reason stored:
		raise @value
	end
end

#wait?(timeout: nil) ⇒ Boolean

Wait for the promise to be resolved (without raising exceptions).

If already resolved, returns immediately. Otherwise, waits until resolution or timeout.

Returns:

  • (Boolean) 


125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
 
# File 'lib/async/promise.rb', line 125

def wait?(timeout: nil)
	unless @resolved
		@mutex.synchronize do
			# Increment waiting count:
			@waiting += 1
			
			begin
				# Wait for resolution if not already resolved:
				unless @resolved
					if timeout.nil?
						wait_indefinitely
					else
						unless wait_with_timeout(timeout)
							# We don't want to race on @resolved after exiting the mutex:
							return nil
						end
					end
				end
			ensure
				# Decrement waiting count when done:
				@waiting -= 1
			end
		end
	end
	
	return @resolved
end

#waiting? ⇒ Boolean

Returns:

  • (Boolean) 


63
64
65
 
# File 'lib/async/promise.rb', line 63

def waiting?
	@mutex.synchronize{@waiting > 0}
end