Class: MagicPipe::Senders::Async::Worker

Inherits:
Object
  • Object
show all
Includes:
MetricsMixin, Sidekiq::Worker
Defined in:
lib/magic_pipe/senders/async.rb

Instance Method Summary collapse

Methods included from MetricsMixin

#track_failure, #track_success

Instance Method Details

#perform(decomposed_object, topic, time, client_name) ⇒ Object 



12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
 
# File 'lib/magic_pipe/senders/async.rb', line 12

def perform(decomposed_object, topic, time, client_name)
  client = MagicPipe.lookup_client(client_name)
  object = client.loader.load(decomposed_object)
  codec = client.codec

  metadata = {
    topic: topic,
    producer: client.config.producer_name,
    time: time.to_i,
    mime: codec::TYPE
  }

  envelope = Envelope.new(
    body: object,
    **metadata
  )

  payload = codec.new(envelope).encode
  client.transport.submit!(payload, metadata)

  track_success(client.metrics, topic)
rescue => e
  track_failure(client.metrics, topic)
  raise e
end