def start(connection_or_group)
      @connection = connection_or_group
      @channel = AMQP::Channel.new(@connection)
      queue_options = {}
      name = queue_name
      if queue_type == :broadcast
        queue_options[:exclusive] = true
        queue_options[:auto_delete] = true
        name = "#{Socket.gethostname}-#{Process.pid}-#{self.object_id}"
        @queue = @channel.queue(name, queue_options)
        @exchange = @channel.fanout(BackgroundBunnies.broadcast_exchange_name(queue_name))
        @queue.bind(@exchange)
      else
        queue_options[:durable] = true
        @queue = @channel.queue(queue_name, queue_options)
      end
      @consumer = @queue.subscribe(:ack=>true) do |metadata, payload|
        info = metadata
        properties = nil
        begin 
          job = Job.new(JSON.parse!(payload), info, properties)
          err = nil
          self.process(job) 
          metadata.ack
        rescue =>err
          # processing went wrong, requeing message
          job = Job.new(nil, info, properties) unless job
          unless on_error(job, err)
            metadata.reject(:requeue=>true)
          else
            metadata.ack
          end
        end
      end
    end
def start(connection_or_group)
      @connection = connection_or_group
      @channel = AMQP::Channel.new(@connection)
      queue_options = {}
      name = queue_name
      if queue_type == :broadcast
        queue_options[:exclusive] = true
        queue_options[:auto_delete] = true
        name = "#{Socket.gethostname}-#{Process.pid}-#{self.object_id}"
        @queue = @channel.queue(name, queue_options)
        @exchange = @channel.fanout(BackgroundBunnies.broadcast_exchange_name(queue_name))
        @queue.bind(@exchange)
      else
        queue_options[:durable] = true
        @queue = @channel.queue(queue_name, queue_options)
      end
      @consumer = @queue.subscribe(:ack=>true) do |metadata, payload|
        info = metadata
        properties = nil
        begin 
          job = Job.new(JSON.parse!(payload), info, properties)
          err = nil
          self.process(job) 
          metadata.ack
        rescue =>err
          # processing went wrong, requeing message
          job = Job.new(nil, info, properties) unless job
          unless on_error(job, err)
            metadata.reject(:requeue=>true)
          else
            metadata.ack
          end
        end
      end
    end
Starts the Worker with the given connection or group name