<commit_before>require 'amqp'
require 'multi_json'

module Travis
  module Amqp
    class Amqp
      DEFAULTS = {
        :persistent  => true,
        :durable     => true,
        :auto_delete => false
      }

      class << self
        def connect(config)
          AMQP.start(config)
        end
      end

      attr_reader :connection, :name

      def initialize(connection, name)
        @connection = connection
        @name = name
        # TODO what does this do? is this actually correct?
        require 'amqp/utilities/event_loop_helper'
        AMQP::Utilities::EventLoopHelper.run
      end

      def subscribe(options, &block)
        queue.subscribe(options, &block)
      end

      def publish(queue, data, options = {})
        data = MultiJson.encode(data) if data.is_a?(Hash)
        options = DEFAULTS.merge(options.merge(:routing_key => queue))
        exchange.publish(data, options)
      end

      protected

        def queue
          @queue ||= channel.queue(name, :durable => true, :exclusive => false)
        end

        def exchange
          @exchange ||= AMQP::Channel.new(connection).default_exchange
        end

        def channel
          @channel ||=  AMQP::Channel.new(connection).prefetch(Travis.config.amqp.prefetch)<commit_msg>gotta use the EventLoopHelper on connection setup<commit_after>require 'amqp'
require 'amqp/utilities/event_loop_helper'
require 'multi_json'

module Travis
  module Amqp
    class Amqp
      DEFAULTS = {
        :persistent  => true,
        :durable     => true,
        :auto_delete => false
      }

      class << self
        def connect(config)
          AMQP::Utilities::EventLoopHelper.run
          AMQP.start(config)
        end
      end

      attr_reader :connection, :name

      def initialize(connection, name)
        @connection = connection
        @name = name
      end

      def subscribe(options, &block)
        queue.subscribe(options, &block)
      end

      def publish(queue, data, options = {})
        data = MultiJson.encode(data) if data.is_a?(Hash)
        options = DEFAULTS.merge(options.merge(:routing_key => queue))
        exchange.publish(data, options)
      end

      protected

        def queue
          @queue ||= channel.queue(name, :durable => true, :exclusive => false)
        end

        def exchange
          @exchange ||= AMQP::Channel.new(connection).default_exchange
        end

        def channel
          @channel ||=  AMQP::Channel.new(connection).prefetch(Travis.config.amqp.prefetch)
{'fraction_non_alphanumeric': 0.09992101105845182, 'fraction_numerical': 0.0, 'mean_word_length': 2.1622971285892634, 'pattern_counts': {'":': 0, '<': 7, '<?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': '7022724', 'n_tokens_mistral': 735, 'n_tokens_neox': 713, 'n_words': 168}