Module: Ductr::Store

Extended by:
JobStore, PipelineStore
Defined in:
lib/ductr/store.rb,
lib/ductr/store/job_store.rb,
lib/ductr/store/job_serializer.rb,
lib/ductr/store/pipeline_store.rb,
lib/ductr/store/pipeline_serializer.rb

Overview

Store interaction helpers for internal usage.

Defined Under Namespace

Modules: JobSerializer, JobStore, PipelineSerializer, PipelineStore

Constant Summary collapse

EXPIRATION_INTERVAL =

Returns The cache expiration of job’s status, default to one day.

Returns:

  • (Integer) —

    The cache expiration of job’s status, default to one day

86_400

Constants included from JobStore

JobStore::JOB_KEY_PREFIX, JobStore::JOB_REGISTRY_KEY

Constants included from PipelineStore

PipelineStore::PIPELINE_KEY_PREFIX, PipelineStore::PIPELINE_REGISTRY_KEY

Class Method Summary collapse

Methods included from JobStore

all_jobs, read_jobs, register_job, write_job

Methods included from JobSerializer

#serialize_job

Methods included from PipelineStore

all_pipelines, register_pipeline, write_pipeline

Methods included from PipelineSerializer

#serialize_pipeline

Class Method Details

.all(registry_key, key_prefix) ⇒ Array<Job>

Get all known job instances for the given registry_key and job’s key_prefix.

Parameters:

  • registry_key (String) —

    The registry key in which job keys will be read

  • key_prefix (String) —

    The cache key prefix for the registry’s job keys

Returns:

  • (Array<Job>) —

    The job instances



25
26
27
28
29
30
31
 
# File 'lib/ductr/store.rb', line 25

def all(registry_key, key_prefix)
  job_ids = Ductr.store.read(registry_key)
  return [] unless job_ids

  keys = job_ids.map { |job_id| "#{key_prefix}:#{job_id}" }
  Ductr.store.read_multi(*keys).values
end

.all_done? ⇒ Boolean

Determines whether all tracked jobs have either a completed or failed status.

Returns:

  • (Boolean) —

    ‘true` when all jobs are done



76
77
78
 
# File 'lib/ductr/store.rb', line 76

def all_done?
  [*all_jobs, *all_pipelines].all?(&:stopped?)
end

.read(key_prefix, *jobs) ⇒ Array<Job>

Read all given jobs in the given key_prefix.

Parameters:

  • key_prefix (String) —

    The cache key prefix for the job_id

  • *jobs (Array<Job>) —

    The jobs to read

Returns:

  • (Array<Job>) —

    The read jobs



41
42
43
44
 
# File 'lib/ductr/store.rb', line 41

def read(key_prefix, *jobs)
  keys = jobs.map { |job| "#{key_prefix}:#{job.job_id}" }
  Ductr.store.read_multi(*keys).values
end

.register(registry_key, job) ⇒ void

This method returns an undefined value.

Add the given job to the store’s job registry. This method is NOT thread-safe.

Parameters:

  • job (Job) —

    The job to register



64
65
66
67
68
69
 
# File 'lib/ductr/store.rb', line 64

def register(registry_key, job)
  job_ids = Ductr.store.read(registry_key) || Set.new

  job_ids.add(job.job_id)
  Ductr.store.write(registry_key, job_ids, expires_in: EXPIRATION_INTERVAL)
end

.write(key_prefix, job) ⇒ void

This method returns an undefined value.

Update the given job in the given key_prefix.

Parameters:

  • job (Job) —

    The job to update in the store



53
54
55
 
# File 'lib/ductr/store.rb', line 53

def write(key_prefix, job)
  Ductr.store.write("#{key_prefix}:#{job.job_id}", job, expires_in: EXPIRATION_INTERVAL)
end