Class: Whoosh::Jobs::MemoryBackend

Inherits:
Object
  • Object
show all
Defined in:
lib/whoosh/jobs/memory_backend.rb

Instance Method Summary collapse

Constructor Details

#initialize ⇒ MemoryBackend

Returns a new instance of MemoryBackend.



6
7
8
9
10
11
12
 
# File 'lib/whoosh/jobs/memory_backend.rb', line 6

def initialize
  @queue = []
  @scheduled = []
  @records = {}
  @mutex = Mutex.new
  @cv = ConditionVariable.new
end

Instance Method Details

#find(id) ⇒ Object 



43
44
45
 
# File 'lib/whoosh/jobs/memory_backend.rb', line 43

def find(id)
  @mutex.synchronize { @records[id]&.dup }
end

#pending_count ⇒ Object 



51
52
53
 
# File 'lib/whoosh/jobs/memory_backend.rb', line 51

def pending_count
  @mutex.synchronize { @queue.size }
end

#pop(timeout: 5) ⇒ Object 



26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/whoosh/jobs/memory_backend.rb', line 26

def pop(timeout: 5)
  @mutex.synchronize do
    # Promote scheduled jobs that are ready
    promote_scheduled

    if @queue.empty?
      @cv.wait(@mutex, timeout)
      promote_scheduled
    end
    @queue.shift
  end
end

#push(job_data) ⇒ Object 



14
15
16
17
18
19
20
21
22
23
24
 
# File 'lib/whoosh/jobs/memory_backend.rb', line 14

def push(job_data)
  @mutex.synchronize do
    if job_data[:run_at] && job_data[:run_at] > Time.now.to_f
      @scheduled << job_data
      @scheduled.sort_by! { |j| j[:run_at] }
    else
      @queue << job_data
    end
    @cv.signal
  end
end

#save(record) ⇒ Object 



39
40
41
 
# File 'lib/whoosh/jobs/memory_backend.rb', line 39

def save(record)
  @mutex.synchronize { @records[record[:id]] = record }
end

#scheduled_count ⇒ Object 



55
56
57
 
# File 'lib/whoosh/jobs/memory_backend.rb', line 55

def scheduled_count
  @mutex.synchronize { @scheduled.size }
end

#shutdown ⇒ Object 



59
60
61
 
# File 'lib/whoosh/jobs/memory_backend.rb', line 59

def shutdown
  @mutex.synchronize { @cv.broadcast }
end

#size ⇒ Object 



47
48
49
 
# File 'lib/whoosh/jobs/memory_backend.rb', line 47

def size
  @mutex.synchronize { @queue.size + @scheduled.size }
end