class Polipus::QueueOverflow::MongoQueue
Public Class Methods
new(mongo_db, queue_name, options = {})
click to toggle source
# File lib/polipus/queue_overflow/mongo_queue.rb, line 7 def initialize(mongo_db, queue_name, options = {}) @mongo_db = mongo_db @collection_name = "polipus_q_overflow_#{queue_name}" @semaphore = Mutex.new @options = options @options[:ensure_uniq] ||= false @options[:ensure_uniq] && ensure_index end
Public Instance Methods
clear()
click to toggle source
# File lib/polipus/queue_overflow/mongo_queue.rb, line 24 def clear @mongo_db[@collection_name].drop @options[:ensure_uniq] && ensure_index end
empty?()
click to toggle source
# File lib/polipus/queue_overflow/mongo_queue.rb, line 20 def empty? !(length > 0) end
length()
click to toggle source
# File lib/polipus/queue_overflow/mongo_queue.rb, line 16 def length @mongo_db[@collection_name].find.count end
Also aliased as: size
pop(_ = false)
click to toggle source
# File lib/polipus/queue_overflow/mongo_queue.rb, line 38 def pop(_ = false) @semaphore.synchronize do doc = @mongo_db[@collection_name].find.sort(_id: 1).limit(1).first return nil if doc.nil? @mongo_db[@collection_name].find(_id: doc['_id']).delete_one doc && doc['payload'] ? doc['payload'] : nil end end
push(data)
click to toggle source
# File lib/polipus/queue_overflow/mongo_queue.rb, line 29 def push(data) if @options[:ensure_uniq] @mongo_db[@collection_name].find(payload: data).replace_one({ payload: data }, upsert: true) else @mongo_db[@collection_name].insert_one(payload: data) end true end
Protected Instance Methods
ensure_index()
click to toggle source
# File lib/polipus/queue_overflow/mongo_queue.rb, line 55 def ensure_index # @TODO: Drop dups option was removed. We may want to add something here to remove duplications @mongo_db[@collection_name].indexes.create_one({ payload: 1 }, background: true, unique: true) end