Class: BBK::AMQP::Publisher

Inherits:
Object
  • Object
show all
Includes:
BBK::App::Dispatcher::_Publisher
Defined in:
lib/bbk/amqp/publisher.rb,
sig/bbk/amqp/publisher.rbs

Overview

Publisher send amqp messages

Defined Under Namespace

Classes: PublishedMessage, ReturnError

Constant Summary collapse

HEADER_PROP_FIELDS =

Returns:

  • (Array[Symbol]) 
%i[message_id reply_to correlation_id, timestamp].freeze
PROTOCOLS = 
%w[mq amqp amqps].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(connection, domains, logger: BBK::AMQP.logger) ⇒ Publisher

Returns a new instance of Publisher.

Parameters:

  • connection (Object)
  • domains (Object)
  • logger: (_Logger) (defaults to: BBK::AMQP.logger) 


43
44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/bbk/amqp/publisher.rb', line 43

def initialize(connection, domains, logger: BBK::AMQP.logger)
  @connection = connection
  @channel = connection.channel
  @domains = domains

  logger = logger.respond_to?(:tagged) ? logger : ActiveSupport::TaggedLogging.new(logger)
  @logger = BBK::Utils::ProxyLogger.new(logger, tags: [self.class.to_s, "Ch##{@channel.id}"])

  @ack_map = Concurrent::Map.new
  @sended_messages = Concurrent::Map.new
  @configured_exchanges = Set.new
  initialize_callbacks
end

Instance Attribute Details

#ack_map ⇒ Object (readonly)

Returns the value of attribute ack_map.



41
42
43
 
# File 'lib/bbk/amqp/publisher.rb', line 41

def ack_map
  @ack_map
end

#channel ⇒ Object (readonly)

Returns the value of attribute channel.



41
42
43
 
# File 'lib/bbk/amqp/publisher.rb', line 41

def channel
  @channel
end

#connection ⇒ Object (readonly)

Returns the value of attribute connection.

Returns:

  • (Object) 


41
42
43
 
# File 'lib/bbk/amqp/publisher.rb', line 41

def connection
  @connection
end

#domains ⇒ Object (readonly)

Returns the value of attribute domains.



41
42
43
 
# File 'lib/bbk/amqp/publisher.rb', line 41

def domains
  @domains
end

#logger ⇒ Object (readonly)

Returns the value of attribute logger.



41
42
43
 
# File 'lib/bbk/amqp/publisher.rb', line 41

def logger
  @logger
end

#sended_messages ⇒ Object (readonly)

Returns the value of attribute sended_messages.



41
42
43
 
# File 'lib/bbk/amqp/publisher.rb', line 41

def sended_messages
  @sended_messages
end

Instance Method Details

#client_name ⇒ String

Get connection user. If tls connectoin try extract CN from connection tls_cert

Returns:

  • (String) —

    user name



145
146
147
148
149
 
# File 'lib/bbk/amqp/publisher.rb', line 145

def client_name
  return connection.user unless connection.ssl?

  @client_name ||= Utils.commonname(connection.transport.tls_certificate_path)
end

#close ⇒ Object

Close publisher - try close amqp channel



64
65
66
67
68
69
70
71
 
# File 'lib/bbk/amqp/publisher.rb', line 64

def close
  @channel.tap do |c|
    return nil unless c

    @channel = nil
    c.close
  end
end

#configure_exchange(exchange_name) ⇒ void

This method returns an undefined value.

Configure on return callback for exchange

Parameters:

  • exchange_name (String, nil) 


134
135
136
137
138
139
140
141
 
# File 'lib/bbk/amqp/publisher.rb', line 134

def configure_exchange(exchange_name)
  return if @configured_exchanges.include?(exchange_name)

  logger.debug "Configure on_return callback for exchange #{exchange_name}"
  exchange = channel.exchange(exchange_name, passive: true)
  exchange.on_return(&method(:on_return).curry(4).call(exchange))
  @configured_exchanges << exchange_name
end

#initialize_callbacks ⇒ void

This method returns an undefined value.

Initialize amqp callbacks



129
130
131
 
# File 'lib/bbk/amqp/publisher.rb', line 129

def initialize_callbacks
  @channel.confirm_select method(:on_confirm).curry(4).call(channel)
end

#on_confirm(channel, ack_id, flag, neg) ⇒ Object

Parameters:

  • channel (Object)
  • ack_id (Integer)
  • flag (Object)
  • neg (Boolean)

Returns:

  • (Object) 


163
164
165
166
167
168
169
170
171
172
173
174
175
176
 
# File 'lib/bbk/amqp/publisher.rb', line 163

def on_confirm(channel, ack_id, flag, neg)
  logger.debug "Call confirmed callback for message with ack_id #{ack_id} with neg=#{neg}"
  args = { channel: channel, ack_id: ack_id, flag: flag, neg: neg }
  if ack_map.delete(ack_id) && (f = sended_messages.delete(ack_id)).present?
    if neg
      f.reject(args)
    else
      f.fulfill(args)
    end
  end
rescue StandardError => e
  # TODO: возможно стоит попробовать почистить ack_map и sended_messages
  logger.error "[CRITICAL]: #{e.inspect}.\n#{e.backtrace.first(10).join("\n")}"
end

#on_return(exchange, basic_return, properties, body) ⇒ void

This method returns an undefined value.

Parameters:

  • exchange (Object)
  • basic_return (Object)
  • properties (Hash[String|Symbol, untyped])
  • body (String, nil) 


151
152
153
154
155
156
157
158
159
160
161
 
# File 'lib/bbk/amqp/publisher.rb', line 151

def on_return(exchange, basic_return, properties, body)
  args = { exchange: exchange, basic_return: basic_return, properties: properties, body: body }
  message_id = properties[:message_id]
  logger.info "Message with message_id #{message_id} returned #{basic_return.inspect}"
  ack_id, = ack_map.each_pair.find {|_, msg_id| msg_id == message_id }

  sended_messages.delete(ack_id)&.reject(ReturnError.new(**args)) if ack_map.delete(ack_id)
rescue StandardError => e
  # TODO: возможно стоит попробовать почистить ack_map и sended_messages
  logger.error "[CRITICAL]: #{e.inspect}.\n#{e.backtrace.first(10).join("\n")}"
end

#protocols ⇒ Array<Symbol>

Returned supported protocols list

Returns:

  • (Array<Symbol>) 


59
60
61
 
# File 'lib/bbk/amqp/publisher.rb', line 59

def protocols
  PROTOCOLS
end

#publish(result) ⇒ Object

Publish dispatcher result

Parameters:

  • result (BBK::App::Dispatcher::Result) —

    sended result

Raises:

  • (ArgumentError) 


75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/bbk/amqp/publisher.rb', line 75

def publish(result)
  logger.debug "Try publish dispatcher result #{result.inspect}"
  route = result.route
  result_domain = route.domain
  raise "Unsupported protocol #{route.scheme}" unless PROTOCOLS.include?(route.scheme)
  raise "Unknown domain #{result_domain}" unless domains.has?(result_domain)

  domain = domains[result_domain]
  raise ArgumentError.new("Unknown route domain #{resutl_domain}") if domain.nil?

  route_info = domain.call(route)
  logger.debug "Route #{route.inspect} transformed to #{route_info.inspect}"
  message = result.message
  publish_message(route_info.routing_key, PublishedMessage.new({**message.headers, **route_info.headers}, message.payload), exchange: route_info.exchange)
end

#publish_message(routing_key, message, exchange:, options: {}) ⇒ Object

Publish message

Parameters:

  • routing_key (String) —

    message routing key

  • message (Object) —

    (object with headers and payload method)

  • exchange (String) —

    exchange for sending message

  • options (Hash) (defaults to: {}) —

    message properties

  • exchange: (String)
  • options: (Hash[String|Symbol, untyped]) (defaults to: {})

Returns:

  • (Object) 


96
97
98
99
100
101
102
103
104
105
106
107
108
109
 
# File 'lib/bbk/amqp/publisher.rb', line 96

def publish_message(routing_key, message, exchange:, options: {})
  message.headers[:timestamp] ||= Time.now.to_i
  logger.debug "Try publish message #{message.headers.inspect}"
  properties = {
    persistent:  true,
    mandatory:   true,
    routing_key: routing_key,
    headers:     message.headers,
    # user_id:     client_name,
    **message.headers.select {|k| HEADER_PROP_FIELDS.include?(k.to_sym) }.compact
  }.merge(options).symbolize_keys
  properties[:user_id] = client_name if message.headers[:user_id].blank?
  send_message(exchange, routing_key, message.payload, properties)
end

#raw_publish(routing_key, exchange:, properties: {}, headers: {}, payload: {}) ⇒ Object

Publish raw payload

Parameters:

  • routing_key (String) —

    routing key for sending data

  • exchange (String) —

    exchange name

  • properties (Hash) (defaults to: {}) —

    amqp message properties

  • headers (Messag) (defaults to: {})
  • exchange: (String)
  • properties: (Hash[String|Symbol, untyped]) (defaults to: {})
  • headers: (Hash[String|Symbol, untyped]) (defaults to: {})
  • payload: (Hash[String|Symbol, untyped], String, nil) (defaults to: {})

Returns:

  • (Object) 


116
117
118
119
120
121
122
123
124
 
# File 'lib/bbk/amqp/publisher.rb', line 116

def raw_publish(routing_key, exchange:, properties: {}, headers: {}, payload: {})
  logger.debug "Publish raw message #{headers.inspect}"
  properties = properties.deep_dup
  properties[:headers] = properties.fetch(:headers, {}).merge headers
  properties = properties.merge(headers.select do |k|
                                  HEADER_PROP_FIELDS.include? k
                                end.compact).symbolize_keys
  send_message(exchange, routing_key, payload, properties)
end

#send_message(exchange, routing_key, payload, options) ⇒ Concurrent::Promises::ResolvableFuture

Send amqp message and save meta information for processing

Parameters:

  • exchange (String) —

    exchange name

  • routing_key (String) —

    message routing key

  • payload (Object) —

    message payload. Converted to json calling to_json method

  • options (Hash) —

    amqp message properties

Returns:

  • (Concurrent::Promises::ResolvableFuture) —

    future for checking success publishing



184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
 
# File 'lib/bbk/amqp/publisher.rb', line 184

def send_message(exchange, routing_key, payload, options)
  configure_exchange(exchange)
  channel.synchronize do
    ack_id = channel.next_publish_seq_no
    options[:message_id] ||= SecureRandom.uuid
    # в случае укаказанного message_id в качестве числа, on_return вернет message_id в качестве строки
    ack_map[ack_id] = options[:message_id].to_s
    future = sended_messages[ack_id] = Concurrent::Promises.resolvable_future
    logger.debug "Publish message #{options[:message_id]} with ack: #{ack_id} to #{exchange}##{routing_key}"
    data = if payload.is_a?(String)
      payload
    else
      Oj.generate(payload)
    end
    channel.basic_publish(data, exchange, routing_key, options)
    future
  end
end