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 |
# File 'lib/em-sequel-async/mysql.rb', line 28 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 |
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
88 89 90 91 92 93 94 95 96 97 |
# File 'lib/em-sequel-async/mysql.rb', line 88 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
104 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 |
# File 'lib/em-sequel-async/mysql.rb', line 104 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) 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 @query_queue << [ query, callback ] self.remove(connection) 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 @query_queue << [ query, callback ] self.remove(connection) handled = true end end |
#log(level, *args) ⇒ Object
84 85 86 |
# File 'lib/em-sequel-async/mysql.rb', line 84 def log(level, *args) @options[:logger] and @options[:logger].send(level, *args) end |
#logger ⇒ Object
80 81 82 |
# File 'lib/em-sequel-async/mysql.rb', line 80 def logger @options[:logger] end |
#pool_connection ⇒ Object
164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 |
# File 'lib/em-sequel-async/mysql.rb', line 164 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
184 185 186 187 188 189 190 191 192 193 194 195 196 |
# File 'lib/em-sequel-async/mysql.rb', line 184 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
99 100 101 102 |
# File 'lib/em-sequel-async/mysql.rb', line 99 def remove(connection) @connections.delete(connection) @connections_active.delete(connection) end |