require File.dirname(__FILE__) + '/asymy'
module Asymy
# I'm thinking, one per connection
class Connection
def initialize(opts={})
@target = opts[:target]
@port = opts[:port]
@password = opts[:password].extend(StringX)
@database = opts[:database].extend(StringX)
@username = opts[:username].extend(StringX)
@queue = []
@state = :preauth
reco
end
# If I exec'd a statement right now, would it run right now? You probably
# don't care. Just call Connection#exec and hope.
def ready?
self.state == :ready
end
# Is there an error on the connection? Since I basically don't handle errors
# at all right now, and barely even catch them, your best bet is to give up.
def error?
self.error || false
end
# Queue up an SQL command, and, if the channel is open, send it. Takes a
# block argument that receives the results, in two arguments, fields (an array of hashes)
# and rows (an array of strings)
def exec(cmd, &block)
@queue << [cmd.extend(StringX), block, Commands::QUERY]
@fp.inject
end
def prepare(cmd, &block)
@queue << [cmd.extend(StringX), block, Commands::STMT_PREPARE]
@fp.inject
end
def execute_prepared(args, &block)
@queue << [args, block, Commands::STMT_EXECUTE]
@fp.inject
end
# no user-servicable parts below (attrs used to communicate with module)
attr_reader :password
attr_reader :database
attr_reader :username
attr_accessor :state
attr_accessor :error
attr_accessor :queue
private
# EM administrivia: the actual I/O for the connection is handled in a module mixed
# in to its own connection object. XXX factor this code out into Asymy#Connection, leave
# only stubs.
module Session
attr_accessor :bp
# EM's idiosyncratic initializer
def post_init; @framer = Framer.new; end
# receive 1-48739 bytes of data, which may contain one, two, zero, or 5.7
# MySQL packets.
def receive_data(buf)
@framer << buf
while @framer.complete?
num, packet = @framer.next_buffer
receive_packet(num, packet)
end
end
# receive a whole MySQL packet and run the state machine
def receive_packet(num, packet)
# special case errors until I waste the time to scan them to see if they're
# 4.0 or 4.1 packets. XXX
if packet[0].ord == 0xFF
self.error = packet[3..-1]
self.state = :error
end
def packet.eof?; self[0].ord == 0xfe; end
def packet.ok?; self[0].ord == 0x00; end
case self.state
when :preauth
handle_preauth(num, Packets::Greeting.new(packet))
when :auth_sent
handle_postauth(num, Packets::OK.new(packet))
# queries on a MySQL connection are synchronous. The response
# packets are:
# - ResultSet packet (which basically just says "OK")
# - Field packets (describing columns)
# - EOF (no more fields)
# - RowData packets (describing a row)
# - EOF (no more rows)
when :ready
inject
when :awaiting_result_set
# XXX just ignore for now
self.state = :awaiting_fields
when :awaiting_fields
if packet.eof?
self.state = :awaiting_rows
else
handle_field(num, Packets::Field.new(packet))
end
when :awaiting_rows
if packet.eof?
@cb.call(@fields, @rows)
@fields = nil
@rows = nil
ready!
else
# rows have a variable number of variable-length strings, and no other
# structure, so just hand the raw data off.
handle_row(num, packet)
end
when :awaiting_statement_handle
if packet.ok?
handle_statement_handle(num, Packets::PrepareOk.new(packet))
else
# XXX handle this case
@state = :error
end
when :awaiting_prepared_params
if packet.eof?
@state = :waiting_prepared_fields
else
# I cannot for the life of me figure out what I'm supposed
# to do with these --- using mysql-ruby, I can't get them
# to change based on their type. Why does MySQL send them?
# I figured it'd be to let me enforce types on the params.
end
when :awaiting_prepared_fields
if packet.eof?
@cb.call(@stmt)
@cb, @stmt, @expect_params, @expect_columns = nil, nil, nil, nil
ready!
else
# I guess I could cache these? But why bother? MySQL is just
# going to send them again. This protocol confuses and infuriates us!
end
when :error
pp self.error
else
raise "don't know how to handle"
end
end
def ready!; self.state = :ready; inject; end
def inject
return if not ready?
# this is going to get untidy real fast XXX
if(now = self.queue.slice!(0))
@cb = now[1]
case now[2]
when Commands::QUERY
self.state = :awaiting_result_set
p = Packets::Command.new
p.command = Commands::QUERY
p.arg = now[0]
send_data(p.marshall)
when Commands::STMT_PREPARE
self.state = :awaiting_statement_handle
p = Packets::Prepare.new
p.query = now[0]
send_data(p.marshall)
when Commands::STMT_EXECUTE
else
raise "wtf?"
end
end
end
def handle_preauth(num, greeting)
response = self.password.crypt(greeting.challenge_head + greeting.challenge_tail) unless self.password.empty?
response ||= "".extend(StringX)
a = Packets::Authenticate.new
a.client_flags = (Capabilities::LONG_PASSWORD |
Capabilities::LONG_FLAG |
Capabilities::CONNECT_WITH_DB |
Capabilities::LOCAL_FILES |
Capabilities::PROTOCOL_41 |
Capabilities::INTERACTIVE |
Capabilities::TRANSACTIONS |
Capabilities::SECURE_CONNECTION | # heh
Capabilities::MULTI_STATEMENTS |
Capabilities::MULTI_RESULTS)
a.charset_number = greeting.server_language
a.name = self.username
a.response = response
a.database = self.database
send_data(a.marshall(num+1))
self.state = :auth_sent
end
def handle_postauth(num, ok)
ready!
end
def handle_field(num, field)
@fields ||= []
fh = Hash.new
fh[:table] = field.table
fh[:name] = field.name
fh[:length] = field.length
fh[:type] = field.type
fh[:flags] = field.flags
fh[:decimals] = field.decimals
@fields << fh
end
def handle_row(num, row)
@rows ||= []
rv = []
while not row.empty?
rv << row.shift_lcstring
end
@rows << rv
end
def handle_statement_handle(num, ok)
@stmt = PreparedStatement.new(:backpointer => @bp,
:handle => ok.stmt_id)
@expect_params = @stmt.parameters
@expect_columns = @stmt.columns
if @expect_params > 0
@state = :awaiting_prepared_params
else
@state = :awaiting_prepared_columns
end
end
def method_missing(meth, *args); @bp.send(meth, *args); end
end
def reco
EventMachine::connect(@target, @port, Session) {|c| c.bp = self; @fp = c;}
end
public
end
end
) is private.
