Class: Execache
- Inherits:
-
Object
- Object
- Execache
- Defined in:
- lib/execache.rb,
lib/execache/client.rb
Defined Under Namespace
Classes: Client
Instance Method Summary collapse
-
#initialize(yaml) ⇒ Execache
constructor
A new instance of Execache.
- #shut_down ⇒ Object
Constructor Details
#initialize(yaml) ⇒ Execache
Returns a new instance of Execache.
17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 |
# File 'lib/execache.rb', line 17 def initialize(yaml) = YAML.load(File.read(yaml)) parallel = ['parallel'] || 3 puts "\nStarting execache server (redis @ #{['redis']})..." redis = Redis.connect(:url => "redis://#{['redis']}") retries = 0 begin while true if request = redis.lpop('execache:request') Timeout.timeout(60) do request = Yajl::Parser.parse(request) # Options global_cache_key = request.delete('cache_key') channel = request.delete('channel') force = request.delete('force') ttl = request.delete('ttl') pending = false results = {} request.each do |cmd_type, | cache_keys = [] groups = [] # Binary + preliminary arguments command = [ [cmd_type]['command'], ['args'] ].join(' ') # For each argument group... ['groups'].each do |args| cache_key = Digest::SHA1.hexdigest( "#{global_cache_key || command} #{args}" ) cache_key = "execache:cache:#{cache_key}" cache = redis.get(cache_key) # If force cache overwrite || no cache || pending cache if force || !cache || cache == '[PENDING]' pending = true # Else, store cache result else results[cmd_type] ||= [] results[cmd_type] << Yajl::Parser.parse(cache) end # If force cache overwrite || no cache if force || !cache redis.set(cache_key, '[PENDING]') redis.expire(cache_key, 60) # Timeout incase execution fails cache_keys << cache_key groups << args end end # Add to command queue if commands present unless groups.empty? command = { :cache_keys => cache_keys, :cmd_type => cmd_type, :command => command, :groups => groups, :ttl => ttl } redis.rpush("execache:commands", Yajl::Encoder.encode(command)) end end redis.publish( "execache:response:#{channel}", pending ? '[PENDING]' : Yajl::Encoder.encode(results) ) end end # Execute queued commands if redis.get("execache:parallel").to_i <= parallel && cmd = redis.lpop("execache:commands") redis.incr("execache:parallel") Thread.new do Timeout.timeout(60) do cmd = Yajl::Parser.parse(cmd) cache_keys = cmd['cache_keys'] cmd_type = cmd['cmd_type'] command = cmd['command'] groups = cmd['groups'] ttl = cmd['ttl'] separators = [cmd_type]['separators'] || {} separators['group'] ||= "[END]" separators['result'] ||= "\n" results = `#{command} #{groups.join(' ')}` results = results.split(separators['group'] + separators['result']) results = results.collect { |r| r.split(separators['result']) } redis.decr("execache:parallel") results.each_with_index do |result, i| redis.set( cache_keys[i], Yajl::Encoder.encode(result) ) redis.expire(cache_keys[i], ttl) if ttl end end end end sleep(1.0 / 1000.0) end rescue Interrupt shut_down rescue Exception => e puts "\nError: #{e.}" puts "\t#{e.backtrace.join("\n\t")}" retries += 1 shut_down if retries >= 10 retry end end |
Instance Method Details
#shut_down ⇒ Object
146 147 148 149 |
# File 'lib/execache.rb', line 146 def shut_down puts "\nShutting down execache server..." exit end |