Class: EventHub::ActorListener
- Inherits:
-
Object
- Object
- EventHub::ActorListener
- Includes:
- Celluloid, Helper
- Defined in:
- lib/eventhub/actor_listener.rb
Overview
Listner Class
Instance Method Summary collapse
- #cleanup ⇒ Object
- #handle_payload(args = {}) ⇒ Object
-
#initialize(processor_instance) ⇒ ActorListener
constructor
A new instance of ActorListener.
- #listen(args = {}) ⇒ Object
- #pass_arguments(args = {}) ⇒ Object
- #publish(args) ⇒ Object
- #restart ⇒ Object
- #start ⇒ Object
- #with_listen(args = {}) {|connection, channel, consumer, queue, queue_name| ... } ⇒ Object
Methods included from Helper
#create_bunny_connection, #get_name_from_class, #now_stamp, #stringify_keys
Constructor Details
#initialize(processor_instance) ⇒ ActorListener
Returns a new instance of ActorListener.
9 10 11 12 13 14 15 |
# File 'lib/eventhub/actor_listener.rb', line 9 def initialize(processor_instance) @actor_publisher = ActorPublisher.new_link @actor_watchdog = ActorWatchdog.new_link @connections = {} @processor_instance = processor_instance start end |
Instance Method Details
#cleanup ⇒ Object
111 112 113 114 115 116 117 118 |
# File 'lib/eventhub/actor_listener.rb', line 111 def cleanup EventHub.logger.info("Listener is cleaning up...") # close all open connections return unless @connections @connections.values.each do |connection| connection&.close end end |
#handle_payload(args = {}) ⇒ Object
72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 |
# File 'lib/eventhub/actor_listener.rb', line 72 def handle_payload(args = {}) = [] connection = args[:connection] # convert to EventHub message = EventHub::Message.from_json(args[:payload]) # append to execution history .append_to_execution_history(EventHub::Configuration.name) # return invalid messages to dispatcher if .invalid? << EventHub.logger.info("-> #{} => return invalid to dispatcher") else begin = @processor_instance.send(:handle_message, , pass_arguments(args)) rescue => exception # this catches unexpected exceptions in handle message method # deadletter the message via dispatcher .status_code = EventHub::STATUS_DEADLETTER . = exception.to_s EventHub.logger.info("-> #{} => return exception to dispatcher") << end end Array().each do || publish(message: .to_json, connection: connection) end end |
#listen(args = {}) ⇒ Object
28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 |
# File 'lib/eventhub/actor_listener.rb', line 28 def listen(args = {}) with_listen(args) do |connection, channel, consumer, queue, queue_name| EventHub.logger.info("Listening to queue [#{queue_name}]") consumer.on_delivery do |delivery_info, , payload| EventHub.logger.info("#{queue_name}: [#{delivery_info.delivery_tag}]" \ " delivery") @processor_instance.statistics.measure(payload.size) do handle_payload(payload: payload, connection: connection, queue_name: queue_name, content_type: [:content_type], priority: [:priority], delivery_tag: delivery_info.delivery_tag) channel.acknowledge(delivery_info.delivery_tag, false) end EventHub.logger.info("#{queue_name}: [#{delivery_info.delivery_tag}]" \ " acknowledged") end queue.subscribe_with(consumer, block: false) end rescue => error EventHub.logger.error("Unexpected exception: #{error}. It should restart now with this exception...") raise end |
#pass_arguments(args = {}) ⇒ Object
106 107 108 109 |
# File 'lib/eventhub/actor_listener.rb', line 106 def pass_arguments(args = {}) keys_to_pass = [:queue_name, :content_type, :priority, :delivery_tag] args.select { |key| keys_to_pass.include?(key) } end |
#publish(args) ⇒ Object
120 121 122 |
# File 'lib/eventhub/actor_listener.rb', line 120 def publish(args) @actor_publisher.publish(args) end |
#restart ⇒ Object
24 25 26 |
# File 'lib/eventhub/actor_listener.rb', line 24 def restart raise "Listener is restarting..." end |
#start ⇒ Object
17 18 19 20 21 22 |
# File 'lib/eventhub/actor_listener.rb', line 17 def start EventHub.logger.info("Listener is starting...") EventHub::Configuration.processor[:listener_queues].each_with_index do |queue_name, index| async.listen(queue_name: queue_name, index: index) end end |
#with_listen(args = {}) {|connection, channel, consumer, queue, queue_name| ... } ⇒ Object
55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 |
# File 'lib/eventhub/actor_listener.rb', line 55 def with_listen(args = {}, &block) connection = create_bunny_connection connection.start queue_name = args[:queue_name] @connections[queue_name] = connection channel = connection.create_channel channel.prefetch(1) queue = channel.queue(queue_name, durable: true) consumer = EventHub::Consumer.new(channel, queue, EventHub::Configuration.name + "-" + args[:index].to_s, false) yield connection, channel, consumer, queue, queue_name end |