<commit_before>require_relative 'app/consumer'
require_relative 'app/offset_stores/all'
require_relative 'app/parsers/all'
require_relative 'shared/runnable'

# Application
#
# * Registration of handlers to topics (Consumer)
# * Preforking of processes, one per consumer
# * Communication between parent and child process via Pipes

module Axe
  class App
    attr_reader :env, :logger, :exception_handler, :offset_store

    prepend Runnable

    DuplicateConsumerId = Class.new(StandardError)

    def initialize(options = {})
      @env       = options.fetch(:env, 'production')
      @logger    = options.fetch(:logger, nil)
      @consumers = []
      @exception_handler = options.fetch(:exception_handler, default_exception_handler)
      @offset_store      = options.fetch(:offset_store, nil)
      @pipes = {}
      set_procname("axe [master]")
    end

    def register(options = {})
      validate_options(options)

      from_parent, to_child = IO.pipe

      @pipes[options.fetch(:id)] = { to_child: to_child }

      @consumers << Consumer.new(id:      options.fetch(:id),
                                 handler: options.fetch(:handler),
                                 topic:   options.fetch(:topic),
                                 parser:  options.fetch(:parser, DefaultParser.new),
                                 env:     env,
                                 logger:  logger,
                                 exception_handler: exception_handler,
                                 offset_store:      offset_store,
                                 from_parent: from_parent)

      self
    end

    def consumers
      @consumers
    end

    def start
      return unless stopped?

      status(Started)

      @consumers.each do |c|
        fork do
          set_procname("axe [#{c.id}]")
          c.start
        end
      end

      @consumers.size.times { Process.wait }

      status(Stopped)
      self
    end

    def stop
      return unless started?
      status(Stopping)
      send_to_children('stop')
      self
    end<commit_msg>ensure to exit after forked process has finished<commit_after>require_relative 'app/consumer'
require_relative 'app/offset_stores/all'
require_relative 'app/parsers/all'
require_relative 'shared/runnable'

# Application
#
# * Registration of handlers to topics (Consumer)
# * Preforking of processes, one per consumer
# * Communication between parent and child process via Pipes

module Axe
  class App
    attr_reader :env, :logger, :exception_handler, :offset_store

    prepend Runnable

    DuplicateConsumerId = Class.new(StandardError)

    def initialize(options = {})
      @env       = options.fetch(:env, 'production')
      @logger    = options.fetch(:logger, nil)
      @consumers = []
      @exception_handler = options.fetch(:exception_handler, default_exception_handler)
      @offset_store      = options.fetch(:offset_store, nil)
      @pipes = {}
      set_procname("axe [master]")
    end

    def register(options = {})
      validate_options(options)

      from_parent, to_child = IO.pipe

      @pipes[options.fetch(:id)] = { to_child: to_child }

      @consumers << Consumer.new(id:      options.fetch(:id),
                                 handler: options.fetch(:handler),
                                 topic:   options.fetch(:topic),
                                 parser:  options.fetch(:parser, DefaultParser.new),
                                 env:     env,
                                 logger:  logger,
                                 exception_handler: exception_handler,
                                 offset_store:      offset_store,
                                 from_parent: from_parent)

      self
    end

    def consumers
      @consumers
    end

    def start
      return unless stopped?

      status(Started)

      @consumers.each do |c|
        fork do
          set_procname("axe [#{c.id}]")
          c.start
          exit!(true)
        end
      end

      @consumers.size.times { Process.wait }

      status(Stopped)
      self
    end

    def stop
      return unless started?
      status(Stopping)
      send_to_children('stop')
      self
    end
{'fraction_non_alphanumeric': 0.08848080133555926, 'fraction_numerical': 0.0, 'mean_word_length': 1.85500340367597, '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': 16, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6836171', 'n_tokens_mistral': 1151, 'n_tokens_neox': 1108, 'n_words': 259}