Class: Ductr::Pipeline

Inherits:
Job
  • Object
show all
Defined in:
lib/ductr/pipeline.rb

Overview

Pipelines allows to easily declare rich data pipelines.

By using the ‘after` annotation, you can define steps execution hierarchy.

‘sync` and `async` are useful to define job sequences inside step methods.

‘Pipeline` inherits from `Job` which means that pipeline are enqueued as any other job. Pipelines are enqueued in the :ductr_pipelines queue.

class MyPipeline < Ductr::Pipeline
  def first_step
    sync(MyJob, 1)
    async(SomeJob) # Executed when `MyJob` is done
  end

  after :first_step
  def first_parallel_step # Returns when all three `HelloJob` are done
    async(HelloJob, :one)
    async(HelloJob, :two)
    async(HelloJob, :three)
  end

  after :first_step
  def second_parallel_step # Executed concurrently with :first_parallel_step
    async(SomeJob)
    async(SomeOtherJob)
    sync(HelloJob, :one) # Executed when `SomeJob` and `SomeOtherJob` are done
  end

  after :first_parallel_step, :second_parallel_step
  def last_step # Executed when `first_parallel_step` and `second_parallel_step` jobs are done
    sync(ByeJob)
  end
end

You can define pipelines with only one step by using ‘after` annotation without parameter:

class MonoStepPipeline < Ductr::Pipeline
  after
  def unique_step
    async(MyJob)
    async(MyJob)
  end
end

A pipeline can inherit from another, allowing you to overload and add steps to the parent pipeline:

class InheritPipeline < MonoStepPipeline
  after :unique_step
  def not_that_unique
    async(MyJob)
  end
end

Instance Attribute Summary collapse

Attributes inherited from Job

#error, #status

Class Method Summary collapse

Instance Method Summary collapse

Methods inherited from Job

#adapter, #logger, #perform

Methods included from JobStatus

included, #stopped?

Constructor Details

#initialize ⇒ Pipeline

Initializes the pipeline runner



89
90
91
92
93
 
# File 'lib/ductr/pipeline.rb', line 89

def initialize(...)
  super(...)

  @runner = PipelineRunner.new(self)
end

Instance Attribute Details

#runner ⇒ PipelineRunner (readonly)

Returns The pipeline’s runner instance.

Returns:



77
78
79
 
# File 'lib/ductr/pipeline.rb', line 77

def runner
  @runner
end

Class Method Details

.after ⇒ void

This method returns an undefined value.

Annotation to define preceding steps on a pipeline step method.

Examples:

after :some_step_method, :some_other_step_method
def my_step
  # ...
end


72
 
# File 'lib/ductr/pipeline.rb', line 72

annotable :after

Instance Method Details

#async(job_class, *params) ⇒ void

This method returns an undefined value.

Enqueues the given job.

Parameters:

  • job_class (Class<Job>) —

    The job to enqueue

  • *params (Array<Object>) —

    The job’s params



117
118
119
 
# File 'lib/ductr/pipeline.rb', line 117

def async(job_class, *params)
  @runner.current_step.enqueue_job job_class.new(*params)
end

#run ⇒ void

This method returns an undefined value.

Starts the pipeline runner.



84
 
# File 'lib/ductr/pipeline.rb', line 84

def_delegators :@runner, :run

#status=(status) ⇒ void

This method returns an undefined value.

Writes the pipeline’s status into the Ductr’s store.

Parameters:

  • status (Symbol) —

    The status of the job



128
129
130
131
 
# File 'lib/ductr/pipeline.rb', line 128

def status=(status)
  @status = status
  Store.write_pipeline(self)
end

#sync(job_class, *params) ⇒ void

This method returns an undefined value.

Puts the given job in the queue and waits for it to be done.

Parameters:

  • job_class (Class<Job>) —

    The job to enqueue

  • *params (Array<Object>) —

    The job’s params



103
104
105
106
107
 
# File 'lib/ductr/pipeline.rb', line 103

def sync(job_class, *params)
  @runner.current_step.flush_jobs
  @runner.current_step.enqueue_job job_class.new(*params)
  @runner.current_step.flush_jobs
end