<commit_before>require 'bunny'
require 'connection_pool'

class Woodhouse::Dispatchers::BunnyDispatcher < Woodhouse::Dispatcher

  def initialize(config)
    super
    @pool = new_pool
  end

  private

  def deliver_job(job)
    run do |conn|
      exchange = conn.exchange(job.exchange_name, :type => :headers)
      exchange.publish(" ", :headers => job.arguments)
    end
  end

  def deliver_job_update(job, data)
    run do |conn|
      exchange = conn.direct("woodhouse.progress")
      # establish durable queue to pick up updates
      conn.queue(job.job_id, :durable => true).bind(exchange, :routing_key => job.job_id)
      exchange.publish(data.to_json, :routing_key => job.job_id)
    end
  rescue => err
    Woodhouse.logger.warn("Error when dispatching job update: #{err.class}: #{err.message}")
  end

  def run
    @pool.with do |conn|
      yield conn
    end
  end

  private

  def new_pool
    @bunny.stop if @bunny

    bunny = @bunny = Bunny.new(@config.server_info || {})
    @bunny.start

    ConnectionPool.new { bunny.create_channel }
  end

end<commit_msg>Try to create a new connection if a Bunny connection times out.<commit_after>require 'bunny'
require 'connection_pool'

class Woodhouse::Dispatchers::BunnyDispatcher < Woodhouse::Dispatcher

  def initialize(config)
    super
    @pool = new_pool
  end

  private

  def deliver_job(job)
    run do |conn|
      exchange = conn.exchange(job.exchange_name, :type => :headers)
      exchange.publish(" ", :headers => job.arguments)
    end
  end

  def deliver_job_update(job, data)
    run do |conn|
      exchange = conn.direct("woodhouse.progress")
      # establish durable queue to pick up updates
      conn.queue(job.job_id, :durable => true).bind(exchange, :routing_key => job.job_id)
      exchange.publish(data.to_json, :routing_key => job.job_id)
    end
  rescue => err
    Woodhouse.logger.warn("Error when dispatching job update: #{err.class}: #{err.message}")
  end

  def run
    retried = false
    @pool.with do |conn|
      yield conn
    end
  rescue Bunny::ClientTimeout
    if retried
      raise
    else
      new_pool!
      retried = true
      retry
    end
  end

  private

  def new_pool!
    @pool = new_pool
  end

  def new_pool
    @bunny.stop if @bunny

    bunny = @bunny = Bunny.new(@config.server_info || {})
    @bunny.start

    ConnectionPool.new { bunny.create_channel }
  end

end
{'fraction_non_alphanumeric': 0.10602910602910603, 'fraction_numerical': 0.0, 'mean_word_length': 3.01669449081803, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '124761', 'n_tokens_mistral': 809, 'n_tokens_neox': 777, 'n_words': 198}