<commit_before>require 'socket'
require 'eventmachine'
require 'eventmachine-tail'
require 'em-dns-resolver'
require 'syslog_protocol'

module RemoteSyslog
  class Reader < EventMachine::FileTail
    COLORED_REGEXP = /\e\[(?:(?:[34][0-7]|[0-9]);){0,2}(?:[34][0-7]|[0-9])m/

    def initialize(path, destination_address, destination_port, options = {})
      super(path, -1)

      @destination_address = destination_address
      @destination_port    = destination_port.to_i

      @parse_fields = options[:parse_fields]
      @strip_color = options[:strip_color]

      @socket = options[:socket] || EventMachine.open_datagram_socket('0.0.0.0', 0)

      @buffer = BufferedTokenizer.new

      @packet = SyslogProtocol::Packet.new

      local_hostname = options[:hostname] || (Socket.gethostname rescue `hostname`.chomp)
      if local_hostname.nil? || local_hostname.empty?
        local_hostname = 'localhost'
      end

      @packet.hostname = local_hostname
      @packet.facility = options[:facility] || 'user'
      @packet.severity = options[:severity] || 'notice'
      @packet.tag      = options[:program]  || File.basename(path) || File.basename($0)

      # Make sure the tag isn't too long
      if @packet.tag.length > 32
        @packet.tag = @packet.tag[0..31]
      end

      # Try to resolve the destination address
      resolve_destination_address

      # Every 60 seconds we'll see if the address has changed
      EventMachine.add_periodic_timer(60) do
        resolve_destination_address
      end
    end

    def resolve_destination_address
      request = EventMachine::DnsResolver.resolve(@destination_address)
      request.callback do |addrs|
        @cached_destination_ip = addrs.first
      end
    end

    def receive_data(data)
      @buffer.extract(data).each do |line|
        transmit(line)
      end
    end

    def destination_address
      @cached_destination_ip || @destination_address
    end

    def transmit(message)
      message = message.gsub(COLORED_REGEXP, '') if @strip_color

      packet = @packet.dup
      packet.content = message

      if @parse_fields
        if message =~ @parse_fields
          packet.hostname = $2
          packet.tag      = $3
          packet.content  = $4
        end
      end

      @socket.send_datagram(packet.assemble, destination_address, @destination_port)
    end
  end<commit_msg>Update code to match documentation.<commit_after>require 'socket'
require 'eventmachine'
require 'eventmachine-tail'
require 'em-dns-resolver'
require 'syslog_protocol'

module RemoteSyslog
  class Reader < EventMachine::FileTail
    COLORED_REGEXP = /\e\[(?:(?:[34][0-7]|[0-9]);){0,2}(?:[34][0-7]|[0-9])m/

    def initialize(path, destination_address, destination_port, options = {})
      super(path, -1)

      @destination_address = destination_address
      @destination_port    = destination_port.to_i

      @parse_fields = options[:parse_fields]
      @strip_color = options[:strip_color]

      @socket = options[:socket] || EventMachine.open_datagram_socket('0.0.0.0', 0)

      @buffer = BufferedTokenizer.new

      @packet = SyslogProtocol::Packet.new

      local_hostname = options[:hostname] || (Socket.gethostname rescue `hostname`.chomp)
      if local_hostname.nil? || local_hostname.empty?
        local_hostname = 'localhost'
      end

      @packet.hostname = local_hostname
      @packet.facility = options[:facility] || 'user'
      @packet.severity = options[:severity] || 'notice'
      @packet.tag      = options[:program]  || File.basename(path) || File.basename($0)

      # Make sure the tag isn't too long
      if @packet.tag.length > 32
        @packet.tag = @packet.tag[0..31]
      end

      # Try to resolve the destination address
      resolve_destination_address

      # Every 60 seconds we'll see if the address has changed
      EventMachine.add_periodic_timer(60) do
        resolve_destination_address
      end
    end

    def resolve_destination_address
      request = EventMachine::DnsResolver.resolve(@destination_address)
      request.callback do |addrs|
        @cached_destination_ip = addrs.first
      end
    end

    def receive_data(data)
      @buffer.extract(data).each do |line|
        transmit(line)
      end
    end

    def destination_address
      @cached_destination_ip || @destination_address
    end

    def transmit(message)
      message = message.gsub(COLORED_REGEXP, '') if @strip_color

      packet = @packet.dup
      packet.content = message

      if @parse_fields
        if message =~ @parse_fields
          packet.hostname = $2 if $2 && $2 != ''
          packet.tag      = $3 if $3 && $2 != ''
          packet.content  = $4 if $4 && $4 != ''
        end
      end

      @socket.send_datagram(packet.assemble, destination_address, @destination_port)
    end
  end
{'fraction_non_alphanumeric': 0.1111341059602649, 'fraction_numerical': 0.014900662251655629, 'mean_word_length': 2.8479299363057327, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, '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': '92788', 'n_tokens_mistral': 1495, 'n_tokens_neox': 1446, 'n_words': 312}