<commit_before>require 'celluloid/io'

module Listen
  module Adapter

    # Adapter to receive file system modifications over TCP
    class TCP < Base
      include Celluloid::IO

      finalizer :finalize

      attr_reader :buffer, :socket

      def self.usable?
        true
      end

      # Initializes and starts a Celluloid::IO-powered TCP-recipient
      def start
        # Passing a regular TCPSocket to Celluloid::IO's seems
        # to work around an issue with Errno::ECONNREFUSED on
        # local addresses
        @socket = TCPSocket.new ::TCPSocket.new(listener.host, listener.port)
        @buffer = String.new
        run
      end

      # Cleans up buffer and socket
      def finalize
        @buffer = nil
        if @socket
          @socket.close
          @socket = nil
        end
      end

      private

      # Number of bytes to receive at a time
      RECEIVE_WINDOW = 1024

      # Continuously receive and asynchronously handle data
      def run
        loop do
          async.handle_data(@socket.recv(RECEIVE_WINDOW))
        end
      end

      # Buffers incoming data and dispatches messages accordingly
      def handle_data data
        @buffer << data
        while message = Listen::TCP::Message.from_buffer(@buffer)
          message.object.flatten.each do |path|<commit_msg>Workaround no longer required as broadcaster now listens on all addresses by default<commit_after>require 'celluloid/io'

module Listen
  module Adapter

    # Adapter to receive file system modifications over TCP
    class TCP < Base
      include Celluloid::IO

      finalizer :finalize

      attr_reader :buffer, :socket

      def self.usable?
        true
      end

      # Initializes and starts a Celluloid::IO-powered TCP-recipient
      def start
        @socket = TCPSocket.new(listener.host, listener.port)
        @buffer = String.new
        run
      end

      # Cleans up buffer and socket
      def finalize
        @buffer = nil
        if @socket
          @socket.close
          @socket = nil
        end
      end

      private

      # Number of bytes to receive at a time
      RECEIVE_WINDOW = 1024

      # Continuously receive and asynchronously handle data
      def run
        loop do
          async.handle_data(@socket.recv(RECEIVE_WINDOW))
        end
      end

      # Buffers incoming data and dispatches messages accordingly
      def handle_data data
        @buffer << data
        while message = Listen::TCP::Message.from_buffer(@buffer)
          message.object.flatten.each do |path|
{'fraction_non_alphanumeric': 0.057511737089201875, 'fraction_numerical': 0.003129890453834116, 'mean_word_length': 2.033214709371293, 'pattern_counts': {'":': 0, '<': 9, '<?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': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '4006624', 'n_tokens_mistral': 707, 'n_tokens_neox': 669, 'n_words': 231}