Class: Orchestra::Subscriber

Inherits:
Object
  • Object
show all
Defined in:
lib/orchestra/subscriber.rb

Instance Method Summary collapse

Constructor Details

#initialize(publisher) ⇒ Subscriber

Returns a new instance of Subscriber.



3
4
5
6
 
# File 'lib/orchestra/subscriber.rb', line 3

def initialize(publisher)
  @publisher, @stream = publisher, []
  @publisher.register_subscriber(self)
end

Instance Method Details

#consume ⇒ Object 



26
27
28
29
30
31
32
 
# File 'lib/orchestra/subscriber.rb', line 26

def consume
  Thread.new do
    loop {
      (event = self.next) ? (yield event) : wait
    }
  end
end

#disconnect! ⇒ Object 



8
9
10
 
# File 'lib/orchestra/subscriber.rb', line 8

def disconnect!
  @publisher.unregister_subscriber(self)
end

#next ⇒ Object 



22
23
24
 
# File 'lib/orchestra/subscriber.rb', line 22

def next
  @stream.shift
end

#push(event) ⇒ Object 



18
19
20
 
# File 'lib/orchestra/subscriber.rb', line 18

def push(event)
  @stream.push(event)
end

#wait ⇒ Object 



12
13
14
15
16
 
# File 'lib/orchestra/subscriber.rb', line 12

def wait
  @publisher.mutex.synchronize do
    @publisher.signaler.wait(@publisher.mutex)
  end
end