Class: Surfliner::MetadataConsumer::Consumer

Inherits:
Object
  • Object
show all
Includes:
Util::Assert
Defined in:
lib/surfliner/metadata_consumer/consumer.rb

Overview

A metadata consumer that subscribes to a RabbitMQ queue and passes messages to the specified handler.

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Util::Assert

#not_nil!

Constructor Details

#initialize(connection:, handler:, tracer: nil) ⇒ Consumer

Initializes a new Consumer

Parameters:

  • connection (Mq::Connection) —

    the connection

  • handler (#handle) —

    an object accepting a JSON string

  • tracer (OpenTelemetry::Trace::Tracer) (defaults to: nil) —

    OpenTelemetry tracer



29
30
31
32
33
 
# File 'lib/surfliner/metadata_consumer/consumer.rb', line 29

def initialize(connection:, handler:, tracer: nil)
  @connection = not_nil!(connection, "connection")
  @handler = not_nil!(handler, "handler")
  @tracer = tracer
end

Instance Attribute Details

#connection ⇒ Mq::Connection (readonly)

Returns the connection.

Returns:



16
17
18
 
# File 'lib/surfliner/metadata_consumer/consumer.rb', line 16

def connection
  @connection
end

#handler ⇒ #handle (readonly)

Returns an object accepting a JSON string.

Returns:

  • (#handle) —

    an object accepting a JSON string



22
23
24
 
# File 'lib/surfliner/metadata_consumer/consumer.rb', line 22

def handler
  @handler
end

#tracer ⇒ OpenTelemetry::Trace::Tracer (readonly)

Returns OpenTelemetry tracer.

Returns:

  • (OpenTelemetry::Trace::Tracer) —

    OpenTelemetry tracer



19
20
21
 
# File 'lib/surfliner/metadata_consumer/consumer.rb', line 19

def tracer
  @tracer
end

Instance Method Details

#handle(payload_json) ⇒ Object

Dispatches the specified payload to the handler.

Parameters:

  • payload_json (String) —

    JSON message payload



52
53
54
55
56
57
58
59
 
# File 'lib/surfliner/metadata_consumer/consumer.rb', line 52

def handle(payload_json)
  logger.info(" [→] message received with payload: #{payload_json}")

  handler.handle(payload_json)
  logger.info(" [✓] message handled")
rescue => err
  logger.error(" [!] failed to handle message: #{err.full_message}")
end

#logger ⇒ Logger

Returns log message destination.

Returns:

  • (Logger) —

    log message destination



36
37
38
 
# File 'lib/surfliner/metadata_consumer/consumer.rb', line 36

def logger
  connection.logger
end

#run(topic_config: Mq::TopicConfig.from_env, queue_config: Mq::QueueConfig.from_env) ⇒ Object

Starts listening to the message queue and passing messages to the handler.

Parameters:

  • topic_config (TopicConfig) (defaults to: Mq::TopicConfig.from_env) —

    topic configuration

  • queue_config (QueueConfig) (defaults to: Mq::QueueConfig.from_env) —

    queue configuration



43
44
45
46
47
48
 
# File 'lib/surfliner/metadata_consumer/consumer.rb', line 43

def run(topic_config: Mq::TopicConfig.from_env, queue_config: Mq::QueueConfig.from_env)
  connection.with_topic(topic_config) do |topic|
    queue = topic.bind_queue(queue_config)
    queue.subscribe(block: true, &method(:on_delivery))
  end
end