<commit_before>require 'rake'
require 'rake/tasklib'

module FastlyNsq
  class RakeTask < Rake::TaskLib
    attr_accessor :name, :channel, :topics
    attr_writer :listener

    def initialize(*args, &task_block)
      @name = args.shift || :begin_listening
      add_rake_task_description_if_one_needed

      task(name, *args) do |_, task_args|
        RakeFileUtils.send(:verbose, verbose) do
          if block_given?
            yield(*[self, task_args].slice(0, task_block.arity))
          end

          @channel  ||= require_arg :channel, task_args
          @topics   ||= require_arg :topics, task_args
          @listener ||= task_args[:listener]

          listen_to_configured_topics
        end
      end
    end

    private

    def listen_to_configured_topics
      topic_per_thread do |topic, processor|
        logger.info "Listening to queue, topic:'#{topic}' and channel: '#{channel}'"
        listener.listen_to topic:     topic,
                           channel:   channel,
                           processor: processor
        logger.info "... done listening on topic:'#{topic}' and channel: '#{channel}'."
      end
    end

    def require_arg(arg, arg_list)
      arg_list.fetch(arg) { raise ArgumentError, "required configuration '#{arg}' is missing." }
    end

    def add_rake_task_description_if_one_needed
      unless ::Rake.application.last_description
        desc 'Listen to NSQ on topic using channel'
      end
    end

    def topic_per_thread
      topics.each do |(topic, processor)|
        Thread.new do
          yield topic, processor
        end
      end

      non_main_threads = (Thread.list - [Thread.main])
      non_main_threads.map(&:join)
    end

    def listener<commit_msg>Threads in the style of 8b55560<commit_after>require 'rake'
require 'rake/tasklib'

module FastlyNsq
  class RakeTask < Rake::TaskLib
    attr_accessor :name, :channel, :topics
    attr_writer :listener

    def initialize(*args, &task_block)
      @name = args.shift || :begin_listening
      add_rake_task_description_if_one_needed

      task(name, *args) do |_, task_args|
        RakeFileUtils.send(:verbose, verbose) do
          if block_given?
            yield(*[self, task_args].slice(0, task_block.arity))
          end

          @channel  ||= require_arg :channel, task_args
          @topics   ||= require_arg :topics, task_args
          @listener ||= task_args[:listener]

          listen_to_configured_topics
        end
      end
    end

    private

    def listen_to_configured_topics
      topic_per_thread do |topic, processor|
        logger.info "Listening to queue, topic:'#{topic}' and channel: '#{channel}'"
        listener.listen_to topic:     topic,
                           channel:   channel,
                           processor: processor
        logger.info "... done listening on topic:'#{topic}' and channel: '#{channel}'."
      end
    end

    def require_arg(arg, arg_list)
      arg_list.fetch(arg) { raise ArgumentError, "required configuration '#{arg}' is missing." }
    end

    def add_rake_task_description_if_one_needed
      unless ::Rake.application.last_description
        desc 'Listen to NSQ on topic using channel'
      end
    end

    def topic_per_thread
      listener_threads = []
      topics.each do |(topic, processor)|
        thread = Thread.new do
          yield topic, processor
        end
        thread.abort_on_exception = true
        listener_threads << thread
      end

      listener_threads.map(&:join)
    end

    def listener
{'fraction_non_alphanumeric': 0.08967851099830795, 'fraction_numerical': 0.0022560631697687537, 'mean_word_length': 2.3462264150943395, 'pattern_counts': {'":': 0, '<': 7, '<?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': '1332397', 'n_tokens_mistral': 1053, 'n_tokens_neox': 1023, 'n_words': 261}