<commit_before>class EmSequelAsync::Mysql
  class ClientPool
    # == Class Methods ======================================================
    
    def self.size=(value)
      @size = value.to_i
      @size = 1 if (@size <= 0)
    end
    
    def self.size
      @size ||= 32
    end
    
    def self.trace?
      !!@trace
    end
    
    def self.trace
      @trace
    end
    
    def self.trace=(value)
      @trace = value
    end
    
    # == Instance Methods ===================================================
    
    def initialize(db)
      @options = {
        :symbolize_keys => true,
        :cast_booleans => true
      }

      db.opts.each do |key, value|
        case (key)
        when :database, :username, :password, :host, :port, :socket, :encoding, :charset, :compress, :timeout
          @options[key] = value
        when :user
          @options[:username] = value
        when :loggers
          if (value and !value.empty?)
            @options[:logging] = true
            @options[:logger] = value.first
          end
        end
      end
      
      @query_queue = [ ]

      @connections = { }
      @connection_pool = [ ]
      @connections_active = { }
      
      @connection_limit = db.opts[:connections] || self.class.size
      
      if (EventMachine.reactor_running? and self.class.trace?)
        EventMachine::PeriodicTimer.new(1) do
          dump_file = "#{self.class.trace}.#{@options[:database]}"
          
          File.open(dump_file, 'w') do |f|
            f.puts @options[:database]

            @connections.each do |c, x|
              next unless (x)
              
              f.puts "\t#{c.inspect} -> #{x.inspect}"
            end
            @connection_pool.each do |c|
              f.puts "\t#{c.inspect} (pool)"
            end
          
            @query_queue.each do |query, callback|
              f.puts "\t#{query}"
            end
          end
        end
      end
    end
    
    def logger
      @options[:logger]
    end
    
    def log(level, *args)
      @options[:logger] and @options[:logger].send(level, *args)
    end
    
    def add(connection)
      @connections[connection] = nil
      
      if (@query_queue.empty?)
        @connections_active.delete(connection)
        @connection_pool << connection
      else
        self.delegate_query(connection, *@query_queue.pop)
      end
    end

    def remove(connection)
      @connections.delete(connection)
      @connections_active.delete(connection)
    end

    def delegate_query(connection, query, callback)
      @connections[connection] = [ query, callback ]

      start = Time.now
      deferrable = connection.query(query)
      
      deferrable.callback do |result|
        log(:debug, "(%.6fs) [OK] %s" % [ Time.now - start, query ])
        
        callback and callback.call(result, (Time.now - start).to_f, connection)

        self.add(connection)
      end

      deferrable.errback do |err|
        handled = false

        case (err)<commit_msg>Pass timezone options to mysql2 gem<commit_after>class EmSequelAsync::Mysql
  class ClientPool
    # == Class Methods ======================================================
    
    def self.size=(value)
      @size = value.to_i
      @size = 1 if (@size <= 0)
    end
    
    def self.size
      @size ||= 32
    end
    
    def self.trace?
      !!@trace
    end
    
    def self.trace
      @trace
    end
    
    def self.trace=(value)
      @trace = value
    end
    
    # == Instance Methods ===================================================
    
    def initialize(db)
      @db = db
      @options = {
        :symbolize_keys => true,
        :cast_booleans => true
      }

      db.opts.each do |key, value|
        case (key)
        when :database, :username, :password, :host, :port, :socket, :encoding, :charset, :compress, :timeout
          @options[key] = value
        when :user
          @options[:username] = value
        when :loggers
          if (value and !value.empty?)
            @options[:logging] = true
            @options[:logger] = value.first
          end
        end
      end
      
      @query_queue = [ ]

      @connections = { }
      @connection_pool = [ ]
      @connections_active = { }
      
      @connection_limit = db.opts[:connections] || self.class.size
      
      if (EventMachine.reactor_running? and self.class.trace?)
        EventMachine::PeriodicTimer.new(1) do
          dump_file = "#{self.class.trace}.#{@options[:database]}"
          
          File.open(dump_file, 'w') do |f|
            f.puts @options[:database]

            @connections.each do |c, x|
              next unless (x)
              
              f.puts "\t#{c.inspect} -> #{x.inspect}"
            end
            @connection_pool.each do |c|
              f.puts "\t#{c.inspect} (pool)"
            end
          
            @query_queue.each do |query, callback|
              f.puts "\t#{query}"
            end
          end
        end
      end
    end
    
    def logger
      @options[:logger]
    end
    
    def log(level, *args)
      @options[:logger] and @options[:logger].send(level, *args)
    end
    
    def add(connection)
      @connections[connection] = nil
      
      if (@query_queue.empty?)
        @connections_active.delete(connection)
        @connection_pool << connection
      else
        self.delegate_query(connection, *@query_queue.pop)
      end
    end

    def remove(connection)
      @connections.delete(connection)
      @connections_active.delete(connection)
    end

    def delegate_query(connection, query, callback)
      @connections[connection] = [ query, callback ]

      start = Time.now
      deferrable = connection.query(query, :database_timezone => @db.timezone, :application_timezone => Sequel.application_timezone)
      
      deferrable.callback do |result|
        log(:debug, "(%.6fs) [OK] %s" % [ Time.now - start, query ])
        
        callback and callback.call(result, (Time.now - start).to_f, connection)

        self.add(connection)
      end

      deferrable.errback do |err|
        handled = false

        case (err)
{'fraction_non_alphanumeric': 0.14564052605942523, 'fraction_numerical': 0.0021107322617308004, 'mean_word_length': 1.8012733060482038, 'pattern_counts': {'":': 0, '<': 9, '<?xml version=': 0, '>': 11, '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': '5297034', 'n_tokens_mistral': 1846, 'n_tokens_neox': 1790, 'n_words': 402}