<commit_before># frozen_string_literal: true

class FastlyNsq::Manager
  DEADLINE = 30
  DEFAULT_POOL_SIZE = 5

  attr_reader :done, :pool, :logger

  def initialize(logger: FastlyNsq.logger, **pool_options)
    @done      = false
    @logger    = logger
    @pool      = FastlyNsq::PriorityThreadPool.new(
      { fallback_policy: :caller_runs, max_threads: DEFAULT_POOL_SIZE }.merge(pool_options),
    )
  end

  def topic_listeners
    @topic_listeners ||= {}
  end

  def topics
    topic_listeners.keys
  end

  def listeners
    topic_listeners.values
  end

  def terminate(deadline = DEADLINE)
    return if done

    stop_listeners

    return if pool.shutdown?

    stop_processing(deadline)

    @done = true
  end

  def stopped?
    done
  end

  def add_listener(listener)
    logger.info { "Listening to topic:'#{listener.topic}' on channel: '#{listener.channel}'" }

    if topic_listeners[listener.topic]
      logger.warn { "topic: #{listener.topic} was added more than once" }
    end

    topic_listeners[listener.topic] = listener
  end

  def transfer(new_manager, deadline: DEADLINE)
    new_manager.topic_listeners.merge!(topic_listeners)
    stop_processing(deadline)
    topic_listeners.clear
    @done = true
  end

  protected

  def stop_processing(deadline)
    logger.info { 'Stopping processors' }
    pool.shutdown

    logger.info { 'Waiting for processors to finish...' }
    return if pool.wait_for_termination(deadline)

    logger.info { 'Killing processors...' }
    pool.kill
  end

  def stop_listeners
    logger.info { 'Stopping listeners' }
    listeners.each(&:terminate)
    topic_listeners.clear
  end
end
<commit_msg>refactor: FastlyNsq::Manager#listeners returns a set of listeners<commit_after># frozen_string_literal: true

class FastlyNsq::Manager
  DEADLINE = 30
  DEFAULT_POOL_SIZE = 5

  attr_reader :done, :pool, :logger

  def initialize(logger: FastlyNsq.logger, **pool_options)
    @done      = false
    @logger    = logger
    @pool      = FastlyNsq::PriorityThreadPool.new(
      { fallback_policy: :caller_runs, max_threads: DEFAULT_POOL_SIZE }.merge(pool_options),
    )
  end

  def topic_listeners
    @topic_listeners ||= {}
  end

  def topics
    topic_listeners.keys
  end

  def listeners
    topic_listeners.values.to_set
  end

  def terminate(deadline = DEADLINE)
    return if done

    stop_listeners

    return if pool.shutdown?

    stop_processing(deadline)

    @done = true
  end

  def stopped?
    done
  end

  def add_listener(listener)
    logger.info { "Listening to topic:'#{listener.topic}' on channel: '#{listener.channel}'" }

    if topic_listeners[listener.topic]
      logger.warn { "topic: #{listener.topic} was added more than once" }
    end

    topic_listeners[listener.topic] = listener
  end

  def transfer(new_manager, deadline: DEADLINE)
    new_manager.topic_listeners.merge!(topic_listeners)
    stop_processing(deadline)
    topic_listeners.clear
    @done = true
  end

  protected

  def stop_processing(deadline)
    logger.info { 'Stopping processors' }
    pool.shutdown

    logger.info { 'Waiting for processors to finish...' }
    return if pool.wait_for_termination(deadline)

    logger.info { 'Killing processors...' }
    pool.kill
  end

  def stop_listeners
    logger.info { 'Stopping listeners' }
    listeners.each(&:terminate)
    topic_listeners.clear
  end
end

{'fraction_non_alphanumeric': 0.0905871938624963, 'fraction_numerical': 0.0017704337562702863, 'mean_word_length': 3.2803030303030303, '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': '5416828', 'n_tokens_mistral': 1131, 'n_tokens_neox': 1078, 'n_words': 250}