Module: Vx::Consumer::Publish

Defined in:
lib/vx/consumer/publish.rb,
lib/vx/consumer/testing.rb

Instance Method Summary collapse

Instance Method Details

#messages ⇒ Object 



36
37
38
 
# File 'lib/vx/consumer/testing.rb', line 36

def messages
  Testing.messages[params.exchange_name]
end

#messages_and_options ⇒ Object 



40
41
42
 
# File 'lib/vx/consumer/testing.rb', line 40

def messages_and_options
  Testing.messages_and_options[params.exchange_name]
end

#publish(message, options = nil) ⇒ Object 



7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
 
# File 'lib/vx/consumer/publish.rb', line 7

def publish(payload, options = {})
  session.open

  options ||= {}
  options[:routing_key]  = params.routing_key if params.routing_key && !options.key?(:routing_key)
  options[:headers]      = params.headers     if params.headers && !options.key?(:headers)

  options[:content_type] ||= params.content_type || configuration.content_type
  options[:message_id]   ||= SecureRandom.uuid

  name = params.exchange_name

  instrumentation = {
    payload:     payload,
    exchange:    name,
    consumer:    params.consumer_name,
    properties:  options,
  }

  with_middlewares :pub, instrumentation do
    session.with_pub_channel do |ch|
      instrument("process_publishing", instrumentation.merge(channel: ch.id)) do
        encoded = encode_payload(payload, options[:content_type])
        x = session.declare_exchange ch, name, params.exchange_options
        x.publish encoded, options
      end
    end
  end
end