Module: Cargosocket::StreamAdapters::EMChannelAdapter

Defined in:
lib/cargosocket/stream_adapters/e_m_channel_adapter.rb

Constant Summary collapse

CHANNELS = 
{}
DISPATCHERS = 
[]

Class Method Summary collapse

Class Method Details

.dispatch(to) ⇒ Object 



8
9
10
 
# File 'lib/cargosocket/stream_adapters/e_m_channel_adapter.rb', line 8

def self.dispatch(to)
  # DISPATCHERS << to
end

.push(*to, message) ⇒ Object 



12
13
14
15
16
17
18
19
20
 
# File 'lib/cargosocket/stream_adapters/e_m_channel_adapter.rb', line 12

def self.push(*to, message)
  to.each do |channel|
    if channel.is_a?(Symbol)
      CHANNELS[channel].push(message) if CHANNELS.has_key?(channel)
    else
      # channel.push(message)
    end
  end
end

.subscribe(*to) ⇒ Object 



22
23
24
25
26
27
28
29
 
# File 'lib/cargosocket/stream_adapters/e_m_channel_adapter.rb', line 22

def self.subscribe(*to)
  callback = Proc.new
  return to.reduce({}) do |acc, channel|
    CHANNELS[channel] ||= EM::Channel.new
    acc[channel] = CHANNELS[channel].subscribe(&callback.curry[channel])
    next acc
  end
end

.unsubscribe(from) ⇒ Object 



31
32
33
34
35
36
37
38
39
 
# File 'lib/cargosocket/stream_adapters/e_m_channel_adapter.rb', line 31

def self.unsubscribe(from)
  from.each do |channel, cid|
    yield(channel)
    CHANNELS[channel].unsubscribe(cid)
    if CHANNELS[channel].num_subscribers == DISPATCHERS.count
      CHANNELS.delete(channel)
    end
  end
end