Class: KinesisProducer::Library::Handler

Inherits:
Object
  • Object
show all
Defined in:
lib/kinesis_producer/library.rb

Instance Method Summary collapse

Constructor Details

#initialize(futures) ⇒ Handler

Returns a new instance of Handler.



21
22
23
 
# File 'lib/kinesis_producer/library.rb', line 21

def initialize(futures)
  @futures = futures
end

Instance Method Details

#on_message(message) ⇒ Object 



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

def on_message(message)
  if !message.put_record_result.nil?
    on_put_record_result(message)
  elsif !message.metrics_response.nil?
    on_metrics_response(message)
  end
end

#on_metrics_response(message) ⇒ Object 



45
46
47
48
49
50
51
 
# File 'lib/kinesis_producer/library.rb', line 45

def on_metrics_response(message)
  source_id = message.source_id
  response = message.metrics_response
  f = @futures.fetch(source_id)
  @futures.delete(source_id)
  f.set(response.metrics)
end

#on_put_record_result(message) ⇒ Object 



33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/kinesis_producer/library.rb', line 33

def on_put_record_result(message)
  source_id = message.source_id
  result = message.put_record_result
  f = @futures.fetch(source_id)
  @futures.delete(source_id)
  if result.success
    f.set(result)
  else
    f.fail(StandardError.new(result.to_hash))
  end
end