Class: Ruote::Redis::RedisStorage

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

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

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

A Redis storage for ruote.



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

def initialize(redis, options={})

  @redis = redis
  @options = options

  def @redis.keys_to_a(opt)
    r = keys(opt)
    r.is_a?(Array) ? r : r.split(' ')
  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
79
 
# File 'lib/ruote/redis/storage.rb', line 76

def @redis.keys_to_a(opt)
  r = keys(opt)
  r.is_a?(Array) ? r : r.split(' ')
end

Instance Method Details

#add_type(type) ⇒ Object

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



268
269
 
# File 'lib/ruote/redis/storage.rb', line 268

def add_type(type)
end

#close ⇒ Object 



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

def close

  @redis.quit
end

#delete(doc) ⇒ Object

Raises:

  • (ArgumentError) 


157
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
 
# File 'lib/ruote/redis/storage.rb', line 157

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 



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

def delete_schedule(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.



256
257
258
259
 
# File 'lib/ruote/redis/storage.rb', line 256

def dump(type)

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

#get(type, key) ⇒ Object 



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

def get(type, key)

  do_get(key_for(type, key))
end

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



193
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
 
# File 'lib/ruote/redis/storage.rb', line 193

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

  opts[:count] ? docs.size : docs.values
end

#ids(type) ⇒ Object 



239
240
241
242
243
244
245
246
 
# File 'lib/ruote/redis/storage.rb', line 239

def ids(type)

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

#purge! ⇒ Object 



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

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).



273
274
275
276
 
# File 'lib/ruote/redis/storage.rb', line 273

def purge_type!(type)

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

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



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

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 



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

def put_msg(action, options)

  doc = prepare_msg_doc(action, options)

  puts "XXX" if @redis.nil?

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

  nil
end

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



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

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 



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

def reserve(doc)

  @redis.del(key_for(doc))
end