<commit_before>require 'rubygems'
require 'lib/net/messagepacket'
require 'eventmachine'

module LogStash; module Net
  # The MessageClient class exists only as an alias
  # to the MessageSocketMux. You should use the
  # client class if you are implementing a client.
  class MessageSocket < EventMachine::Connection
    # connection init callback from EventMachine::Connection
    def post_init
      #set_comm_inactivity_timeout(30)
      @buffer = ""
    end

    # data receiver callback from EventMachine::Connection
    def receive_data(data)
      @buffer += data

      len = 0
      count = 0
      MessagePacket.each(@buffer) do |packet|
        len += packet.length
        count += 1
        obj = JSON::load(packet.content)
        msg = Message.new_from_data(obj)

        if !@handler
          $stderr.puts "No message handler set. Can't handle #{msg.class.name}"
          next
        end

        name = msg.class.name.split(":")[-1]
        func = "#{name}Handler"
        if @handler.respond_to?(func):
          operation = lambda do 
            @handler.send(func, msg) do |response|
              sendmsg(response)
            end
          end
          EventMachine.defer(operation, nil)
          #@handler.send(func, msg) do |response|
           #sendmsg(response)
          #end
        else
          $stderr.puts "#{@handler.class.name} does not support #{func}"
        end
      end

      if len > 0
        puts "Removing #{len} bytes (#{count} packets)"
        @buffer[0 .. len - 1] = ""
        sleep 1
      end
    end # def receive_data

    def sendmsg(msg)
      if msg.is_a?(RequestMessage) and msg.id == nil
        msg.generate_id!
      end<commit_msg>- Don't use EventMachine.defer for running message handlers. Turns out it's   faster if we don't push the call off to another thread.<commit_after>require 'rubygems'
require 'lib/net/messagepacket'
require 'eventmachine'

module LogStash; module Net
  # The MessageClient class exists only as an alias
  # to the MessageSocketMux. You should use the
  # client class if you are implementing a client.
  class MessageSocket < EventMachine::Connection
    # connection init callback from EventMachine::Connection
    def post_init
      #set_comm_inactivity_timeout(30)
      @buffer = ""
    end

    # data receiver callback from EventMachine::Connection
    def receive_data(data)
      @buffer += data

      len = 0
      count = 0
      MessagePacket.each(@buffer) do |packet|
        len += packet.length
        count += 1
        obj = JSON::load(packet.content)
        msg = Message.new_from_data(obj)

        if !@handler
          $stderr.puts "No message handler set. Can't handle #{msg.class.name}"
          next
        end

        name = msg.class.name.split(":")[-1]
        func = "#{name}Handler"
        if @handler.respond_to?(func):
          #operation = lambda do 
            #@handler.send(func, msg) do |response|
              #sendmsg(response)
            #end
          #end
          #EventMachine.defer(operation, nil)
          
          # We actually get better performance if we don't defer processing
          # to another thread. This should be done carefully, though, as
          # blocking here will block the receiving thread for this socket
          # (maybe for all of eventmachine?).
          @handler.send(func, msg) do |response|
            sendmsg(response)
          end
        else
          $stderr.puts "#{@handler.class.name} does not support #{func}"
        end
      end

      if len > 0
        puts "Removing #{len} bytes (#{count} packets)"
        @buffer[0 .. len - 1] = ""
      end
    end # def receive_data

    def sendmsg(msg)
      if msg.is_a?(RequestMessage) and msg.id == nil
        msg.generate_id!
      end
{'fraction_non_alphanumeric': 0.0949367088607595, 'fraction_numerical': 0.0050105485232067514, 'mean_word_length': 2.1115668580803937, 'pattern_counts': {'":': 2, '<': 5, '<?xml version=': 0, '>': 5, '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': '1500307', 'n_tokens_mistral': 1080, 'n_tokens_neox': 1048, 'n_words': 337}