<commit_before># encoding: utf-8


module Autobahn
  class TransportSystem
    attr_reader :cluster

    def initialize(api_uri, exchange_name, options={})
      @cluster = (options[:cluster_factory] || Cluster).new(api_uri, options)
      @connection_factory = options[:connection_factory] || HotBunnies
      @exchange_name = exchange_name
      @host_resolver = options[:host_resolver] || DefaultHostResolver.new
      @encoder = options[:encoder] || StringEncoder.new
      @batch_options = options[:batch]
      @logger = options[:logger] || NullLogger.new
      @consumers = []
      @publishers = []
    end

    def exists?(reload=false)
      setup!(reload)
      @routing.keys.any?
    end

    def size
      setup!
      @routing.keys.size
    end

    def connected?
      !!@connections
    end

    def create!(options={})
      return if exists?(true)
      connect!
      connections = @connections.sort.map(&:last)
      queues_per_node = options[:queues_per_node] || 1
      rks_per_queue = options[:routing_keys_per_queue] || 1
      queue_prefix = options[:queue_prefix] || "#{@exchange_name}_"
      rk_prefix = options[:routing_key_prefix] || queue_prefix
      queue_suffix_width = [Math.log10(connections.size * queues_per_node).ceil, 2].max
      rk_suffix_width = [Math.log10(connections.size * queues_per_node * rks_per_queue).ceil, 2].max
      exchange = connections.sample.create_channel.exchange(@exchange_name, :type => :direct, :durable => options.fetch(:durable, true))
      queue_options = {:durable => options.fetch(:durable, true), :arguments => options.fetch(:arguments, {})}
      queue_options[:arguments]['x-ha-policy'] = 'all' if options[:ha] == :all
      connections.size.times do |node_index|
        connection = connections[node_index]
        channel = connection.create_channel
        queues_per_node.times do |queue_offset|
          queue_index = node_index * queues_per_node + queue_offset
          queue_name = "#{queue_prefix}#{queue_index.to_s.rjust(queue_suffix_width, '0')}"
          queue = channel.queue(queue_name, queue_options)
          rks_per_queue.times do |rk_offset|
            rk_index = queue_index * rks_per_queue + rk_offset
            rk_name = "#{rk_prefix}#{rk_index.to_s.rjust(rk_suffix_width, '0')}"
            queue.bind(exchange, :routing_key => rk_name)
          end
        end
      end
      setup!(true)
    end

    def destroy!
      setup!(true)
      connect!
      disconnect_channels!
      channel = @connections.values.sample.create_channel
      if cluster.exchanges.find { |e| e['name'] == @exchange_name }
        channel.exchange_delete(@exchange_name)
      end
      @routing.keys.each do |queue_name|
        channel.queue_delete(queue_name)
      end
      setup!(true)
    end

    def consumer(options={})
      setup!
      connect!
      options[:preferred_decoder] = options.delete(:encoder) if options.has_key?(:encoder) # for backwards compatibility
      options = {:logger => @logger, :preferred_decoder => @encoder}.merge(options)
      consumer = Consumer.new(@routing, @connections, Encoder, options)
      @consumers << consumer
      @consumers.last
    end

    def publisher(options={})
      setup!
      connect!
      if @batch_options
        raise ArgumentError, 'Encoder does not support batching!' unless @encoder.encodes_batches?
        options[:batch] = @batch_options.merge(options.fetch(:batch,{}))
      end
      options = {:logger => @logger}.merge(options)
      publisher = Publisher.new(@exchange_name, @routing, @connections, @encoder, options)
      publisher.start!
      @publishers << publisher
      @publishers.last
    end

    def disconnect!
      disconnect_channels!
      @connections.values.each(&:close) if @connections
    end

    private

    attr_reader :logger

    class DefaultHostResolver
      def resolve(node)
        node.split('@').last
      end
    end

    def setup!(reload=false)
      @routing = nil if reload
      return if @routing
      bindings = @cluster.bindings.select { |b| b['source'] == @exchange_name && b['destination_type'] == 'queue' }
      @routing = bindings.reduce({}) do |acc, binding|
        queue_name = binding['destination']
        acc[queue_name] ||= {:routing_keys => []}
        acc[queue_name][:routing_keys] << binding['routing_key']
        acc
      end
      @cluster.queues.each do |queue|
        queue_name = queue['name']
        if @routing.key?(queue_name)
          @routing[queue_name][:node] = queue['node']
        end<commit_msg>Fixed a bug in #disconnect!<commit_after># encoding: utf-8


module Autobahn
  class TransportSystem
    attr_reader :cluster

    def initialize(api_uri, exchange_name, options={})
      @cluster = (options[:cluster_factory] || Cluster).new(api_uri, options)
      @connection_factory = options[:connection_factory] || HotBunnies
      @exchange_name = exchange_name
      @host_resolver = options[:host_resolver] || DefaultHostResolver.new
      @encoder = options[:encoder] || StringEncoder.new
      @batch_options = options[:batch]
      @logger = options[:logger] || NullLogger.new
      @consumers = []
      @publishers = []
    end

    def exists?(reload=false)
      setup!(reload)
      @routing.keys.any?
    end

    def size
      setup!
      @routing.keys.size
    end

    def connected?
      !!@connections
    end

    def create!(options={})
      return if exists?(true)
      connect!
      connections = @connections.sort.map(&:last)
      queues_per_node = options[:queues_per_node] || 1
      rks_per_queue = options[:routing_keys_per_queue] || 1
      queue_prefix = options[:queue_prefix] || "#{@exchange_name}_"
      rk_prefix = options[:routing_key_prefix] || queue_prefix
      queue_suffix_width = [Math.log10(connections.size * queues_per_node).ceil, 2].max
      rk_suffix_width = [Math.log10(connections.size * queues_per_node * rks_per_queue).ceil, 2].max
      exchange = connections.sample.create_channel.exchange(@exchange_name, :type => :direct, :durable => options.fetch(:durable, true))
      queue_options = {:durable => options.fetch(:durable, true), :arguments => options.fetch(:arguments, {})}
      queue_options[:arguments]['x-ha-policy'] = 'all' if options[:ha] == :all
      connections.size.times do |node_index|
        connection = connections[node_index]
        channel = connection.create_channel
        queues_per_node.times do |queue_offset|
          queue_index = node_index * queues_per_node + queue_offset
          queue_name = "#{queue_prefix}#{queue_index.to_s.rjust(queue_suffix_width, '0')}"
          queue = channel.queue(queue_name, queue_options)
          rks_per_queue.times do |rk_offset|
            rk_index = queue_index * rks_per_queue + rk_offset
            rk_name = "#{rk_prefix}#{rk_index.to_s.rjust(rk_suffix_width, '0')}"
            queue.bind(exchange, :routing_key => rk_name)
          end
        end
      end
      setup!(true)
    end

    def destroy!
      setup!(true)
      connect!
      disconnect_channels!
      channel = @connections.values.sample.create_channel
      if cluster.exchanges.find { |e| e['name'] == @exchange_name }
        channel.exchange_delete(@exchange_name)
      end
      @routing.keys.each do |queue_name|
        channel.queue_delete(queue_name)
      end
      setup!(true)
    end

    def consumer(options={})
      setup!
      connect!
      options[:preferred_decoder] = options.delete(:encoder) if options.has_key?(:encoder) # for backwards compatibility
      options = {:logger => @logger, :preferred_decoder => @encoder}.merge(options)
      consumer = Consumer.new(@routing, @connections, Encoder, options)
      @consumers << consumer
      @consumers.last
    end

    def publisher(options={})
      setup!
      connect!
      if @batch_options
        raise ArgumentError, 'Encoder does not support batching!' unless @encoder.encodes_batches?
        options[:batch] = @batch_options.merge(options.fetch(:batch,{}))
      end
      options = {:logger => @logger}.merge(options)
      publisher = Publisher.new(@exchange_name, @routing, @connections, @encoder, options)
      publisher.start!
      @publishers << publisher
      @publishers.last
    end

    def disconnect!
      disconnect_channels!
      @connections.values.each(&:close) if @connections
      @connections = nil
    end

    private

    attr_reader :logger

    class DefaultHostResolver
      def resolve(node)
        node.split('@').last
      end
    end

    def setup!(reload=false)
      @routing = nil if reload
      return if @routing
      bindings = @cluster.bindings.select { |b| b['source'] == @exchange_name && b['destination_type'] == 'queue' }
      @routing = bindings.reduce({}) do |acc, binding|
        queue_name = binding['destination']
        acc[queue_name] ||= {:routing_keys => []}
        acc[queue_name][:routing_keys] << binding['routing_key']
        acc
      end
      @cluster.queues.each do |queue|
        queue_name = queue['name']
        if @routing.key?(queue_name)
          @routing[queue_name][:node] = queue['node']
        end
{'fraction_non_alphanumeric': 0.12208729898260584, 'fraction_numerical': 0.0024067388688327317, 'mean_word_length': 2.985178727114211, 'pattern_counts': {'":': 0, '<': 15, '<?xml version=': 0, '>': 21, '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': '2625127', 'n_tokens_mistral': 2809, 'n_tokens_neox': 2733, 'n_words': 543}