Class: DataCollector::Pipeline

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

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(options = {}) ⇒ Pipeline

Returns a new instance of Pipeline.



7
8
9
10
11
12
13
14
15
16
17
18
19
20
 
# File 'lib/data_collector/pipeline.rb', line 7

def initialize(options = {})
  @running = false
  @paused = false

  @input = DataCollector::Input.new
  @output = DataCollector::Output.new
  @run_count = 0

  @schedule = options[:schedule] || {}
  @cron = options[:cron || '']
  @name = options[:name] || "pipeline-#{Time.now.to_i}-#{rand(10000)}"
  @options = options
  @listeners = []
end

Instance Attribute Details

#name ⇒ Object (readonly)

Returns the value of attribute name.



6
7
8
 
# File 'lib/data_collector/pipeline.rb', line 6

def name
  @name
end

#run_count ⇒ Object (readonly)

Returns the value of attribute run_count.



6
7
8
 
# File 'lib/data_collector/pipeline.rb', line 6

def run_count
  @run_count
end

Instance Method Details

#on_message(&block) ⇒ Object 



22
23
24
 
# File 'lib/data_collector/pipeline.rb', line 22

def on_message(&block)
  @on_message_callback = block
end

#pause ⇒ Object 



95
96
97
98
99
100
101
102
 
# File 'lib/data_collector/pipeline.rb', line 95

def pause
  if @running
  @paused = !@paused
    @listeners.each do |listener|
      listener.pause if listener.running?
    end
  end
end

#paused? ⇒ Boolean

Returns:

  • (Boolean) 


112
113
114
 
# File 'lib/data_collector/pipeline.rb', line 112

def paused?
  @paused
end

#run ⇒ Object 



26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
 
# File 'lib/data_collector/pipeline.rb', line 26

def run
  if paused? && @running
    @paused = false
    @listeners.each do |listener|
      listener.run if listener.paused?
    end
  end

  @running = true
  if @schedule && !@schedule.empty?
    while running?
      @run_count += 1
      start_time = ISO8601::DateTime.new(Time.now.to_datetime.to_s)
      begin
        duration = ISO8601::Duration.new(@schedule)
      rescue StandardError => e
        raise DataCollector::Error, "PIPELINE - bad schedule: #{e.message}"
      end
      interval = ISO8601::TimeInterval.from_duration(start_time, duration)

      DataCollector::Core.log("PIPELINE running in #{interval.size} seconds")
      sleep interval.size
      handle_on_message(@input, @output) unless paused?
    end
  elsif @cron && !@cron.empty?
    cron_parser = CronParser.new(@cron)
    while running?
      @run_count += 1
      start_time = ISO8601::DateTime.new(Time.now.to_datetime.to_s)
      next_run = cron_parser.next(start_time.to_time)

      interval = ISO8601::TimeInterval.from_datetimes(start_time,  ISO8601::DateTime.new(next_run.to_datetime.to_s))

      DataCollector::Core.log("PIPELINE running at #{next_run.to_datetime.strftime('%Y-%m-%dT%H:%M:%S')} or in #{interval.size} seconds")
      sleep interval.size

      handle_on_message(@input, @output) unless paused?
    end
  else # run once
    @run_count += 1
    if @options.key?(:uri)
      listener = Input.new.from_uri(@options[:uri], @options)
      listener.on_message do |input, output, filename|
        DataCollector::Core.log("PIPELINE triggered by #{filename}")
        handle_on_message(@input, @output, filename)
      end
      @listeners << listener

      listener.run(true)

    else
      DataCollector::Core.log("PIPELINE running once")
      handle_on_message(@input, @output)
    end
  end
rescue StandardError => e
  DataCollector::Core.error("PIPELINE run failed: #{e.message}")
  raise e
  #puts e.backtrace.join("\n")
end

#running? ⇒ Boolean

Returns:

  • (Boolean) 


104
105
106
 
# File 'lib/data_collector/pipeline.rb', line 104

def running?
  @running
end

#stop ⇒ Object 



87
88
89
90
91
92
93
 
# File 'lib/data_collector/pipeline.rb', line 87

def stop
  @running = false
  @paused = false
  @listeners.each do |listener|
    listener.stop if listener.running?
  end
end

#stopped? ⇒ Boolean

Returns:

  • (Boolean) 


108
109
110
 
# File 'lib/data_collector/pipeline.rb', line 108

def stopped?
  !@running
end