1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
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
|
require 'concurrent/maybe'
require 'concurrent/channel/selector/after_clause'
require 'concurrent/channel/selector/default_clause'
require 'concurrent/channel/selector/put_clause'
require 'concurrent/channel/selector/take_clause'
module Concurrent
class Channel
# @!visibility private
class Selector
def initialize
@clauses = []
@error_handler = nil
end
def case(channel, action, message = nil, &block)
if [:take, :poll, :receive, :~].include?(action)
take(channel, &block)
elsif [:put, :offer, :send, :<<].include?(action)
put(channel, message, &block)
else
raise ArgumentError.new('invalid action')
end
end
def take(channel, &block)
raise ArgumentError.new('no block given') unless block_given?
@clauses << TakeClause.new(channel, block)
end
alias_method :receive, :take
def put(channel, message, &block)
@clauses << PutClause.new(channel, message, block)
end
alias_method :send, :put
def after(seconds, &block)
@clauses << AfterClause.new(seconds, block)
end
alias_method :timeout, :after
def default(&block)
raise ArgumentError.new('no block given') unless block_given?
@clauses << DefaultClause.new(block)
end
def error(&block)
raise ArgumentError.new('no block given') unless block_given?
raise ArgumentError.new('only one error handler allowed') if @error_handler
@error_handler = block
end
def execute
raise Channel::Error.new('no clauses given') if @clauses.empty?
loop do
done = @clauses.each do |clause|
result = clause.execute
break result if result.just?
end
break done.value if done.is_a?(Concurrent::Maybe)
Thread.pass
end
rescue => ex
if @error_handler
@error_handler.call(ex)
else
raise ex
end
end
end
end
end
|
