Class: Hoze::Worker

Inherits:
Object
  • Object
show all
Defined in:
lib/hoze/worker.rb

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ Worker

Returns a new instance of Worker.



13
14
15
16
 
# File 'lib/hoze/worker.rb', line 13

def initialize
  @source_config = Hoze::Configuration.new
  @source_factory = Hoze::SourceFactory.new
end

Instance Attribute Details

#source_factory ⇒ Object

Returns the value of attribute source_factory.



11
12
13
 
# File 'lib/hoze/worker.rb', line 11

def source_factory
  @source_factory
end

Instance Method Details

#go! ⇒ Object 



38
39
40
41
42
43
44
45
46
47
48
49
50
51
 
# File 'lib/hoze/worker.rb', line 38

def go!
  @source = @source_factory.build(@source_config)
  @destination = @source_factory.build(@destination_config) if @destination_config
  @source.listen do |msg|
    begin
      msg.ack! if @source_config.auto_ack
      result = @process_fn.call msg
      @destination.push result, msg.metadata if @destination
    rescue Exception => e
      msg.retry!
      raise e
    end
  end
end

#listen_to {|@source_config| ... } ⇒ Object

Yields:

  • (@source_config) 


18
19
20
21
22
 
# File 'lib/hoze/worker.rb', line 18

def listen_to &block
  yield @source_config
  puts "Source Configuration is: #{@source_config.inspect}"
  self
end

#process(&block) ⇒ Object 



31
32
33
34
35
36
 
# File 'lib/hoze/worker.rb', line 31

def process &block
  @process_fn = Proc.new do |msg|
    yield msg
  end
  self
end

#push_to {|@destination_config| ... } ⇒ Object

Yields:

  • (@destination_config) 


24
25
26
27
28
29
 
# File 'lib/hoze/worker.rb', line 24

def push_to &block
  @destination_config = Hoze::Configuration.new 'DEST'
  yield @destination_config
  puts "Destination Configuration is: #{@destination_config.inspect}"
  self
end