Contents

module Sneakers
  class Publisher
    def initialize(opts = {})
      @mutex = Mutex.new
      @opts = Sneakers::CONFIG.merge(opts)
    end

    def publish(msg, options = {})
      @mutex.synchronize do
        ensure_connection! unless connected?
      end
      to_queue = options.delete(:to_queue)
      options[:routing_key] ||= to_queue
      Sneakers.logger.info {"publishing <#{msg}> to [#{options[:routing_key]}]"}
      @exchange.publish(msg, options)
    end

    private

    attr_reader :exchange

    def ensure_connection!
      @bunny = Bunny.new(@opts[:amqp], heartbeat: @opts[:heartbeat], vhost: @opts[:vhost], :logger => Sneakers::logger)
      @bunny.start
      @channel = @bunny.create_channel
      @exchange = @channel.exchange(@opts[:exchange], type: @opts[:exchange_type], durable: @opts[:durable])
    end

    def connected?
      @bunny && @bunny.connected?
    end
  end
end

Version data entries

4 entries across 4 versions & 1 rubygems

Version	Path
sneakers-1.0.3	lib/sneakers/publisher.rb
sneakers-1.0.2	lib/sneakers/publisher.rb
sneakers-1.0.1	lib/sneakers/publisher.rb
sneakers-1.0.0	lib/sneakers/publisher.rb