<commit_before>require 'pg'
module PG
  module EM
    class FeaturedDeferrable < ::EM::DefaultDeferrable
      def initialize(&blk)
        if block_given?
          callback(&blk)
          errback(&blk)
        end
      end
      
      def protect(fail_value = nil)
        yield
      rescue Exception => e
        ::EM.next_tick { fail(e) }
        fail_value
      end
      
      def protect_and_succeed(fail_value = nil)
        ret = yield
      rescue Exception => e
        ::EM.next_tick { fail(e) }
        fail_value
      else
        ::EM.next_tick { succeed(ret) }
        ret
      end
    end
    # == PostgreSQL EventMachine client
    #
    # Author:: <NAME> (mailto:<EMAIL>)
    # Licence:: MIT License
    #
    #
    # PG::EM::Client is a wrapper for PG::Connection which (re)defines methods:
    #
    # - +async_exec+ (alias: +async_query+)
    # - +async_prepare+
    # - +async_exec_prepared+
    #
    # and following:
    #
    # - +exec+ (alias: +query+)
    # - +exec_prepared+
    # - +prepare+
    #
    # which autodetects if EventMachine is running and uses appropriate
    # (async or sync) method version.
    #
    # Additionally to the above, there are asynchronous methods defined for
    # establishing connection and reseting it:
    #
    # - +Client.async_connect+
    # - +async_reset+
    #
    # They are async equivalents of +Client.connect+ (which is also aliased
    # by PG::Connection as +new+, +open+, +setdb+, +setdblogin+) and +reset+.
    #
    # Async methods might try to reset connection on connection error.
    # You won't even notice that (except for warning message from PG).
    # If you want to detect such event use +on_autoreconnect+ property.
    #
    # To disable such behavior set:
    #   client.async_autoreconnect = false
    #
    # or pass as new() hash argument:
    #   PG::EM::Client.new database: 'bar', async_autoreconnect: false
    #
    # Otherwise nothing changes in PG::Connect API.
    # See PG::Connect docs for arguments to above methods.
    #
    # *Warning:*
    #
    # +async_exec_prepared+ after +async_prepare+ should only be invoked on
    # the *same* connection.
    # If you are using connection pool, make sure to acquire single connection first.
    #
    class Client < PG::Connection


      # Connection timeout. Changing this property only affects
      # +Client.async_connect+ and +async_reset+.
      # However if passed as initialization option also affects blocking
      # +Client.connect+ and +reset+.
      attr_accessor :connect_timeout

      # (EXPERIMENTAL)
      # Aborts async command processing if waiting for response from server
      # exceedes +query_timeout+ seconds. This does not apply to
      # +Client.async_connect+ and +async_reset+. For those two use
      # +connect_timeout+ instead.
      #
      # To enable it set to seconds (> 0). To disable: set to 0.
      # You can also specify this as initialization option.
      attr_accessor :query_timeout

      # Enable/disable auto-reconnect feature (+true+/+false+).
      # Default is +true+.
      attr_accessor :async_autoreconnect

      # +on_autoreconnect+ is a user defined Proc that is called after a connection
      # with the server has been re-established.
      # It's invoked with +connection+ as first argument and original
      # +exception+ that caused the reconnecting process as second argument.
      #
      # Certain rules should apply to on_autoreconnect proc:
      #
      # - If proc returns +false+ (explicitly, +nil+ is ignored)
      #   the original +exception+ is passed to +Defferable#fail+ and the send
      #   query command is not invoked at all.
      # - If return value is an instance of exception it is passed to
      #   +Defferable#fail+ and the send query command is not invoked at all.
      # - If return value responds to +callback+ and +errback+ methods
      #   (like +Deferrable+), the send query command will be bound to this
      #   deferrable's success callback. Otherwise the send query command is
      #   called immediately after on_autoreconnect proc is executed.
      # - Other return values are ignored and the send query command is called
      #   immediately after on_autoreconnect proc is executed.
      #
      # You may pass this proc as +:on_autoreconnect+ option to PG::EM::Client.new.
      #
      # Example:
      #   pg.on_autoreconnect = proc do |conn, ex|
      #     conn.prepare("birds_by_name", "select id, name from animals order by name where species=$1", ['birds'])
      #   end
      #
      attr_accessor :on_autoreconnect

      # Used internally for marking connection as aborted on query timeout.
      attr_accessor :async_command_aborted

      module Watcher
        def initialize(client, deferrable, send_proc)
          @client = client
          @deferrable = deferrable
          @send_proc = send_proc
          if (timeout = client.query_timeout) > 0
            @notify_timestamp = Time.now
            setup_timer timeout
          else
            @timer = nil
          end
        end
        
        def setup_timer(timeout, adjustment = 0)
          @timer = ::EM::Timer.new(timeout - adjustment) do
            if (last_interval = Time.now - @notify_timestamp) >= timeout
              detach
              @client.async_command_aborted = true
              IO.for_fd(@client.socket).close # break connection now (hack)
              @deferrable.protect do
                raise PG::Error, "query timeout expired (async)"
              end
            else
              setup_timer timeout, last_interval
            end
          end
        end

        def notify_readable
          @notify_timestamp = Time.now if @timer
          @client.consume_input
          result = if @client.is_busy
            false
          else
            @timer.cancel if @timer
            detach
            @client.get_last_result
          end
        rescue PG::Error => e
          @client.async_autoreconnect!(@deferrable, e, &@send_proc)
        rescue Exception => e
          @deferrable.fail(e)
        else
          @deferrable.succeed(result) unless result == false
        end
      end

      module ConnectWatcher
        def initialize(client, deferrable, poll_method)
          @client = client
          @deferrable = deferrable
          @poll_method = :"#{poll_method}_poll"
          if (timeout = client.connect_timeout) > 0
            @timer = ::EM::Timer.new(timeout) do
              detach
              @deferrable.protect do
                raise PG::Error, "timeout expired (async)"
              end
            end<commit_msg>multiple statements are non-blocking thanks to #get_result + leave connection in not busy state after query_timeout<commit_after>require 'pg'
module PG
  class Result
    class << self
      unless method_defined? :check_result
        # pg_check_result is internal ext function. I whish it was rubified.
        # https://bitbucket.org/ged/ruby-pg/issue/123/rubify-pg_check_result
        def check_result(connection, result)
          if result.nil?
            if (error_message = connection.error_message)
              error = PG::Error.new(error_message)
            end
          else
            case result.result_status
              when PG::PGRES_BAD_RESPONSE,
                   PG::PGRES_FATAL_ERROR,
                   PG::PGRES_NONFATAL_ERROR
                error = PG::Error.new(result.error_message)
                error.instance_variable_set('@result', result)
            end
          end
          if error
            error.instance_variable_set('@connection', connection)
            raise error 
          end
        end
      end
    end
  end
  module EM
    class FeaturedDeferrable < ::EM::DefaultDeferrable
      def initialize(&blk)
        if block_given?
          callback(&blk)
          errback(&blk)
        end
      end
      
      def protect(fail_value = nil)
        yield
      rescue Exception => e
        ::EM.next_tick { fail(e) }
        fail_value
      end
      
      def protect_and_succeed(fail_value = nil)
        ret = yield
      rescue Exception => e
        ::EM.next_tick { fail(e) }
        fail_value
      else
        ::EM.next_tick { succeed(ret) }
        ret
      end
    end
    # == PostgreSQL EventMachine client
    #
    # Author:: <NAME> (mailto:<EMAIL>)
    # Licence:: MIT License
    #
    #
    # PG::EM::Client is a wrapper for PG::Connection which (re)defines methods:
    #
    # - +async_exec+ (alias: +async_query+)
    # - +async_prepare+
    # - +async_exec_prepared+
    #
    # and following:
    #
    # - +exec+ (alias: +query+)
    # - +exec_prepared+
    # - +prepare+
    #
    # which autodetects if EventMachine is running and uses appropriate
    # (async or sync) method version.
    #
    # Additionally to the above, there are asynchronous methods defined for
    # establishing connection and reseting it:
    #
    # - +Client.async_connect+
    # - +async_reset+
    #
    # They are async equivalents of +Client.connect+ (which is also aliased
    # by PG::Connection as +new+, +open+, +setdb+, +setdblogin+) and +reset+.
    #
    # Async methods might try to reset connection on connection error.
    # You won't even notice that (except for warning message from PG).
    # If you want to detect such event use +on_autoreconnect+ property.
    #
    # To disable such behavior set:
    #   client.async_autoreconnect = false
    #
    # or pass as new() hash argument:
    #   PG::EM::Client.new database: 'bar', async_autoreconnect: false
    #
    # Otherwise nothing changes in PG::Connect API.
    # See PG::Connect docs for arguments to above methods.
    #
    # *Warning:*
    #
    # +async_exec_prepared+ after +async_prepare+ should only be invoked on
    # the *same* connection.
    # If you are using connection pool, make sure to acquire single connection first.
    #
    class Client < PG::Connection


      # Connection timeout. Changing this property only affects
      # +Client.async_connect+ and +async_reset+.
      # However if passed as initialization option also affects blocking
      # +Client.connect+ and +reset+.
      attr_accessor :connect_timeout

      # (EXPERIMENTAL)
      # Aborts async command processing if waiting for response from server
      # exceedes +query_timeout+ seconds. This does not apply to
      # +Client.async_connect+ and +async_reset+. For those two use
      # +connect_timeout+ instead.
      #
      # To enable it set to seconds (> 0). To disable: set to 0.
      # You can also specify this as initialization option.
      attr_accessor :query_timeout

      # Enable/disable auto-reconnect feature (+true+/+false+).
      # Default is +true+.
      attr_accessor :async_autoreconnect

      # +on_autoreconnect+ is a user defined Proc that is called after a connection
      # with the server has been re-established.
      # It's invoked with +connection+ as first argument and original
      # +exception+ that caused the reconnecting process as second argument.
      #
      # Certain rules should apply to on_autoreconnect proc:
      #
      # - If proc returns +false+ (explicitly, +nil+ is ignored)
      #   the original +exception+ is passed to +Defferable#fail+ and the send
      #   query command is not invoked at all.
      # - If return value is an instance of exception it is passed to
      #   +Defferable#fail+ and the send query command is not invoked at all.
      # - If return value responds to +callback+ and +errback+ methods
      #   (like +Deferrable+), the send query command will be bound to this
      #   deferrable's success callback. Otherwise the send query command is
      #   called immediately after on_autoreconnect proc is executed.
      # - Other return values are ignored and the send query command is called
      #   immediately after on_autoreconnect proc is executed.
      #
      # You may pass this proc as +:on_autoreconnect+ option to PG::EM::Client.new.
      #
      # Example:
      #   pg.on_autoreconnect = proc do |conn, ex|
      #     conn.prepare("birds_by_name", "select id, name from animals order by name where species=$1", ['birds'])
      #   end
      #
      attr_accessor :on_autoreconnect

      # Used internally for marking connection as aborted on query timeout.
      attr_accessor :async_command_aborted

      module Watcher
        def initialize(client, deferrable, send_proc)
          @last_result = nil
          @client = client
          @deferrable = deferrable
          @send_proc = send_proc
          if (timeout = client.query_timeout) > 0
            @notify_timestamp = Time.now
            setup_timer timeout
          else
            @timer = nil
          end
        end
        
        def setup_timer(timeout, adjustment = 0)
          @timer = ::EM::Timer.new(timeout - adjustment) do
            if (last_interval = Time.now - @notify_timestamp) >= timeout
              detach
              @client.async_command_aborted = true
              @client.consume_input
              IO.for_fd(@client.socket).close # break connection now (hack)
              @deferrable.protect do
                while result = @client.get_result
                  result.clear
                end
                raise PG::Error, "query timeout expired (async)"
              end
            else
              setup_timer timeout, last_interval
            end
          end
        end

        def notify_readable
          result = false
          @client.consume_input
          until @client.is_busy && !self.error?
            break if (single_result = @client.get_result).nil?
            @last_result.clear if @last_result
            @last_result = single_result
          end
          unless @client.is_busy
            result = @last_result
            Result.check_result(@client, result)
            detach
            @timer.cancel if @timer
          end
        rescue Exception => e
          detach
          @timer.cancel if @timer
          if e.is_a?(PG::Error)
            @client.async_autoreconnect!(@deferrable, e, &@send_proc)
          else
            @deferrable.fail(e)
          end
        else
          if result == false
            @notify_timestamp = Time.now if @timer
          else
            @deferrable.succeed(result) 
          end
        end
      end

      module ConnectWatcher
        def initialize(client, deferrable, poll_method)
          @client = client
          @deferrable = deferrable
          @poll_method = :"#{poll_method}_poll"
          if (timeout = client.connect_timeout) > 0
            @timer = ::EM::Timer.new(timeout) do
              detach
              @deferrable.protect do
                raise PG::Error, "timeout expired (async)"
              end
            end
{'fraction_non_alphanumeric': 0.08011310084825636, 'fraction_numerical': 0.0010098290022889458, 'mean_word_length': 2.032251479893856, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 22, 'https://': 1, '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': '288315', 'n_tokens_mistral': 4154, 'n_tokens_neox': 4017, 'n_words': 1322}