Class: ConcurrentPipeline::Pipelines::Schema

Inherits:
Object
  • Object
show all
Defined in:
lib/concurrent_pipeline/pipelines/schema.rb

Defined Under Namespace

Classes: Producer, Stats, Step, Timer

Constant Summary collapse

PROCESSORS = 
{
  sync: Processors::Synchronous,
  async: Processors::Asynchronous,
}

Instance Method Summary collapse

Instance Method Details

#around(&block) ⇒ Object 



69
70
71
 
# File 'lib/concurrent_pipeline/pipelines/schema.rb', line 69

def around(&block)
  arounds << block
end

#arounds ⇒ Object 



43
44
45
 
# File 'lib/concurrent_pipeline/pipelines/schema.rb', line 43

def arounds
  @arounds ||= []
end

#before_process(&block) ⇒ Object 



73
74
75
 
# File 'lib/concurrent_pipeline/pipelines/schema.rb', line 73

def before_process(&block)
  before_process_hooks << block
end

#before_process_hooks ⇒ Object 



51
52
53
 
# File 'lib/concurrent_pipeline/pipelines/schema.rb', line 51

def before_process_hooks
  @before_process_hooks ||= []
end

#build_processor(store) ⇒ Object 



59
60
61
62
63
 
# File 'lib/concurrent_pipeline/pipelines/schema.rb', line 59

def build_processor(store)
  PROCESSORS
    .fetch(@processor.fetch(:type))
    .new(store:, producers:, before_process_hooks:, timers:, **@processor.fetch(:attrs))
end

#process(query, label: nil, &block) ⇒ Object 



65
66
67
 
# File 'lib/concurrent_pipeline/pipelines/schema.rb', line 65

def process(query, label: nil, &block)
  producers << Producer.new(query:, block:, label:)
end

#processor(type, **attrs) ⇒ Object 



55
56
57
 
# File 'lib/concurrent_pipeline/pipelines/schema.rb', line 55

def processor(type, **attrs)
  @processor = {type:, attrs:}
end

#producers ⇒ Object 



39
40
41
 
# File 'lib/concurrent_pipeline/pipelines/schema.rb', line 39

def producers
  @producers ||= []
end

#timer(interval, &block) ⇒ Object 



77
78
79
 
# File 'lib/concurrent_pipeline/pipelines/schema.rb', line 77

def timer(interval, &block)
  timers << Timer.new(interval:, block:)
end

#timers ⇒ Object 



47
48
49
 
# File 'lib/concurrent_pipeline/pipelines/schema.rb', line 47

def timers
  @timers ||= []
end