<commit_before>require 'socket'

module Rack
  module QueueMetrics
    class Middleware

      def initialize(app)
        @app = app
        @addr = IPSocket.getaddress(Socket.gethostname).to_s + ':'+ENV['PORT']
        @instrument_name = "rack.queue-metrics"
        @should_notify = should_notify?
        Thread.new {report(1)}
      end

      def call(env)
        return @app.call(env) unless ENV['PORT']
        status, headers, body = @app.call(env)
        notify(raindrops_stats) if @should_notify
        [status, headers, body]
      end

    private

      def report(interval)
        loop do
          $stdout.puts(["measure=#{@instrument_name}",
            "addr=#{@addr}",
            "queue_depth=#{raindrops_stats[:requests][:queued]}"].join(' '))
          sleep(interval)
        end
      end

      def should_notify?
        if defined?(ActiveSupport::Notifications)
          ActiveSupport::Notifications.notifier.listening?(@instrument_name)
        end
      end

      def notify(data)
        ActiveSupport::Notifications.instrument(@instrument_name, data)
      end

      def raindrops_stats
        if defined? Raindrops::Linux.tcp_listener_stats
          stats = Raindrops::Linux.tcp_listener_stats([ '0.0.0.0:'+ENV['PORT'] ])['0.0.0.0:'+ENV['PORT']]
          return { :requests => { :active => stats.active, :queued => stats.queued }}
        else
          return { :requests => { :active => 0, :queued => 0 }}
        end
      end

    end
  end
end<commit_msg>move as notify to report loop<commit_after>require 'socket'

module Rack
  module QueueMetrics
    class Middleware

      def initialize(app)
        @app = app
        @addr = IPSocket.getaddress(Socket.gethostname).to_s + ':'+ENV['PORT']
        @instrument_name = "rack.queue-metrics"
        Thread.new {report(1)}
      end

      def call(env)
        return @app.call(env) unless ENV['PORT']
        status, headers, body = @app.call(env)
        [status, headers, body]
      end

    private

      def report(interval)
        loop do
          stats = raindrops_stats
          notify(stats) if should_notify
          $stdout.puts(["measure=#{@instrument_name}",
            "addr=#{@addr}",
            "queue_depth=#{stats[:requests][:queued]}"].join(' '))
          sleep(interval)
        end
      end

      def should_notify?
        if defined?(ActiveSupport::Notifications)
          ActiveSupport::Notifications.notifier.listening?(@instrument_name)
        end
      end

      def notify(data)
        ActiveSupport::Notifications.instrument(@instrument_name, data)
      end

      def raindrops_stats
        if defined? Raindrops::Linux.tcp_listener_stats
          stats = Raindrops::Linux.tcp_listener_stats([ '0.0.0.0:'+ENV['PORT'] ])['0.0.0.0:'+ENV['PORT']]
          return { :requests => { :active => stats.active, :queued => stats.queued }}
        else
          return { :requests => { :active => 0, :queued => 0 }}
        end
      end

    end
  end
end
{'fraction_non_alphanumeric': 0.1292153589315526, 'fraction_numerical': 0.007345575959933222, 'mean_word_length': 2.306843267108168, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 15, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 4, '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': '2245884', 'n_tokens_mistral': 933, 'n_tokens_neox': 910, 'n_words': 182}