Module: Rbgo::Channel::Chan
Class Method Summary collapse
- .after(seconds, &blk) ⇒ Object
- .new(max = 0) ⇒ Object
- .perform(timeout: nil, &blk) ⇒ Object
- .tick(every_seconds, &blk) ⇒ Object
Instance Method Summary collapse
Class Method Details
.after(seconds, &blk) ⇒ Object
33 34 35 36 37 38 39 40 41 42 43 44 45 |
# File 'lib/rbgo/select_chan.rb', line 33 def self.after(seconds, &blk) ch = new CoRun::Routine.new(new_thread: true, queue_tag: :none) do begin sleep seconds v = blk.nil? ? Time.now : blk.call ch << v rescue nil ensure ch.close end end ch end |
.new(max = 0) ⇒ Object
24 25 26 27 28 29 30 31 |
# File 'lib/rbgo/select_chan.rb', line 24 def self.new(max = 0) max = max.to_i if max <= 0 NonBufferChan.new else BufferChan.new(max) end end |
.perform(timeout: nil, &blk) ⇒ Object
67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 |
# File 'lib/rbgo/select_chan.rb', line 67 def self.perform(timeout: nil, &blk) ch = new CoRun::Routine.new(new_thread: false, queue_tag: :default) do begin res = nil Timeout::timeout(timeout) do res = blk.call end ch << res rescue nil rescue Timeout::Error ensure ch.close end end ch end |
.tick(every_seconds, &blk) ⇒ Object
47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 |
# File 'lib/rbgo/select_chan.rb', line 47 def self.tick(every_seconds, &blk) ch = new CoRun::Routine.new(new_thread: true, queue_tag: :none) do begin loop do break if ch.closed? sleep every_seconds v = blk.nil? ? Time.now : blk.call begin ch.enq(v, true) rescue ThreadError end end ensure ch.close end end ch end |
Instance Method Details
#each ⇒ Object
9 10 11 12 13 14 15 16 17 18 19 20 21 22 |
# File 'lib/rbgo/select_chan.rb', line 9 def each if block_given? loop do obj, ok = pop if ok yield obj else break end end else enum_for(:each) end end |