Class: Ruote::Redis::Storage

Inherits:
Object
  • Object
show all
Includes:
StorageBase
Defined in:
lib/ruote/redis/storage.rb

Overview

A Redis storage for ruote.

The constructor accepts two arguments, the first one is a Redis instance ( see github.com/ezmobius/redis-rb ), the second one is the classic ruote engine options( see ruote.rubyforge.org/configuration.html#engine )

require 'redis' # gem install redis
require 'ruote' # gem install ruote
require 'ruote-redis' # gem install ruote-redis

engine = Ruote::Engine.new(
  Ruote::Worker.new(
    Ruote::Redis::RedisStorage.new(
      ::Redis.new(:db => 14, :thread_safe => true), {})))

em-redis

Not tried, but I guess, that substituting an instance of em-redis for the redis instance passed to the constructor might work. github.com/madsimian/em-redis

If you try and it works, feedback is welcome groups.google.com/group/openwferu-users

Direct Known Subclasses

RedisStorage

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(redis, options = {}) ⇒ Storage

A Redis storage for ruote.



71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/ruote/redis/storage.rb', line 71

def initialize(redis, options={})

  @redis = redis
  @options = options

  def @redis.keys_to_a(opt)
    keys(opt) rescue []
  end

  put_configuration
end

Instance Attribute Details

#redis ⇒ Object (readonly)

Returns the value of attribute redis.



67
68
69
 
# File 'lib/ruote/redis/storage.rb', line 67

def redis
  @redis
end

Class Method Details

.keys_to_a(opt) ⇒ Object 



76
77
78
 
# File 'lib/ruote/redis/storage.rb', line 76

def @redis.keys_to_a(opt)
  keys(opt) rescue []
end

Instance Method Details

#add_type(type) ⇒ Object

Mainly used by ruote’s test/unit/ut_17_storage.rb



286
287
288
289
 
# File 'lib/ruote/redis/storage.rb', line 286

def add_type(type)

  # nothing to be done
end

#close ⇒ Object

Shuts this worker down.

(This close / shutdown dichotomy has to be resolved at some point…)



270
271
272
273
 
# File 'lib/ruote/redis/storage.rb', line 270

def close

  @redis.quit
end

#delete(doc) ⇒ Object

Raises:

  • (ArgumentError) 


158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
 
# File 'lib/ruote/redis/storage.rb', line 158

def delete(doc)

  rev = doc['_rev']

  raise ArgumentError.new("can't delete doc without _rev") unless rev

  key = key_for(doc)

  lock(key) do

    current_doc = do_get(key)

    if current_doc.nil?
      #
      # document is [already] gone, delete fails (return true)
      #
      true

    elsif current_doc['_rev'] != rev
      #
      # version in storage doesn't match version to delete
      # (return version in storage)
      #
      current_doc

    else
      #
      # delete is successful (return nil)
      #
      @redis.del(key)

      nil
    end
  end
end

#delete_schedule(schedule_id) ⇒ Object 



110
111
112
113
114
115
 
# File 'lib/ruote/redis/storage.rb', line 110

def delete_schedule(schedule_id)

  return unless schedule_id

  @redis.del(key_for('schedules', schedule_id))
end

#dump(type) ⇒ Object

Returns a String containing a representation of the current content of in this Redis storage.



261
262
263
264
 
# File 'lib/ruote/redis/storage.rb', line 261

def dump(type)

  @redis.keys_to_a("#{type}/*").sort.join("\n")
end

#get(type, key) ⇒ Object 



153
154
155
156
 
# File 'lib/ruote/redis/storage.rb', line 153

def get(type, key)

  do_get(key_for(type, key))
end

#get_many(type, key = nil, opts = {}) ⇒ Object 



194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
 
# File 'lib/ruote/redis/storage.rb', line 194

def get_many(type, key=nil, opts={})

  keys = key ? Array(key) : nil

  #ids = if type == 'msgs' || type == 'schedules'
  #  @redis.keys_to_a("#{type}/*")

  ids = if keys == nil

    @redis.keys_to_a("#{type}/*")

  elsif keys.first.is_a?(String)

    keys.collect { |k| @redis.keys_to_a("#{type}/*!#{k}") }.flatten

  else #if keys.first.is_a?(Regexp)

    @redis.keys_to_a("#{type}/*").select { |i|

      i = i[type.length + 1..-1]
        # removing "^type/"

      keys.find { |k| k.match(i) }
    }
  end

  ids = ids.reject { |i| i.match(LOCK_KEY) }
  ids = ids.sort
  ids = ids.reverse if opts[:descending]

  skip = opts[:skip] || 0
  limit = opts[:limit] || ids.length
  ids = ids[skip, limit]

  docs = ids.length > 0 ? @redis.mget(*ids) : []
  docs = docs.inject({}) do |h, doc|
    if doc
      doc = Rufus::Json.decode(doc)
      h[doc['_id']] = doc
    end
    h
  end

  return docs.size if opts[:count]

  docs = docs.values.sort_by { |d| d['_id'] }

  opts[:descending] ? docs.reverse : docs
end

#ids(type) ⇒ Object 



244
245
246
247
248
249
250
251
 
# File 'lib/ruote/redis/storage.rb', line 244

def ids(type)

  @redis.keys_to_a("#{type}/*").reject { |i|
    i.match(LOCK_KEY)
  }.collect { |i|
    i.split('/').last
  }.sort
end

#purge! ⇒ Object 



253
254
255
256
 
# File 'lib/ruote/redis/storage.rb', line 253

def purge!

  @redis.keys_to_a('*').each { |k| @redis.del(k) }
end

#purge_type!(type) ⇒ Object

Nukes a db type and reputs it(losing all the documents that were in it).



293
294
295
296
 
# File 'lib/ruote/redis/storage.rb', line 293

def purge_type!(type)

  @redis.keys_to_a("#{type}/*").each { |k| (@redis.del(k) rescue nil) }
end

#put(doc, opts = {}) ⇒ Object 



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
145
146
147
148
149
150
151
 
# File 'lib/ruote/redis/storage.rb', line 117

def put(doc, opts={})

  key = key_for(doc)
  rev = doc['_rev']

  lock(key) do

    current_doc = do_get(key)
    current_rev = current_doc ? current_doc['_rev'] : nil

    if current_rev && rev != current_rev
      #
      # version in storage is newer than version being put,
      # (eturn version in storage)
      #
      current_doc

    elsif rev && current_rev.nil?
      #
      # document deleted, put fails (return true)
      #
      true

    else
      #
      # put is successful (return nil)
      #
      nrev = (rev.to_i + 1).to_s
      @redis.set(key, to_json(doc.merge('_rev' => nrev)))
      doc['_rev'] = nrev if opts[:update_rev]

      nil
    end
  end
end

#put_msg(action, options) ⇒ Object 



90
91
92
93
94
95
96
97
 
# File 'lib/ruote/redis/storage.rb', line 90

def put_msg(action, options)

  doc = prepare_msg_doc(action, options)

  @redis.set(key_for(doc), to_json(doc))

  nil
end

#put_schedule(flavour, owner_fei, s, msg) ⇒ Object 



99
100
101
102
103
104
105
106
107
108
 
# File 'lib/ruote/redis/storage.rb', line 99

def put_schedule(flavour, owner_fei, s, msg)

  doc = prepare_schedule_doc(flavour, owner_fei, s, msg)

  return nil unless doc

  @redis.set(key_for(doc), to_json(doc))

  doc['_id']
end

#reserve(doc) ⇒ Object

Returns true if the doc is successfully deleted.



85
86
87
88
 
# File 'lib/ruote/redis/storage.rb', line 85

def reserve(doc)

  (@redis.del(key_for(doc)) == 1)
end

#shutdown ⇒ Object

Shuts this worker down.

(This close / shutdown dichotomy has to be resolved at some point…)



279
280
281
282
 
# File 'lib/ruote/redis/storage.rb', line 279

def shutdown

  @redis.quit
end