<commit_before>require 'bunny'

module RabbitWQ
  module Work

    def self.enqueue( worker, options={} )
      payload = worker.to_yaml
      enqueue_payload( payload, options )
    end

    def self.enqueue_payload( payload, options={} )
      delay = options.delete( :delay )
      delay = nil if delay && delay < 5000

      if delay
        with_channel do |channel|
          delay_x = channel.direct( "#{RabbitWQ.configuration.delayed_exchange_prefix}-#{delay}ms", durable: true )

          work_x = channel.send( RabbitWQ.configuration.work_exchange_type,
                                 RabbitWQ.configuration.work_exchange,
                                 durable: true )

          channel.queue( "#{RabbitWQ.configuration.delayed_queue_prefix}-#{delay}ms",
                         durable: true,
                         arguments: { "x-dead-letter-exchange" => work_x.name,
                                      "x-message-ttl" => delay } ).
                  bind( delay_x )

          delay_x.publish( payload, durable: true,
                                    content_type: 'application/yaml',
                                    headers: options )
        end

        return
      end

      with_work_exchange do |work_x, work_q|
        work_x.publish( payload, durable: true,
                                 content_type: 'application/yaml',
                                 headers: options )
      end
    end

    def self.enqueue_error_payload( payload, options={} )
      with_channel do |channel|
        error_q = channel.queue( RabbitWQ.configuration.error_queue, durable: true )
        error_q.publish( payload, durable: true,
                                  content_type: 'application/yaml',
                                  headers: options )
      end
    end

    def self.with_work_exchange
      with_channel do |channel|
        begin
          exchange = channel.send( RabbitWQ.configuration.work_exchange_type,
                                   RabbitWQ.configuration.work_exchange,
                                   durable: true )

          channel.queue( RabbitWQ.configuration.work_queue, durable: true ).tap do |q|
            q.bind( exchange )
            yield exchange, q
          end
        ensure
        end
      end
    end

    def self.with_channel
      Bunny.new.tap do |b|
        b.start
        begin
          b.create_channel.tap do |c|
            yield c
          end
        ensure
          b.stop
        end
      end
    end

    #def self.with_exchange
      #Bunny.new.tap do |b|
        #b.start
        #begin
          #b.create_channel.tap do |c|
            #queue = c.queue( 'replication', durable: true )
            #yield c.default_exchange, queue.name
          #end
        #ensure
          #b.stop
        #end
      #end
    #end

  end
end<commit_msg>Add a config convenience method to RabbitWQ::Work.<commit_after>require 'bunny'

module RabbitWQ
  module Work

    def self.enqueue( worker, options={} )
      payload = worker.to_yaml
      enqueue_payload( payload, options )
    end

    def self.enqueue_payload( payload, options={} )
      delay = options.delete( :delay )
      delay = nil if delay && delay < 5000

      if delay
        with_channel do |channel|
          delay_x = channel.direct( "#{config.delayed_exchange_prefix}-#{delay}ms", durable: true )

          work_x = channel.send( config.work_exchange_type,
                                 config.work_exchange,
                                 durable: true )

          channel.queue( "#{config.delayed_queue_prefix}-#{delay}ms",
                         durable: true,
                         arguments: { "x-dead-letter-exchange" => work_x.name,
                                      "x-message-ttl" => delay } ).
                  bind( delay_x )

          delay_x.publish( payload, durable: true,
                                    content_type: 'application/yaml',
                                    headers: options )
        end

        return
      end

      with_work_exchange do |work_x, work_q|
        work_x.publish( payload, durable: true,
                                 content_type: 'application/yaml',
                                 headers: options )
      end
    end

    def self.enqueue_error_payload( payload, options={} )
      with_channel do |channel|
        error_q = channel.queue( config.error_queue, durable: true )
        error_q.publish( payload, durable: true,
                                  content_type: 'application/yaml',
                                  headers: options )
      end
    end

    def self.with_work_exchange
      with_channel do |channel|
        begin
          exchange = channel.send( config.work_exchange_type,
                                   config.work_exchange,
                                   durable: true )

          channel.queue( config.work_queue, durable: true ).tap do |q|
            q.bind( exchange )
            yield exchange, q
          end
        ensure
        end
      end
    end

    def self.with_channel
      Bunny.new.tap do |b|
        b.start
        begin
          b.create_channel.tap do |c|
            yield c
          end
        ensure
          b.stop
        end
      end
    end

    def self.config
      RabbitWQ.configuration
    end

  end
end
{'fraction_non_alphanumeric': 0.07922272047832586, 'fraction_numerical': 0.0014947683109118087, 'mean_word_length': 1.3416447944007, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5591851', 'n_tokens_mistral': 1471, 'n_tokens_neox': 1383, 'n_words': 337}