Class: Going::Channel

Inherits:
Object
  • Object
show all
Extended by:
Forwardable
Defined in:
lib/going/channel.rb

Overview

This class represents message channels of specified capacity. The push operation may be blocked if the capacity is full. The pop operation may be blocked if no messages have been sent.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(capacity = 0) {|_self| ... } ⇒ Channel

Creates a fixed-length channel with a capacity of capacity.

Yields:

  • (_self)

Yield Parameters:



15
16
17
18
19
20
21
22
23
24
 
# File 'lib/going/channel.rb', line 15

def initialize(capacity = 0)
  fail ArgumentError, 'channel capacity must be 0 or greater' unless capacity >= 0
  @capacity = capacity
  @closed = false
  @mutex = Mutex.new
  @push_semaphore = ConditionVariable.new
  @pop_semaphore = ConditionVariable.new

  yield self if block_given?
end

Instance Attribute Details

#capacity ⇒ Object (readonly)

Returns the capacity of the channel.



29
30
31
 
# File 'lib/going/channel.rb', line 29

def capacity
  @capacity
end

Instance Method Details

#close ⇒ Object

Closes the channel. Any data in the buffer may still be retrieved.



41
42
43
44
45
46
47
48
 
# File 'lib/going/channel.rb', line 41

def close
  synchronize do
    return false if closed?
    @messages = messages.first(capacity)
    broadcast_close
    @closed = true
  end
end

#closed? ⇒ Boolean

Returns whether or not the channel is closed.



34
35
36
 
# File 'lib/going/channel.rb', line 34

def closed?
  @closed
end

#inspect ⇒ Object 



101
102
103
104
105
106
 
# File 'lib/going/channel.rb', line 101

def inspect
  inspection = [:capacity, :messages].map do |attr|
    "#{attr}: #{send(attr).inspect}"
  end
  "#<#{self.class} #{inspection.join(', ')}>"
end

#pop ⇒ Object Also known as: receive, next

Receives data from the channel. If the channel is already empty, waits until a thread pushes to it.



75
76
77
78
79
80
81
82
83
 
# File 'lib/going/channel.rb', line 75

def pop
  synchronize do
    return if closed?
    wait_for_push if messages.empty?
    signal_pop
    check_for_close
    messages.shift
  end
end

#push(obj) ⇒ Object Also known as: <<, yield

Pushes obj to the channel. If the channel is already full, waits until a thread pops from it.



54
55
56
57
58
59
60
61
62
63
 
# File 'lib/going/channel.rb', line 54

def push(obj)
  synchronize do
    fail 'cannot push to a closed channel' if closed?
    messages.push obj
    signal_push
    wait_for_pop if messages.length > capacity
    check_for_close
    self
  end
end