Class: EventedMysql
- Inherits:
-
EM::Connection
- Object
- EM::Connection
- EventedMysql
- Defined in:
- lib/em/mysql.rb,
lib/em/mysql.rb
Constant Summary collapse
- DisconnectErrors =
[ 'query: not connected', 'MySQL server has gone away', 'Lost connection to MySQL server during query' ]
Instance Attribute Summary collapse
-
#connected ⇒ Object
readonly
Returns the value of attribute connected.
-
#opts ⇒ Object
(also: #settings)
readonly
Returns the value of attribute opts.
-
#processing ⇒ Object
readonly
Returns the value of attribute processing.
Class Method Summary collapse
-
._connect(opts) ⇒ Object
stolen from sequel.
- .all(query, type = nil, &blk) ⇒ Object
- .connect(*servers_opts) ⇒ Object
- .connection_pool ⇒ Object
- .execute(query, type = nil, cblk = nil, eblk = nil, &blk) ⇒ Object
- .reset! ⇒ Object
- .settings ⇒ Object
Instance Method Summary collapse
- #close ⇒ Object
- #execute(sql, response = nil, cblk = nil, eblk = nil, &blk) ⇒ Object
-
#initialize(mysql, opts) ⇒ EventedMysql
constructor
A new instance of EventedMysql.
- #notify_readable ⇒ Object
- #unbind ⇒ Object
Constructor Details
#initialize(mysql, opts) ⇒ EventedMysql
Returns a new instance of EventedMysql.
13 14 15 16 17 18 19 20 21 22 23 24 25 |
# File 'lib/em/mysql.rb', line 13 def initialize mysql, opts @mysql = mysql @fd = mysql.socket @opts = opts @current = nil @@queue ||= [] @processing = false @connected = true log 'mysql connected' make_socket_blocking EM.add_timer(0){ next_query } end |
Instance Attribute Details
#connected ⇒ Object (readonly)
Returns the value of attribute connected.
26 27 28 |
# File 'lib/em/mysql.rb', line 26 def connected @connected end |
#opts ⇒ Object (readonly) Also known as: settings
Returns the value of attribute opts.
26 27 28 |
# File 'lib/em/mysql.rb', line 26 def opts @opts end |
#processing ⇒ Object (readonly)
Returns the value of attribute processing.
26 27 28 |
# File 'lib/em/mysql.rb', line 26 def processing @processing end |
Class Method Details
._connect(opts) ⇒ Object
stolen from sequel
213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 |
# File 'lib/em/mysql.rb', line 213 def self._connect opts opts = settings.merge(opts) conn = Mysql.init # set encoding _before_ connecting if charset = opts[:charset] || opts[:encoding] conn.(Mysql::SET_CHARSET_NAME, charset) end conn.(Mysql::OPT_LOCAL_INFILE, 'client') conn.real_connect( opts[:host] || 'localhost', opts[:user] || 'root', opts[:password], opts[:database], opts[:port], opts[:socket], 0 + # XXX multi results require multiple callbacks to parse # Mysql::CLIENT_MULTI_RESULTS + # Mysql::CLIENT_MULTI_STATEMENTS + (opts[:compress] == false ? 0 : Mysql::CLIENT_COMPRESS) ) # increase timeout so mysql server doesn't disconnect us # this is especially bad if we're disconnected while EM.attach is # still in progress, because by the time it gets to EM, the FD is # no longer valid, and it throws a c++ 'bad file descriptor' error # (do not use a timeout of -1 for unlimited, it does not work on mysqld > 5.0.60) conn.query("set @@wait_timeout = #{opts[:timeout] || 2592000}") # we handle reconnecting (and reattaching the new fd to EM) conn.reconnect = false # By default, MySQL 'where id is null' selects the last inserted id # Turn this off. http://dev.rubyonrails.org/ticket/6778 conn.query("set SQL_AUTO_IS_NULL=0") # get results for queries conn.query_with_result = true conn rescue Mysql::Error => e if cb = opts[:on_error] cb.call(e) nil elsif opts[:failover] connect(opts[:failover]) else raise e end end |
.all(query, type = nil, &blk) ⇒ Object
292 293 294 295 296 297 298 299 300 |
# File 'lib/em/mysql.rb', line 292 def self.all query, type = nil, &blk responses = 0 connection_pool.each do |c| c.execute(query, type) do responses += 1 blk.call if blk and responses == @connection_pool.size end end end |
.connect(*servers_opts) ⇒ Object
196 197 198 199 200 201 202 203 204 205 206 207 208 |
# File 'lib/em/mysql.rb', line 196 def self.connect *servers_opts unless EM.respond_to?(:attach) and Mysql.method_defined?(:socket) raise RuntimeError, 'mysqlplus and EM.attach are required for EventedMysql' end opts = servers_opts.pop opts[:failover] = servers_opts if conn = _connect(opts) EM.attach conn.socket, self, conn, opts else EM.add_timer(5){ connect opts } end end |
.connection_pool ⇒ Object
302 303 304 305 306 307 308 309 |
# File 'lib/em/mysql.rb', line 302 def self.connection_pool @connection_pool ||= (1..settings[:connections]).map{ EventedMysql.connect(settings) } # p ['connpool', settings[:connections], @connection_pool.size] # (1..(settings[:connections]-@connection_pool.size)).each do # @connection_pool << EventedMysql.connect(settings) # end unless settings[:connections] == @connection_pool.size # @connection_pool end |
.execute(query, type = nil, cblk = nil, eblk = nil, &blk) ⇒ Object
274 275 276 277 278 279 280 281 282 |
# File 'lib/em/mysql.rb', line 274 def self.execute query, type = nil, cblk = nil, eblk = nil, &blk unless nil#connection = connection_pool.find{|c| not c.processing and c.connected } @n ||= 0 connection = connection_pool[@n] @n = 0 if (@n+=1) >= connection_pool.size end connection.execute(query, type, cblk, eblk, &blk) end |
.reset! ⇒ Object
311 312 313 314 315 316 |
# File 'lib/em/mysql.rb', line 311 def self.reset! @connection_pool.each do |c| c.close end @connection_pool = nil end |
.settings ⇒ Object
270 271 272 |
# File 'lib/em/mysql.rb', line 270 def self.settings @settings ||= { :connections => 4, :logging => false } end |
Instance Method Details
#close ⇒ Object
158 159 160 161 162 163 164 165 166 167 168 169 170 |
# File 'lib/em/mysql.rb', line 158 def close @connected = false @mysql.close # IO.pipe # EM.add_timer(0){ close_connection } # close_connection fd = detach if EM.reactor_running? if @io @io.close rescue nil @io = nil end log 'detached fd', fd end |
#execute(sql, response = nil, cblk = nil, eblk = nil, &blk) ⇒ Object
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 |
# File 'lib/em/mysql.rb', line 121 def execute sql, response = nil, cblk = nil, eblk = nil, &blk cblk ||= blk begin unless @processing or !@connected # begin # log 'mysql ping', @mysql.ping # # log 'mysql stat', @mysql.stat # # log 'mysql errno', @mysql.errno # rescue # log 'mysql ping failed' # @@queue << [response, sql, blk] # return close # end @processing = true log 'mysql sending', sql @mysql.send_query(sql) else @@queue << [response, sql, cblk, eblk] return end rescue Mysql::Error => e log 'mysql error', e. if DisconnectErrors.include? e. @@queue << [response, sql, cblk, eblk] return close else raise e end end log 'queuing', response, sql @current = [Time.now, response, sql, cblk, eblk] end |
#notify_readable ⇒ Object
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 80 81 82 83 84 85 86 87 88 89 90 91 92 |
# File 'lib/em/mysql.rb', line 35 def notify_readable log 'readable' if item = @current @current = nil start, response, sql, cblk, eblk = item log 'mysql response', Time.now-start, sql arg = case response when :raw result = @mysql.get_result @mysql.instance_variable_set('@cur_result', result) @mysql when :select ret = [] result = @mysql.get_result result.each_hash{|h| ret << h } log 'mysql result', ret ret when :update result = @mysql.get_result @mysql.affected_rows when :insert result = @mysql.get_result @mysql.insert_id else result = @mysql.get_result log 'got a result??', result if result nil end @processing = false # result.free if result.is_a? Mysql::Result next_query cblk.call(arg) if cblk else log 'readable, but nothing queued?! probably an ERROR state' return close end rescue Mysql::Error => e log 'mysql error', e. if e. =~ /Deadlock/ @@queue << [response, sql, cblk, eblk] @processing = false next_query elsif DisconnectErrors.include? e. @@queue << [response, sql, cblk, eblk] return close elsif cb = (eblk || @opts[:on_error]) cb.call(e) @processing = false next_query else raise e end # ensure # res.free if res.is_a? Mysql::Result # @processing = false # next_query end |
#unbind ⇒ Object
94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 |
# File 'lib/em/mysql.rb', line 94 def unbind log 'mysql disconnect', $!, *($! ? $!.backtrace[0..5] : []) # cp = EventedMysql.instance_variable_get('@connection_pool') and cp.delete(self) @connected = false # XXX wait for the next tick until the current fd is removed completely from the reactor # # XXX in certain cases the new FD# (@mysql.socket) is the same as the old, since FDs are re-used # XXX without next_tick in these cases, unbind will get fired on the newly attached signature as well # # XXX do _NOT_ use EM.next_tick here. if a bunch of sockets disconnect at the same time, we want # XXX reconnects to happen after all the unbinds have been processed EM.add_timer(0) do log 'mysql reconnecting' @processing = false @mysql = EventedMysql._connect @opts @fd = @mysql.socket @signature = EM.attach_fd @mysql.socket, true, false log 'mysql connected' EM.instance_variable_get('@conns')[@signature] = self @connected = true make_socket_blocking next_query end end |