<commit_before>require 'philotic/constants'
require 'philotic/message'
require 'philotic/serialization/serializer'

module Philotic
  class Subscriber

    attr_accessor :connection

    def initialize(connection)
      @connection = connection
    end

    def logger
      connection.logger
    end

    def config
      connection.config
    end

    def subscription_callback(&block)
      lambda do |delivery_info, metadata, payload|
        hash_payload = Philotic::Serialization::Serializer.load(payload, metadata)

        message = Class.new(Philotic::Message).new(metadata[:headers], hash_payload)
        message.delivery_info = delivery_info

        instance_exec(message, &block)
      end
    end

    def subscribe(subscription = {}, subscribe_options = Philotic::DEFAULT_SUBSCRIBE_OPTIONS, &block)
      connection.connect!
      connection.channel.prefetch(connection.config.prefetch_count)

      subscription_settings = get_subscription_settings subscription, subscribe_options

      queue = initialize_queue(subscription_settings)

      queue.subscribe(subscription_settings[:subscribe_options], &subscription_callback(&block))

    end

    def initialize_queue(subscription_settings)
      queue = connection.channel.queue(subscription_settings[:queue_name], subscription_settings[:queue_options])

      queue.bind(connection.exchange, arguments: subscription_settings[:arguments]) if subscription_settings[:arguments]
      queue
    end

    def get_subscription_settings(subscription, subscribe_options)

      if [Symbol, String].include? subscription.class
        queue_name    = subscription
        subscription  = subscribe_options
        queue_options = Philotic::DEFAULT_NAMED_QUEUE_OPTIONS

      else
        queue_name           = subscription[:queue_name] || ''
        queue_options        = Philotic::DEFAULT_ANONYMOUS_QUEUE_OPTIONS
        subscribe_options    = subscribe_options.merge(subscription[:subscribe_options]) if subscription[:subscribe_options]
        arguments            = subscription[:arguments] || subscription
        arguments['x-match'] ||= 'all'
      end

      queue_options.merge!(subscription[:queue_options] || {})

      queue_options[:auto_delete] ||= true if queue_name == ''

      {
          queue_name:        queue_name,
          queue_options:     queue_options,
          arguments:         arguments,
          subscribe_options: subscribe_options,
      }
    end

    def acknowledge(message, up_to_and_including=false)
      connection.channel.acknowledge(message.delivery_tag, up_to_and_including)
    end

    def reject(message, requeue=true)
      connection.channel.reject(message.delivery_tag, requeue)
    end

    def subscribe_to_any(options = {})
      if block_given?
        subscribe(options.merge(:'x-match' => :any), &Proc.new)
      end
    end

    def endure
      while true
        Thread.pass
      end
    end
  end
end<commit_msg>use Thread.stop instead of Thread.pass for endure<commit_after>require 'philotic/constants'
require 'philotic/message'
require 'philotic/serialization/serializer'

module Philotic
  class Subscriber

    attr_accessor :connection

    def initialize(connection)
      @connection = connection
    end

    def logger
      connection.logger
    end

    def config
      connection.config
    end

    def subscription_callback(&block)
      lambda do |delivery_info, metadata, payload|
        hash_payload = Philotic::Serialization::Serializer.load(payload, metadata)

        message               = Class.new(Philotic::Message).new(metadata[:headers], hash_payload)
        message.delivery_info = delivery_info

        instance_exec(message, &block)
      end
    end

    def subscribe(subscription = {}, subscribe_options = Philotic::DEFAULT_SUBSCRIBE_OPTIONS, &block)
      connection.connect!
      connection.channel.prefetch(connection.config.prefetch_count)

      subscription_settings = get_subscription_settings subscription, subscribe_options

      queue = initialize_queue(subscription_settings)

      queue.subscribe(subscription_settings[:subscribe_options], &subscription_callback(&block))

    end

    def initialize_queue(subscription_settings)
      queue = connection.channel.queue(subscription_settings[:queue_name], subscription_settings[:queue_options])

      queue.bind(connection.exchange, arguments: subscription_settings[:arguments]) if subscription_settings[:arguments]
      queue
    end

    def get_subscription_settings(subscription, subscribe_options)

      if [Symbol, String].include? subscription.class
        queue_name    = subscription
        subscription  = subscribe_options
        queue_options = Philotic::DEFAULT_NAMED_QUEUE_OPTIONS

      else
        queue_name           = subscription[:queue_name] || ''
        queue_options        = Philotic::DEFAULT_ANONYMOUS_QUEUE_OPTIONS
        subscribe_options    = subscribe_options.merge(subscription[:subscribe_options]) if subscription[:subscribe_options]
        arguments            = subscription[:arguments] || subscription
        arguments['x-match'] ||= 'all'
      end

      queue_options.merge!(subscription[:queue_options] || {})

      queue_options[:auto_delete] ||= true if queue_name == ''

      {
        queue_name:        queue_name,
        queue_options:     queue_options,
        arguments:         arguments,
        subscribe_options: subscribe_options,
      }
    end

    def acknowledge(message, up_to_and_including=false)
      connection.channel.acknowledge(message.delivery_tag, up_to_and_including)
    end

    def reject(message, requeue=true)
      connection.channel.reject(message.delivery_tag, requeue)
    end

    def subscribe_to_any(options = {})
      if block_given?
        subscribe(options.merge(:'x-match' => :any), &Proc.new)
      end
    end

    def endure
      Thread.stop
    end
  end
end
{'fraction_non_alphanumeric': 0.08179195655862888, 'fraction_numerical': 0.0, 'mean_word_length': 3.2648335745296673, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, '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': '7456319', 'n_tokens_mistral': 1586, 'n_tokens_neox': 1542, 'n_words': 290}