Class: JobWorkflow::QueueAdapters::NullAdapter

Inherits:
Abstract
  • Object
show all
Defined in:
lib/job_workflow/queue_adapters/null_adapter.rb,
sig/generated/job_workflow/queue_adapters/null_adapter.rbs

Overview

rubocop:disable Naming/PredicateMethod

Instance Method Summary collapse

Methods inherited from Abstract

#persist_job_context

Constructor Details

#initialize ⇒ NullAdapter

: () -> void



15
16
17
18
19
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 15

def initialize
  @paused_queues = Set.new #: Set[String]
  @queue_jobs = {} #: Hash[String, Array[untyped]]
  @stored_jobs = {} #: Hash[String, Hash[String, untyped]]
end

Instance Method Details

#clear_queue(queue_name) ⇒ Boolean

: (String) -> bool

Parameters:

  • (String)

Returns:

  • (Boolean) 


87
88
89
90
91
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 87

def clear_queue(queue_name)
  empty_jobs = [] #: Array[untyped]
  @queue_jobs[queue_name] = empty_jobs
  true
end

#enqueue_test_job(queue_name, job) ⇒ void

Note:

Test helpers

This method returns an undefined value.

: (String, untyped) -> void

Parameters:

  • (String)
  • (Object) 


122
123
124
125
126
127
128
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 122

def enqueue_test_job(queue_name, job)
  unless @queue_jobs.key?(queue_name)
    empty_jobs = [] #: Array[untyped]
    @queue_jobs[queue_name] = empty_jobs
  end
  @queue_jobs[queue_name] << job
end

#fetch_job_contexts(_job_ids) ⇒ Array[Hash[String, untyped]]

: (Array) -> Array[Hash[String, untyped]]

Parameters:

  • (Array[String])

Returns:

  • (Array[Hash[String, untyped]]) 


110
111
112
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 110

def fetch_job_contexts(_job_ids)
  []
end

#fetch_job_statuses(_job_ids) ⇒ Hash[String, untyped]

: (Array) -> Hash[String, untyped]

Parameters:

  • (Array[String])

Returns:

  • (Hash[String, untyped]) 


37
38
39
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 37

def fetch_job_statuses(_job_ids)
  {}
end

#fetch_root_workflow_job_page(job_class_name:, limit:, cursor:) ⇒ Hash[Symbol, untyped]

: (job_class_name: String, limit: Integer, cursor: String?) -> Hash[Symbol, untyped]

Parameters:

  • job_class_name: (String)
  • limit: (Integer)
  • cursor: (String, nil)

Returns:

  • (Hash[Symbol, untyped]) 


99
100
101
102
103
104
105
106
107
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 99

def fetch_root_workflow_job_page(job_class_name:, limit:, cursor:)
  jobs = @stored_jobs.values.select { |job| job["class_name"] == job_class_name }.reverse
  offset = cursor.nil? ? 0 : cursor.to_i
  page_jobs = jobs.slice(offset, limit + 1) || []
  {
    jobs: page_jobs.first(limit),
    next_cursor: page_jobs.size > limit ? (offset + limit).to_s : nil
  }
end

#find_job(job_id) ⇒ Hash[String, untyped]?

: (String) -> Hash[String, untyped]?

Parameters:

  • (String)

Returns:

  • (Hash[String, untyped], nil) 


94
95
96
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 94

def find_job(job_id)
  @stored_jobs[job_id]
end

#initialize_adapter! ⇒ void

This method returns an undefined value.

: () -> void



12
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 12

def initialize_adapter!; end

#job_status(_job) ⇒ Symbol

: (untyped) -> Symbol

Parameters:

  • (Object)

Returns:

  • (Symbol) 


42
43
44
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 42

def job_status(_job)
  :pending
end

#pause_queue(queue_name) ⇒ Boolean

: (String) -> bool

Parameters:

  • (String)

Returns:

  • (Boolean) 


52
53
54
55
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 52

def pause_queue(queue_name)
  @paused_queues.add(queue_name)
  true
end

#paused_queues ⇒ Array[String]

: () -> Array

Returns:

  • (Array[String]) 


69
70
71
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 69

def paused_queues
  @paused_queues.to_a
end

#queue_latency(_queue_name) ⇒ Integer?

: (String) -> Integer?

Parameters:

  • (String)

Returns:

  • (Integer, nil) 


74
75
76
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 74

def queue_latency(_queue_name)
  nil
end

#queue_paused?(queue_name) ⇒ Boolean

: (String) -> bool

Parameters:

  • (String)

Returns:

  • (Boolean) 


64
65
66
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 64

def queue_paused?(queue_name)
  @paused_queues.include?(queue_name)
end

#queue_size(queue_name) ⇒ Integer

: (String) -> Integer

Parameters:

  • (String)

Returns:

  • (Integer) 


79
80
81
82
83
84
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 79

def queue_size(queue_name)
  jobs = @queue_jobs[queue_name]
  return 0 if jobs.nil?

  jobs.size
end

#reschedule_job(_job, _wait) ⇒ Boolean

: (_JobInterface, Numeric) -> bool

Parameters:

  • (_JobInterface)
  • (Numeric)

Returns:

  • (Boolean) 


115
116
117
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 115

def reschedule_job(_job, _wait)
  false
end

#reset! ⇒ void

Note:

Test helpers

This method returns an undefined value.

: () -> void



140
141
142
143
144
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 140

def reset!
  @paused_queues.clear
  @queue_jobs.clear
  @stored_jobs.clear
end

#resume_queue(queue_name) ⇒ Boolean

: (String) -> bool

Parameters:

  • (String)

Returns:

  • (Boolean) 


58
59
60
61
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 58

def resume_queue(queue_name)
  @paused_queues.delete(queue_name)
  true
end

#semaphore_available? ⇒ Boolean

: () -> bool

Returns:

  • (Boolean) 


22
23
24
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 22

def semaphore_available?
  false
end

#semaphore_signal(_semaphore) ⇒ Boolean

: (Semaphore) -> bool

Parameters:

Returns:

  • (Boolean) 


32
33
34
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 32

def semaphore_signal(_semaphore)
  true
end

#semaphore_wait(_semaphore) ⇒ Boolean

: (Semaphore) -> bool

Parameters:

Returns:

  • (Boolean) 


27
28
29
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 27

def semaphore_wait(_semaphore)
  true
end

#store_job(job_id, job_data) ⇒ void

Note:

Test helpers - stores job data for find_job

This method returns an undefined value.

: (String, Hash[String, untyped]) -> void

Parameters:

  • (String)
  • (Hash[String, untyped]) 


133
134
135
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 133

def store_job(job_id, job_data)
  @stored_jobs[job_id] = job_data
end

#supports_concurrency_limits? ⇒ Boolean

: () -> bool

Returns:

  • (Boolean) 


47
48
49
 
# File 'lib/job_workflow/queue_adapters/null_adapter.rb', line 47

def supports_concurrency_limits?
  false
end