Class: EventMachine::Channel

Inherits:
Object
  • Object
show all
Defined in:
lib/em/channel.rb

Overview

Provides a simple thread-safe way to transfer data between (typically) long running tasks in defer and event loop thread.

Examples:


channel = EventMachine::Channel.new
sid     = channel.subscribe { |msg| p [:got, msg] }

channel.push('hello world')
channel.unsubscribe(sid)

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Channel

Returns a new instance of Channel.



15
16
17
18
 
# File 'lib/em/channel.rb', line 15

def initialize
  @subs = {}
  @uid  = 0
end

Instance Method Details

#num_subscribers ⇒ Object

Return the number of current subscribers.



21
22
23
 
# File 'lib/em/channel.rb', line 21

def num_subscribers
  return @subs.size
end

#pop(*a, &b) ⇒ Object

Fetches one message from the channel.



53
54
55
56
57
58
59
60
 
# File 'lib/em/channel.rb', line 53

def pop(*a, &b)
  EM.schedule {
    name = subscribe do |*args|
      unsubscribe(name)
      EM::Callback(*a, &b).call(*args)
    end
  }
end

#push(*items) ⇒ Object Also known as: <<

Add items to the channel, which are pushed out to all subscribers.



46
47
48
49
 
# File 'lib/em/channel.rb', line 46

def push(*items)
  items = items.dup
  EM.schedule { items.each { |i| @subs.values.each { |s| s.call i } } }
end

#subscribe(*a, &b) ⇒ Integer

Takes any arguments suitable for EM::Callback() and returns a subscriber id for use when unsubscribing.

Returns:

  • (Integer) —

    Subscribe identifier

See Also:



30
31
32
33
34
35
 
# File 'lib/em/channel.rb', line 30

def subscribe(*a, &b)
  name = gen_id
  EM.schedule { @subs[name] = EM::Callback(*a, &b) }

  name
end

#unsubscribe(name) ⇒ Object

Removes subscriber from the list.

Parameters:

  • Subscriber (Integer) —

    identifier

See Also:



41
42
43
 
# File 'lib/em/channel.rb', line 41

def unsubscribe(name)
  EM.schedule { @subs.delete name }
end