<commit_before>module Queight
  class Client
    PublishFailure = Class.new(StandardError)

    def initialize(channel_pool)
      @channel_pool = channel_pool
    end

    def with_channel
      @channel_pool.with_channel do |channel|
        yield(channel)
      end
    end

    def with_subscribe_channel(prefetch)
      @channel_pool.with_subscribe_channel(prefetch) do |channel|
        yield(channel)
      end
    end

    def declare(queue)
      with_channel do |channel|
        queue.declare(channel)
      end
    end

    def publish_without_transaction(config, message, routing_key)
      with_channel do |channel|
        config.publish(channel, message, routing_key)
      end
    end

    def publish(exchange, message, routing_key)
      with_channel do |channel|
        channel.tx_select
        exchange.publish(channel, message, routing_key)
        raise PublishFailure unless channel.tx_commit
      end
    end

    def publish_to_queue(message, queue, message_options = {})
      declare(queue)
      publish(Queight.default_exchange(message_options), message, queue.name)
    end

    def publish_to_queue_without_transaction(message, queue, options = {})
      declare(queue)
      publish_without_transaction(
        Queight.default_exchange(options),
        message,
        queue.name
      )
    end

    # TODO: subscribe, sudddenly, a wild prefetch appears
    def subscribe(queue, prefetch = 1, &block)<commit_msg>Add #publish*! aliases for #*_without_transaction<commit_after>module Queight
  class Client
    PublishFailure = Class.new(StandardError)

    def initialize(channel_pool)
      @channel_pool = channel_pool
    end

    def with_channel
      @channel_pool.with_channel do |channel|
        yield(channel)
      end
    end

    def with_subscribe_channel(prefetch)
      @channel_pool.with_subscribe_channel(prefetch) do |channel|
        yield(channel)
      end
    end

    def declare(queue)
      with_channel do |channel|
        queue.declare(channel)
      end
    end

    def publish(exchange, message, routing_key)
      with_channel do |channel|
        channel.tx_select
        exchange.publish(channel, message, routing_key)
        raise PublishFailure unless channel.tx_commit
      end
    end

    def publish_without_transaction(config, message, routing_key)
      with_channel do |channel|
        config.publish(channel, message, routing_key)
      end
    end
    alias publish! publish_without_transaction

    def publish_to_queue(message, queue, message_options = {})
      declare(queue)
      publish(Queight.default_exchange(message_options), message, queue.name)
    end

    def publish_to_queue_without_transaction(message, queue, options = {})
      declare(queue)
      publish_without_transaction(
        Queight.default_exchange(options),
        message,
        queue.name
      )
    end
    alias publish_to_queue! publish_to_queue_without_transaction

    # TODO: subscribe, sudddenly, a wild prefetch appears
    def subscribe(queue, prefetch = 1, &block)
{'fraction_non_alphanumeric': 0.06913499344692005, 'fraction_numerical': 0.000655307994757536, 'mean_word_length': 2.9393548387096775, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, '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': '4784626', 'n_tokens_mistral': 880, 'n_tokens_neox': 848, 'n_words': 199}