<commit_before>require_relative '../mforwd'

class MForwd::Worker
  include MForwd::Logging

  def initialize
    @config = MForwd::Config.new
    @logger = logger config: @config.log
    @buffer = MForwd::Buffer.new config: @config, role: :server
    @item_merger = MForwd::Item::Merge.new config: @config.deliver['item']
    @interval = 5
  end

  def run
    Signal.trap :INT, sighandler(:shutdown)
    @running = true
    @waiting = 0
    while @running
      data = @buffer.pop_all
      if data.empty?
        @waiting += 1
        @logger.debug "#{@waiting} No queue in buffer."
        sleep @interval
        next
      end

      @logger.debug "Queued in buffer. wait interval"
      @waiting = 0
      sleep @config.deliver['min_interval']

      data.concat(@buffer.pop_all)
      list = data.map do |d|
        MForwd::Item.decode(d)
      end
      @logger.debug "Fetched from buffer:\n  #{list}"
      messages = @item_merger.merge_into_messages list
      @logger.debug "Messages to deliver:\n  #{messages}"
      sleep @interval
    end
    Signal.trap :INT, :DEFAULT
  end

  private

    def sighandler sym=:shutdown
      @sighandlers ||= {
        shutdown: Proc.new { |sig|
          p "Signal #{sig} trapped. Exiting ..." if ENV['MFORWD_DEBUG']
          @running = false
        },
      }
      @sighandlers[sym]
    end
end<commit_msg>Send Messages by Deliver class<commit_after>require_relative '../mforwd'

class MForwd::Worker
  include MForwd::Logging

  def initialize
    @config      = MForwd::Config.new
    @logger      = logger config: @config.log
    @buffer      = MForwd::Buffer.new config: @config, role: :server
    @deliver     = MForwd::Deliver.new
    @item_merger = MForwd::Item::Merge.new config: @config.deliver['item']
    @interval    = 5
  end

  def run
    Signal.trap :INT, sighandler(:shutdown)
    @running = true
    @waiting = 0
    while @running
      data = @buffer.pop_all
      if data.empty?
        @waiting += 1
        @logger.debug "#{@waiting} No queue in buffer."
        sleep @interval
        next
      end

      @logger.debug "Queued in buffer. wait interval"
      @waiting = 0
      sleep @config.deliver['min_interval']

      data.concat(@buffer.pop_all)
      list = data.map do |d|
        MForwd::Item.decode(d)
      end
      @logger.debug "Fetched from buffer:\n  #{list}"
      messages = @item_merger.merge_into_messages list
      @logger.debug "Messages to deliver:\n  #{messages}"
      messages.each do |msg|
        @deliver.deliver msg
      end
      sleep @interval
    end
    Signal.trap :INT, :DEFAULT
  end

  private

    def sighandler sym=:shutdown
      @sighandlers ||= {
        shutdown: Proc.new { |sig|
          p "Signal #{sig} trapped. Exiting ..." if ENV['MFORWD_DEBUG']
          @running = false
        },
      }
      @sighandlers[sym]
    end
end
{'fraction_non_alphanumeric': 0.11731843575418995, 'fraction_numerical': 0.002793296089385475, 'mean_word_length': 2.281786941580756, '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': '4078952', 'n_tokens_mistral': 953, 'n_tokens_neox': 928, 'n_words': 233}