<commit_before>require 'set'
require 'em-hiredis'

module Juglite
  # Let's go for plain Rack y'all
  class App
    AsyncResponse = [-1, {}, []].freeze
    Headers = {'Content-Type' => 'text/event-stream;charset=utf-8'}

    # Options include:
    # +path+ : the URI path to listen to ('/stream')
    # +keepalive_timeout+ : the timeout in seconds between keepalive comments
    def initialize(app, options = {})
      @app = app
      @options = options.reverse_merge({
        path: '/stream',
        keepalive_timeout: 20
      })
      STDERR.puts "Registered Juglite to listen to #{@options[:path]}"
      @subscription_map = {}
      EventMachine::next_tick { setup_redis }
      EventMachine::next_tick { setup_keepalive }
    end

    def call(env)
      if env["PATH_INFO"] == @options[:path]
        handle_stream(env)
      else
        @app.call(env)
      end
    end

    private
    def handle_stream(env)
      request = Rack::Request.new(env)
      connection = SseConnection.new(request)

      # Get the headers out there, let the client know we're alive...
      EventMachine::next_tick do
        register_connection(connection)
        # Calling thin's Connection.post_process([status, headers, body])
        # This is how you start a response to the client asynchronously
        env['async.callback'].call [200, Headers, connection.body]
        connection.comment("registered to channels: #{channels_for_request(request).to_a.join(', ')}")
      end

      connection.callback { unregister_connection(connection) }
      connection.errback { unregister_connection(connection) }

      # Needed for crappy Rack::Lint
      throw :async

      # Returning a status of -1 keeps the connection open<commit_msg>Juglite can run as middleware OR application<commit_after>require 'set'
require 'em-hiredis'

module Juglite
  # Let's go for plain Rack y'all
  class App
    AsyncResponse = [-1, {}, []].freeze
    Headers = {'Content-Type' => 'text/event-stream;charset=utf-8'}

    # Options include:
    # +path+ : the URI path to listen to ('/stream')
    # +keepalive_timeout+ : the timeout in seconds between keepalive comments
    def initialize(app = nil, options = {})
      @app = app
      @options = {
        path: '/stream',
        keepalive_timeout: 20
      }.merge(options)
      STDERR.puts "Registered Juglite to listen to #{@options[:path]}"
      @subscription_map = {}
      EventMachine::next_tick { setup_redis }
      EventMachine::next_tick { setup_keepalive }
    end

    def call(env)
      if @app.nil? || (env["PATH_INFO"] == @options[:path])
        handle_stream(env)
      else
        @app.call(env)
      end
    end

    private
    def handle_stream(env)
      request = Rack::Request.new(env)
      connection = SseConnection.new(request)

      # Get the headers out there, let the client know we're alive...
      EventMachine::next_tick do
        register_connection(connection)
        # Calling thin's Connection.post_process([status, headers, body])
        # This is how you start a response to the client asynchronously
        env['async.callback'].call [200, Headers, connection.body]
        connection.comment("registered to channels: #{channels_for_request(request).to_a.join(', ')}")
      end

      connection.callback { unregister_connection(connection) }
      connection.errback { unregister_connection(connection) }

      # Needed for crappy Rack::Lint
      throw :async

      # Returning a status of -1 keeps the connection open
{'fraction_non_alphanumeric': 0.11054131054131054, 'fraction_numerical': 0.004558404558404558, 'mean_word_length': 2.8924611973392462, 'pattern_counts': {'":': 0, '<': 3, '<?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': '4249201', 'n_tokens_mistral': 1052, 'n_tokens_neox': 1015, 'n_words': 281}