Sha256: 19b91299f56378dd2404968b8c4606fac727b6b86f3f79e7e1618184757a1f30
Contents?: true
Size: 1021 Bytes
Versions: 34
Compression:
Stored size: 1021 Bytes
Contents
require 'dry-initializer'
module Messaging
module Adapters
class Test
class Consumer
include Messaging::Routing
extend Dry::Initializer
option :name
option :logger, default: -> { Config.logger }
option :queue, default: -> { Queue.new }
def start
logger.info "Consumer #{name} started"
@running = true
process_messages
ensure
shutdown
end
def stop
@running = false
queue << :stop
logger.info "Consumer #{name} stopping"
end
def shutdown
logger.info "Consumer #{name} stopped"
end
private
def process_messages
while @running
message = queue.pop
logger.info "Consumer #{name} got #{message}"
break if message == :stop
RunMiddlewares.call(Config.consumer.middlewares, message) { handle message }
end
end
end
end
end
end
Version data entries
34 entries across 34 versions & 1 rubygems