<commit_before>require 'wumomq/version'
require 'bunny'

module Wumomq
  class Consumer
    def process delivery_info, properties, payload
      raise NotImplementedError, "Implement this method in a child class"

      # 应该返回2个值，succeed，abort
    end
  end

  class Base
    def initialize uri, queue_name, queue_num = 1, exchange_name = nil
      @conn      = Bunny.new uri
      @conn.start
      @channel   = @conn.create_channel
      @num       = queue_num
      @queue     = Array.new
      (0...@num).each { |i|
        @queue.push(@channel.queue("#{queue_name}.#{i}", :auto_delete => false))
      }

      @consumer  = nil
      @exchange  = @channel.default_exchange
    end

    def get_number_of_subscribe
      @num
    end

    def hash value
      Digest::MD5.hexdigest(value)[0...4].to_i(16) % @num
    end

    def publish options = {}
      (puts "没有消息主体(#{__FILE__}.#{__LINE__})"; return nil) if options[:message].nil? 
      (puts "没有指定key(#{__FILE__}.#{__LINE__})"; return nil) if options[:key].nil?
      q = hash(options[:key])
      @exchange.publish(options[:message], :routing_key => @queue[q].name)
    end

    def subscribe options = {}
      # 在登记的时候, key应该是整数
      (puts "没有处理的consumer(#{__FILE__}.#{__LINE__})"; return false) if options[:consumer].nil? 
      (puts "key为空(#{__FILE__}.#{__LINE__})"; return false) if options[:key].nil?

      @consumer = @queue[options[:key]].subscribe(:manual_ack => true, :block => false) do |delivery_info, properties, payload|
         succeed, abort = options[:consumer].process delivery_info, properties, payload
         @channel.acknowledge delivery_info.delivery_tag, false if succeed || abort
      end

      true
    end

    def cancel
      @consumer.cancel if @consumer
    end
<commit_msg>value.to_s<commit_after>require 'wumomq/version'
require 'bunny'

module Wumomq
  class Consumer
    def process delivery_info, properties, payload
      raise NotImplementedError, "Implement this method in a child class"

      # 应该返回2个值，succeed，abort
    end
  end

  class Base
    def initialize uri, queue_name, queue_num = 1, exchange_name = nil
      @conn      = Bunny.new uri
      @conn.start
      @channel   = @conn.create_channel
      @num       = queue_num
      @queue     = Array.new
      (0...@num).each { |i|
        @queue.push(@channel.queue("#{queue_name}.#{i}", :auto_delete => false))
      }

      @consumer  = nil
      @exchange  = @channel.default_exchange
    end

    def get_number_of_subscribe
      @num
    end

    def hash value
      Digest::MD5.hexdigest(value.to_s)[0...4].to_i(16) % @num
    end

    def publish options = {}
      (puts "没有消息主体(#{__FILE__}.#{__LINE__})"; return nil) if options[:message].nil? 
      (puts "没有指定key(#{__FILE__}.#{__LINE__})"; return nil) if options[:key].nil?
      q = hash(options[:key])
      @exchange.publish(options[:message], :routing_key => @queue[q].name)
    end

    def subscribe options = {}
      # 在登记的时候, key应该是整数
      (puts "没有处理的consumer(#{__FILE__}.#{__LINE__})"; return false) if options[:consumer].nil? 
      (puts "key为空(#{__FILE__}.#{__LINE__})"; return false) if options[:key].nil?

      @consumer = @queue[options[:key]].subscribe(:manual_ack => true, :block => false) do |delivery_info, properties, payload|
         succeed, abort = options[:consumer].process delivery_info, properties, payload
         @channel.acknowledge delivery_info.delivery_tag, false if succeed || abort
      end

      true
    end

    def cancel
      @consumer.cancel if @consumer
    end

{'fraction_non_alphanumeric': 0.1319052987598647, 'fraction_numerical': 0.004509582863585118, 'mean_word_length': 2.887185104052574, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 11, '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': '3338789', 'n_tokens_mistral': 1215, 'n_tokens_neox': 1186, 'n_words': 266}