Class: Hutch::Worker

Inherits:
Object
  • Object
show all
Includes:
Logging
Defined in:
lib/hutch/worker.rb

Instance Method Summary collapse

Methods included from Logging

logger, #logger, logger=, setup_logger

Constructor Details

#initialize(broker, consumers, setup_procs) ⇒ Worker

Returns a new instance of Worker.



13
14
15
16
17
18
 
# File 'lib/hutch/worker.rb', line 13

def initialize(broker, consumers, setup_procs)
  @broker         = broker
  @recovery_lock  = Mutex.new
  self.consumers  = consumers
  self.setup_procs = setup_procs
end

Instance Method Details

#acknowledge_error(delivery_info, properties, broker, ex) ⇒ Object 



127
128
129
130
131
132
133
 
# File 'lib/hutch/worker.rb', line 127

def acknowledge_error(delivery_info, properties, broker, ex)
  acks = error_acknowledgements +
    [Hutch::Acknowledgements::NackOnAllFailures.new]
  acks.find do |backend|
    backend.handle(delivery_info, properties, broker, ex)
  end
end

#consumers=(val) ⇒ Object 



135
136
137
138
139
140
 
# File 'lib/hutch/worker.rb', line 135

def consumers=(val)
  if val.empty?
    logger.warn "no consumer loaded, ensure there's no configuration issue"
  end
  @consumers = val
end

#error_acknowledgements ⇒ Object 



142
143
144
 
# File 'lib/hutch/worker.rb', line 142

def error_acknowledgements
  Hutch::Config[:error_acknowledgements]
end

#handle_cancellation(queue_name, cancelled_channel) ⇒ Object

A server-sent basic.cancel carries no reason, so a consumer timeout is indistinguishable from a queue deletion. Only the former is recoverable.



66
67
68
69
70
71
72
73
74
75
76
 
# File 'lib/hutch/worker.rb', line 66

def handle_cancellation(queue_name, cancelled_channel)
  unless @broker.queue_exists?(queue_name)
    logger.error "consumer on queue #{queue_name} was cancelled by the server: the queue no longer exists"
    return
  end

  logger.warn "consumer on queue #{queue_name} was cancelled by the server, re-subscribing"
  resubscribe_on_a_new_channel(cancelled_channel)
rescue => ex
  logger.error "consumer re-subscription failed: #{ex.class}: #{ex.message}"
end

#handle_error(*args) ⇒ Object 



121
122
123
124
125
 
# File 'lib/hutch/worker.rb', line 121

def handle_error(*args)
  Hutch::Config[:error_handlers].each do |backend|
    backend.handle *args.first(backend.method(:handle).arity)
  end
end

#handle_message(consumer, delivery_info, properties, payload) ⇒ Object

Called internally when a new messages comes in from RabbitMQ. Responsible for wrapping up the message and passing it to the consumer.



98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
 
# File 'lib/hutch/worker.rb', line 98

def handle_message(consumer, delivery_info, properties, payload)
  serializer = consumer.get_serializer || Hutch::Config[:serializer]
  logger.debug {
    spec   = serializer.binary? ? "#{payload.bytesize} bytes" : "#{payload}"
    "message(#{properties.message_id || '-'}): " +
    "routing key: #{delivery_info.routing_key}, " +
    "consumer: #{consumer}, " +
    "payload: #{spec}"
  }

  message = Message.new(delivery_info, properties, payload, serializer)
  consumer_instance = consumer.new.tap { |c| c.broker, c.delivery_info = @broker, delivery_info }
  with_tracing(consumer_instance).handle(message)
  @broker.ack(delivery_info.delivery_tag, channel: delivery_info.channel) unless consumer_instance.message_rejected?
rescue => ex
  acknowledge_error(delivery_info, properties, @broker, ex)
  handle_error(properties, payload, consumer, ex, delivery_info)
end

#resubscribe_on_a_new_channel(cancelled_channel) ⇒ Object

Runs on its own thread: closing the channel kills the consumer work pool this callback runs on. Returns the thread so that tests can join it.



80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
 
# File 'lib/hutch/worker.rb', line 80

def resubscribe_on_a_new_channel(cancelled_channel)
  Thread.new do
    begin
      @recovery_lock.synchronize do
        # All consumers share the channel, so only the first cancellation replaces it.
        next unless @broker.channel.equal?(cancelled_channel)

        @broker.replace_channel!
        setup_queues
      end
    rescue => ex
      logger.error "consumer re-subscription failed: #{ex.class}: #{ex.message}"
    end
  end
end

#run ⇒ Object

Run the main event loop. The consumers will be set up with queues, and process the messages in their respective queues indefinitely. This method never returns.



23
24
25
26
27
28
29
30
 
# File 'lib/hutch/worker.rb', line 23

def run
  setup_queues
  setup_procs.each(&:call)

  Waiter.wait_until_signaled

  stop
end

#setup_queue(consumer) ⇒ Object

Bind a consumer's routing keys to its queue, and set up a subscription to receive messages sent to the queue.



48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
 
# File 'lib/hutch/worker.rb', line 48

def setup_queue(consumer)
  queue_name = consumer.get_queue_name
  queue_name = @broker.namespaced_queue_name(queue_name) unless consumer.without_namespace?
  logger.info "setting up queue: #{queue_name}"

  queue = @broker.queue(queue_name, consumer.get_options)
  @broker.bind_queue(queue, consumer.routing_keys)

  on_cancellation = proc { handle_cancellation(queue_name, queue.channel) }
  queue.subscribe(consumer_tag: unique_consumer_tag, manual_ack: true,
                  on_cancellation: on_cancellation) do |*args|
    delivery_info, properties, payload = Hutch::Adapter.decode_message(*args)
    handle_message(consumer, delivery_info, properties, payload)
  end
end

#setup_queues ⇒ Object

Set up the queues for each of the worker's consumers.



38
39
40
41
42
43
44
 
# File 'lib/hutch/worker.rb', line 38

def setup_queues
  logger.info 'setting up queues'
  vetted = @consumers.reject { |c| group_configured? && group_restricted?(c) }
  vetted.each do |c|
    setup_queue(c)
  end
end

#stop ⇒ Object

Stop a running worker by killing all subscriber threads.



33
34
35
 
# File 'lib/hutch/worker.rb', line 33

def stop
  @broker.stop
end

#with_tracing(klass) ⇒ Object 



117
118
119
 
# File 'lib/hutch/worker.rb', line 117

def with_tracing(klass)
  Hutch::Config[:tracer].new(klass)
end