Class: EmSequelAsync::Mysql::ClientPool
- Inherits:
-
Object
- Object
- EmSequelAsync::Mysql::ClientPool
- Defined in:
- lib/em-sequel-async/mysql.rb
Class Method Summary collapse
- .size ⇒ Object
-
.size=(value) ⇒ Object
Class Methods ======================================================.
- .trace ⇒ Object
- .trace=(value) ⇒ Object
- .trace? ⇒ Boolean
Instance Method Summary collapse
- #add(connection) ⇒ Object
- #delegate_query(connection, query, callback) ⇒ Object
-
#initialize(db) ⇒ ClientPool
constructor
Instance Methods ===================================================.
- #log(level, *args) ⇒ Object
- #logger ⇒ Object
- #pool_connection ⇒ Object
- #query(query, callback = nil, &block) ⇒ Object
- #remove(connection) ⇒ Object
Constructor Details
#initialize(db) ⇒ ClientPool
Instance Methods ===================================================
28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 |
# File 'lib/em-sequel-async/mysql.rb', line 28 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 |
Class Method Details
.size ⇒ Object
10 11 12 |
# File 'lib/em-sequel-async/mysql.rb', line 10 def self.size @size ||= 32 end |
.size=(value) ⇒ Object
Class Methods ======================================================
5 6 7 8 |
# File 'lib/em-sequel-async/mysql.rb', line 5 def self.size=(value) @size = value.to_i @size = 1 if (@size <= 0) end |
.trace ⇒ Object
18 19 20 |
# File 'lib/em-sequel-async/mysql.rb', line 18 def self.trace @trace end |
.trace=(value) ⇒ Object
22 23 24 |
# File 'lib/em-sequel-async/mysql.rb', line 22 def self.trace=(value) @trace = value end |
.trace? ⇒ Boolean
14 15 16 |
# File 'lib/em-sequel-async/mysql.rb', line 14 def self.trace? !!@trace end |
Instance Method Details
#add(connection) ⇒ Object
89 90 91 92 93 94 95 96 97 98 |
# File 'lib/em-sequel-async/mysql.rb', line 89 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 |
#delegate_query(connection, query, callback) ⇒ Object
105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 |
# File 'lib/em-sequel-async/mysql.rb', line 105 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) when Mysql2::Error case (err.) when /^Deadlock/i self.delegate_query(connection, query, callback) handled = true when /Duplicate entry/i callback and callback.call(nil, (Time.now - start).to_f, connection, err) self.add(connection) handled = true when /MySQL server has gone away|Lost connection/i self.remove(connection) EventMachine::Timer.new(10) do self.query(query, &callback) end handled = true end end unless (handled) log(:error, "(%.6fs) [ERR] %s (%s: %s)" % [ Time.now - start, query, err.class, err ]) log(:error, err.backtrace) callback and callback.call(false, (Time.now - start).to_f, connection, err) self.add(connection) end end rescue Mysql2::Error => err case (err.) when /MySQL server has gone away|Lost connection/i self.remove(connection) EventMachine::Timer.new(10) do self.query(query, &callback) end end end |
#log(level, *args) ⇒ Object
85 86 87 |
# File 'lib/em-sequel-async/mysql.rb', line 85 def log(level, *args) @options[:logger] and @options[:logger].send(level, *args) end |
#logger ⇒ Object
81 82 83 |
# File 'lib/em-sequel-async/mysql.rb', line 81 def logger @options[:logger] end |
#pool_connection ⇒ Object
171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 |
# File 'lib/em-sequel-async/mysql.rb', line 171 def pool_connection connection = @connection_pool.pop if (!connection and @connections.length < @connection_limit) connection = Mysql2::EM::Client.new(@options) end @connections_active[connection] = true connection rescue Mysql2::Error => err case (err.) when /Too many connections/i return else raise err end end |
#query(query, callback = nil, &block) ⇒ Object
191 192 193 194 195 196 197 198 199 200 201 202 203 |
# File 'lib/em-sequel-async/mysql.rb', line 191 def query(query, callback = nil, &block) callback ||= block if (connection = self.pool_connection) self.delegate_query(connection, query, callback) :executing else @query_queue << [ query, callback ] :queued end end |
#remove(connection) ⇒ Object
100 101 102 103 |
# File 'lib/em-sequel-async/mysql.rb', line 100 def remove(connection) @connections.delete(connection) @connections_active.delete(connection) end |