Class: ElmerFudd::Worker

Inherits:
Object
  • Object
show all
Defined in:
lib/ElmerFudd/worker.rb

Defined Under Namespace

Classes: Env, Message, Route

Class Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(connection, concurrency: 1, logger: Logger.new($stdout)) ⇒ Worker

Returns a new instance of Worker.



62
63
64
65
66
 
# File 'lib/ElmerFudd/worker.rb', line 62

def initialize(connection, concurrency: 1, logger: Logger.new($stdout))
  @connection = connection
  @concurrency = concurrency
  @logger = logger
end

Class Attribute Details

.durable_queues ⇒ Object 



9
 
# File 'lib/ElmerFudd/worker.rb', line 9

def durable_queues; @durable_queues.nil? ? true : @durable_queues; end

.single_channel ⇒ Object 



13
 
# File 'lib/ElmerFudd/worker.rb', line 13

def single_channel; @single_channel.nil? ? true : @single_channel; end

Class Method Details

.default_filters(*filters) ⇒ Object 



25
26
27
 
# File 'lib/ElmerFudd/worker.rb', line 25

def self.default_filters(*filters)
  @filters = filters
end

.handle_call(route, filters: [], handler: nil, &block) ⇒ Object 



39
40
41
42
 
# File 'lib/ElmerFudd/worker.rb', line 39

def self.handle_call(route, filters: [], handler: nil, &block)
  handlers << RpcHandler.new(route, handler || block, (@filters + filters).uniq,
                             durable: false)
end

.handle_cast(route, filters: [], handler: nil, &block) ⇒ Object 



34
35
36
37
 
# File 'lib/ElmerFudd/worker.rb', line 34

def self.handle_cast(route, filters: [], handler: nil, &block)
  handlers << DirectHandler.new(route, handler || block, (@filters + filters + [DiscardReturnValueFilter]).uniq,
                                durable: durable_queues)
end

.handle_event(route, filters: [], handler: nil, &block) ⇒ Object 



29
30
31
32
 
# File 'lib/ElmerFudd/worker.rb', line 29

def self.handle_event(route, filters: [], handler: nil, &block)
  handlers << TopicHandler.new(route, handler || block, (@filters + filters + [DiscardReturnValueFilter]).uniq,
                               durable: durable_queues)
end

.handlers ⇒ Object 



16
17
18
 
# File 'lib/ElmerFudd/worker.rb', line 16

def self.handlers
  @handlers ||= []
end

.payload_as_kwargs(handler, only: nil) ⇒ Object

Helper allowing to use any method taking hash as a handler def example(text:, **_)

puts text

end # then in worker handle_cast(…

handler: payload_as_kwargs(method(:example)))

Thanks to usage of **_ in arguments list it will accept any payload contaning ‘text’ key. Skipping **_ will require listing all payload keys in argument list



54
55
56
57
58
59
60
 
# File 'lib/ElmerFudd/worker.rb', line 54

def self.payload_as_kwargs(handler, only: nil)
  lambda do |_env, message|
    symbolized_payload = message.payload.inject({}){|memo,(k,v)| memo[k.to_sym] = v; memo}
    symbolized_payload = symbolized_payload.select { |k,v| Array(only).include?(k) } if only
    handler.call(symbolized_payload)
  end
end

.Route(queue_name, exchange_and_routing_keys = {"" => queue_name}) ⇒ Object 



20
21
22
23
 
# File 'lib/ElmerFudd/worker.rb', line 20

def self.Route(queue_name, exchange_and_routing_keys = {"" => queue_name})
  exchange, routing_keys = exchange_and_routing_keys.first
  Route.new(exchange, routing_keys, queue_name)
end

Instance Method Details

#start ⇒ Object 



68
69
70
71
72
 
# File 'lib/ElmerFudd/worker.rb', line 68

def start
  self.class.handlers.each do |handler|
    subscribe_handler(handler)
  end
end