Class: Faye::Engine::Memory

Inherits:
Base
  • Object
show all
Defined in:
lib/faye/engines/memory.rb

Constant Summary

Constants included from Logging

Logging::DEFAULT_LOG_LEVEL, Logging::LOG_LEVELS

Instance Attribute Summary

Attributes inherited from Base

#interval, #timeout

Attributes included from Logging

#log_level

Instance Method Summary collapse

Methods inherited from Base

#close_connection, #connect, #connection, #flush

Methods included from Timeouts

#add_timeout, #remove_timeout

Methods included from Logging

#log

Constructor Details

#initialize(options) ⇒ Memory

Returns a new instance of Memory.



5
6
7
8
9
10
11
 
# File 'lib/faye/engines/memory.rb', line 5

def initialize(options)
  @namespace = Namespace.new
  @clients   = {}
  @channels  = {}
  @messages  = {}
  super
end

Instance Method Details

#client_exists(client_id, &callback) ⇒ Object 



34
35
36
 
# File 'lib/faye/engines/memory.rb', line 34

def client_exists(client_id, &callback)
  callback.call(@namespace.exists?(client_id))
end

#create_client(&callback) ⇒ Object 



13
14
15
16
17
18
 
# File 'lib/faye/engines/memory.rb', line 13

def create_client(&callback)
  client_id = @namespace.generate
  debug 'Created new client ?', client_id
  ping(client_id)
  callback.call(client_id)
end

#destroy_client(client_id, &callback) ⇒ Object 



20
21
22
23
24
25
26
27
28
29
30
31
32
 
# File 'lib/faye/engines/memory.rb', line 20

def destroy_client(client_id, &callback)
  return unless @namespace.exists?(client_id)
  
  if @clients.has_key?(client_id)
    @clients[client_id].each { |channel| unsubscribe(client_id, channel) }
  end
  
  remove_timeout(client_id)
  @namespace.release(client_id)
  @messages.delete(client_id)
  debug 'Destroyed client ?', client_id
  callback.call if callback
end

#ping(client_id) ⇒ Object 



38
39
40
41
42
43
44
 
# File 'lib/faye/engines/memory.rb', line 38

def ping(client_id)
  timeout = @options[:timeout]
  return unless Numeric === timeout
  debug 'Ping ?, ?', client_id, timeout
  remove_timeout(client_id)
  add_timeout(client_id, 2 * timeout) { destroy_client(client_id) }
end

#publish(message) ⇒ Object 



72
73
74
75
76
77
78
79
80
81
82
83
84
 
# File 'lib/faye/engines/memory.rb', line 72

def publish(message)
  debug 'Publishing message ?', message
  channels = Channel.expand(message['channel'])
  channels.each do |channel|
    next unless clients = @channels[channel]
    clients.each do |client_id|
      debug 'Queueing for client ?: ?', client_id, message
      @messages[client_id] ||= Set.new
      @messages[client_id].add(message)
      empty_queue(client_id)
    end
  end
end

#subscribe(client_id, channel, &callback) ⇒ Object 



46
47
48
49
50
51
52
53
54
55
 
# File 'lib/faye/engines/memory.rb', line 46

def subscribe(client_id, channel, &callback)
  @clients[client_id] ||= Set.new
  @clients[client_id].add(channel)
  
  @channels[channel] ||= Set.new
  @channels[channel].add(client_id)
  
  debug 'Subscribed client ? to channel ?', client_id, channel
  callback.call(true) if callback
end

#unsubscribe(client_id, channel, &callback) ⇒ Object 



57
58
59
60
61
62
63
64
65
66
67
68
69
70
 
# File 'lib/faye/engines/memory.rb', line 57

def unsubscribe(client_id, channel, &callback)
  if @clients.has_key?(client_id)
    @clients[client_id].delete(channel)
    @clients.delete(client_id) if @clients[client_id].empty?
  end
  
  if @channels.has_key?(channel)
    @channels[channel].delete(client_id)
    @channels.delete(channel) if @channels[channel].empty?
  end
  
  debug 'Unsubscribed client ? from channel ?', client_id, channel
  callback.call(true) if callback
end