Class: Starling

Inherits:
MemCache
  • Object
show all
Defined in:
lib/starling.rb

Constant Summary collapse

WAIT_TIME = 
0.25

Instance Method Summary collapse

Instance Method Details

#_original_delete ⇒ Object 



7
 
# File 'lib/starling.rb', line 7

alias_method :_original_delete, :delete

#_original_get ⇒ Object 



6
 
# File 'lib/starling.rb', line 6

alias_method :_original_get, :get

#available_queues(statistics = nil) ⇒ Object

returns a list of available (currently allocated) queues.



91
92
93
94
95
96
97
98
99
100
101
 
# File 'lib/starling.rb', line 91

def available_queues(statistics = nil)
  statistics ||= stats

  statistics.map { |k,v|
    v.keys
  }.flatten.uniq.grep(/^queue_(.*)_items/).map { |v|
    v.gsub(/^queue_/, '').gsub(/_items$/, '')
  }.reject { |v|
    v =~ /_total$/ || v =~ /_expired$/
  }
end

#delete(key, expiry = 0) ⇒ Object

Delete the key (queue) from all Starling servers. This is necessary because the random way a server is chosen in #get_server_for_key implies that the queue could easily be spread across the entire Starling cluster.

Raises:

  • (MemCacheError) 


33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/starling.rb', line 33

def delete(key, expiry = 0)
  raise MemCacheError, "Update of readonly cache" if @readonly
  cache_key = make_cache_key key
 
  @buckets.each do |server|
    with_socket_management(server) do |socket|
      socket.write "delete #{cache_key} #{expiry}\r\n"
      socket.gets
    end
  end
end

#fetch(*args) ⇒ Object

will return the next item or nil



23
24
25
 
# File 'lib/starling.rb', line 23

def fetch(*args)
  _original_get(*args)
end

#flush(queue) ⇒ Object

iterator to flush queue. Each element will be passed to the provided block



107
108
109
110
111
112
 
# File 'lib/starling.rb', line 107

def flush(queue)
  sizeof(queue).times do
    v = get(queue)
    yield v if block_given?
  end
end

#get(*args) ⇒ Object

fetch an item from a queue.



12
13
14
15
16
17
18
 
# File 'lib/starling.rb', line 12

def get(*args)
  loop do
    response = _original_get(*args)
    return response unless response.nil?
    sleep WAIT_TIME
  end
end

#set(queue, value, expiry = 0, raw = false) ⇒ Object

insert value into queue.

expiry is expressed as a UNIX timestamp

If raw is true, value will not be Marshalled. If raw = :yaml, value will be serialized with YAML, instead.



53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/starling.rb', line 53

def set(queue, value, expiry = 0, raw = false)
  retries = 0
  begin
    if raw == :yaml
      value = YAML.dump(value)
      raw = true
    end

    super(queue, value, expiry, raw)
  rescue MemCache::MemCacheError => e
    retries += 1
    sleep WAIT_TIME
    retry unless retries > 3
    raise e
  end
end

#sizeof(queue, statistics = nil) ⇒ Object

returns the number of items in queue. If queue is :all, a hash of all queue sizes will be returned.



74
75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/starling.rb', line 74

def sizeof(queue, statistics = nil)
  statistics ||= stats

  if queue == :all
    queue_sizes = {}
    available_queues(statistics).each do |queue|
      queue_sizes[queue] = sizeof(queue, statistics)
    end
    return queue_sizes
  end

  statistics.inject(0) { |m,(k,v)| m + v["queue_#{queue}_items"].to_i }
end