File Coverage

blib/lib/Net/Wire10.pm

Criterion	Covered	Total	%
statement	152	1035	14.6
branch	4	370	1.0
condition	0	66	0.0
subroutine	42	141	29.7
pod	12	12	100.0
total	210	1624	12.9

package Net::Wire10;

use strict;
use warnings;
use IO::Socket;
use IO::Select;
use utf8;
use Encode;
use vars qw($VERSION $DEBUG);

use constant {
	DEFAULT_PORT_NUMBER      => 3306,
	DEFAULT_CONNECT_TIMEOUT  => 5,
	DEFAULT_QUERY_TIMEOUT    => 30,
};

$VERSION = '1.08';

use constant STREAM_BUFFER_LENGTH => 65536;
use constant MACKET_HEADER_LENGTH => 4;
use constant TIMEOUT_GRANULARITY => 1;

BEGIN {
	package Net::Wire10;

	# Macket (MySQL messages) types.
	our %MACKET_NAMES = (
		   1 => 'HANDSHAKE',
		   2 => 'AUTHENTICATE',
		   4 => 'OK',
		   8 => 'ERROR',
		  16 => 'COMMAND',
		  32 => 'RESULT_SET_HEADER',
		  64 => 'COLUMN_INFO',
		 128 => 'EOF',
		 256 => 'ROW_DATA',
		 512 => 'MORE_DATA',
	);

	# Type of commands that can be sent to the server.
	# Only QUIT, QUERY and PING are currently used.
	our %COMMAND_NAMES = (
		"\x00" => 'SLEEP',
		"\x01" => 'QUIT',
		"\x02" => 'INIT_DB',
		"\x03" => 'QUERY',
		"\x04" => 'FIELD_LIST',
		"\x05" => 'CREATE_DB',
		"\x06" => 'DROP_DB',
		"\x07" => 'REFRESH',
		"\x08" => 'SHUTDOWN',
		"\x09" => 'STATISTICS',
		"\x0A" => 'PROCESS_INFO',
		"\x0B" => 'CONNECT',
		"\x0C" => 'PROCESS_KILL',
		"\x0D" => 'DEBUG',
		"\x0E" => 'PING',
		"\x0F" => 'TIME',
		"\x10" => 'DELAYED_INSERT',
		"\x11" => 'CHANGE_USER',
		"\x12" => 'BINLOG_DUMP',
		"\x13" => 'TABLE_DUMP',
		"\x14" => 'CONNECT_OUT',
		"\x15" => 'REGISTER_SLAVE',
		"\x16" => 'STMT_PREPARE',
		"\x17" => 'STMT_EXECUTE',
		"\x18" => 'STMT_SEND_LONG_DATA',
		"\x19" => 'STMT_CLOSE',
		"\x1A" => 'STMT_RESET',
		"\x1B" => 'SET_OPTION',
		"\x1C" => 'STMT_FETCH',
		"\x1D" => 'DAEMON',
	);

	# Per-connection flags, some of which are sent to the
	# server during handshake to convey client capabilities.
	our %FLAG_NAMES = (
		0x00000001 => 'LONG_PASSWORD',
		0x00000002 => 'FOUND_ROWS',
		0x00000004 => 'LONG_FLAG',
		0x00000008 => 'CONNECT_WITH_DB',
		0x00000010 => 'NO_SCHEMA',
		0x00000020 => 'COMPRESS',
		0x00000040 => 'ODBC',
		0x00000080 => 'LOCAL_FILES',
		0x00000100 => 'IGNORE_SPACE',
		0x00000200 => 'PROTOCOL_41',
		0x00000400 => 'INTERACTIVE',
		0x00000800 => 'SSL',
		0x00001000 => 'IGNORE_SIGPIPE',
		0x00002000 => 'TRANSACTIONS',
		0x00004000 => 'RESERVED',
		0x00008000 => 'SECURE_CONNECTION',
		0x00010000 => 'MULTI_STATEMENTS',
		0x00020000 => 'MULTI_RESULTS',
		0x40000000 => 'SSL_VERIFY_SERVER_CERT',
		0x80000000 => 'REMEMBER_OPTIONS',
	);

	our %UTF8_COLLATIONS = (
		33 => 'GENERAL_CI',
		83 => 'BIN',
		192 => 'UNICODE_CI',
		193 => 'ICELANDIC_CI',
		194 => 'LATVIAN_CI',
		195 => 'ROMANIAN_CI',
		196 => 'SLOVENIAN_CI',
		197 => 'POLISH_CI',
		198 => 'ESTONIAN_CI',
		199 => 'SPANISH_CI',
		200 => 'SWEDISH_CI',
		201 => 'TURKISH_CI',
		202 => 'CZECH_CI',
		203 => 'DANISH_CI',
		204 => 'LITHUANIAN_CI',
		205 => 'SLOVAK_CI',
		206 => 'SPANISH2_CI',
		207 => 'ROMAN_CI',
		208 => 'PERSIAN_CI',
		209 => 'ESPERANTO_CI',
		210 => 'HUNGARIAN_CI',
	);

	our %DATA_TYPES = (
		0 => 'TEXT',
		1 => 'BINARY',
	);

	our %SERVER_STATUS = (
		0x00000001 => 'IN_TRANS',
		0x00000002 => 'AUTOCOMMIT',
		0x00000004 => 'MORE_RESULTS',
		0x00000008 => 'MORE_RESULTS_EXISTS',
		0x00000010 => 'NO_GOOD_INDEX_USED',
		0x00000020 => 'NO_INDEX_USED',
		0x00000040 => 'CURSOR_EXISTS',
		0x00000080 => 'LAST_ROW_SENT',
		0x00000100 => 'DB_DROPPED',
		0x00000200 => 'NO_BACKSLASH_ESCAPES',
		0x00000400 => 'METADATA_CHANGED',
	);

	our %COLUMN_TYPES = (
		0 => 'DECIMAL',
		1 => 'TINY',
		2 => 'SHORT',
		3 => 'LONG',
		4 => 'FLOAT',
		5 => 'DOUBLE5',
		6 => 'NULL',
		7 => 'TIMESTAMP',
		8 => 'LONGLONG',
		9 => 'INT24',
		10 => 'DATE',
		11 => 'TIME',
		12 => 'DATETIME',
		13 => 'YEAR',
		14 => 'NEWDATE',
		15 => 'VARCHAR',
		16 => 'BIT',
		246 => 'NEWDECIMAL',
		247 => 'ENUM',
		248 => 'SET',
		249 => 'TINY_BLOB',
		250 => 'MEDIUM_BLOB',
		251 => 'LONG_BLOB',
		252 => 'BLOB',
		253 => 'VAR_STRING',
		254 => 'STRING',
		255 => 'GEOMETRY',
	);

	our %COLUMN_FLAGS = (
		0x000001 => 'NOT_NULL',           # Column has a NOT NULL constraint
		# ... Skip flags that provide incomplete and/or ambiguous information under nearly all circumstances ...
		# ... Skip deprecated flags ...
		0x000020 => 'UNSIGNED',           # Column data type is unsigned
		0x000040 => 'ZEROFILL',           # Column was created with the ZEROFILL flag
		# ... Skip deprecated flags ...
		0x000200 => 'AUTO_INCREMENT',     # Default field values for this column are generated by a thread-safe sequencer
		# ... Skip deprecated flags ...
		0x001000 => 'NO_DEFAULT_VALUE',   # Column does not have a default value
		0x002000 => 'ON_UPDATE_NOW',      # Column default value is NOW() when an UPDATE affects the column
		# ... Skip deprecated flags, internal and temporary flags ...
	);

	# Enumerations that are currently not included here:
	#  * Server capabilities
	#  * Server language
	#  * Server error code
	#  * Server error state
	#  * Column flags (complete)
	#  * Collations (complete)

	# TODO: An enumeration with a 2xxx error code for each client-side error.

	sub _assign {
		no strict 'refs';
		my $pkg = caller() . '::';
		my $key = shift;
		my $value = shift;
		*{"$pkg$key"} = sub () { $value };
	}

	sub _make_constant {
		my $what = shift;
		my $prefix = shift;
		my $i = 0;
		while ((my $key, my $value) = each(%$what)) {
			_assign("$prefix$value", $key);
		}
	}

	sub _find_largest_uint {
		# Assumes the Perl compiler uses signed integers.
		my $cur = 0;
		my $next = 0;
		for (my $i = 0; $i <= 16; $i++) {
			$next += (0xff << ($i * 8));
			return $i if sprintf("%d", $next) < 0;
			$cur = $next;
		}
	}

	sub _detect_alarm_bug {
		# Try for 2sec to activate a 1sec (potentially less)
		# alarm(), while cycling through select().  During
		# package construction, this takes 0.1s if bug-free
		# and ~2s otherwise.
		my $alarm_bug = 1;
		local $SIG{ALRM} = sub { $alarm_bug = 0; };
		alarm 1;
		for (my $i = 0; $i < 20; $i++) {
			last if $alarm_bug == 0;
			select(undef, undef, undef, 0.1);
		}
		local $SIG{ALRM} = sub { };
		sleep 0;
		return $alarm_bug;
	}

	sub _flags_included {
		my $flags = shift;
		my $i = 0;
		while ((my $key, my $value) = each(%$flags)) {
			$i |= $key;
		}
		return $i;
	}

	_assign("MAX_UINT_SIZE", _find_largest_uint);
	_assign("USEFUL_COLUMN_FLAGS", _flags_included(\%COLUMN_FLAGS));

	# Pick your poison..  Detecting whether the current platform has
	# this bug takes 1-2 seconds if the bug is present; always
	# activating the workaround introduces an (as yet) unmeasured
	# delay on platforms where the bug is not present; activating
	# the workaround based on a platform identifier does not adapt
	# when a new platform appears or an old platform gets fixed.
	#
	#_assign("FIX_ALARM_BUG", _detect_alarm_bug);
	#_assign("FIX_ALARM_BUG", 1);
	_assign("FIX_ALARM_BUG", $^O =~ /^MSWin/ ? 1 : 0);

	_make_constant(\%MACKET_NAMES, "MACKET_");
	_make_constant(\%COMMAND_NAMES, "COMMAND_");
	_make_constant(\%FLAG_NAMES, "FLAG_");
	_make_constant(\%UTF8_COLLATIONS, "UTF8_");
	_make_constant(\%DATA_TYPES, "DATA_");
	_make_constant(\%SERVER_STATUS, "STATUS_");
	_make_constant(\%COLUMN_TYPES, "TYPE_");
	_make_constant(\%COLUMN_FLAGS, "COLUMN_");
}

my %MACKET_NAMES = %{*MACKET_NAMES};
my %COMMAND_NAMES = %{*COMMAND_NAMES};
my %FLAG_NAMES = %{*FLAG_NAMES};
my %UTF8_COLLATIONS = %{*UTF8_COLLATIONS};
my %DATA_TYPES = %{*DATA_TYPES};
my %SERVER_STATUS = %{*SERVER_STATUS};
my %COLUMN_TYPES = %{*COLUMN_TYPES};
my %COLUMN_FLAGS = %{*COLUMN_FLAGS};

# Constructor
sub new {
	my $class = shift;
	my %args = @_;

	my $self = bless {
		host             => $args{host},
		port             => $args{port} || DEFAULT_PORT_NUMBER,
		database         => $args{database},
		user             => $args{user},
		password         => $args{password},
		connect_timeout  => defined($args{connect_timeout}) ? $args{connect_timeout} : DEFAULT_CONNECT_TIMEOUT,
		query_timeout    => defined($args{query_timeout}) ? $args{query_timeout} : DEFAULT_QUERY_TIMEOUT,
		flags            => $args{flags} || 0,
		debug            => $args{debug} || 0,
	}, $class;
	$self->_reset_connection_state;
	$self->_reset_command_state;
	return $self;
}

# Initializes the connection to the server.
# An error is raised by an inner class if already connected
sub connect {
	my $self = shift;
	$self->_connect;
	$self->_perform_handshake;
	$self->_perform_authentication;
}

# Sends an SQL query
sub query {
	my $self = shift;
	my $sql = shift;

	return $self->_execute_query($sql, 0);
}

# Sends an SQL query, but does not read any rows in the result set
sub stream {
	my $self = shift;
	my $sql = shift;

	return $self->_execute_query($sql, 1);
}

# Creates a prepared statement object.
sub prepare {
	my $self = shift;
	my $sql = shift;

	return Net::Wire10::PreparedStatement->new($self, $sql);
}

# Sends a wire protocol ping
sub ping {
	my $self = shift;

	$self->_check_streaming;
	$self->_check_connected;
	$self->_reset_command_state;
	$self->_reset_timeout($self->{connect_timeout});

	return $self->_execute_command(COMMAND_PING, '', undef);
}

# Close the database connection
sub disconnect {
	my $self = shift;
	my $socket = $self->{socket};
	my $select = $self->{io_select};

	eval {
		if ($socket) {
			if ($select) {
				if ($select->can_write(TIMEOUT_GRANULARITY)) {
					my $body = COMMAND_QUIT;
					$self->_send_mackets($body, 0, MACKET_COMMAND);
				}
			}
			$socket->close;
		}
	};
	warn $@ if $@;

	$self->_reset_command_state;
	$self->_reset_connection_state;

	return undef;
}

# Cancels a running query
sub cancel {
	my $self = shift;
	$self->{cancelling} = 1;
	return undef;
}

# Get the connection id
sub get_connection_id {
	my $self = shift;
	return $self->{server_thread_id};
}

# Get the server version string
sub get_server_version {
	my $self = shift;
	return $self->{server_version};
}

# Is the driver currently connected?
# If a fatal error has occurred, this will return false
sub is_connected {
	my $self = shift;
	return defined($self->{socket});
}

# Return the current error object, if any
sub get_error_info {
	my $self = shift;
	return $self->{error};
}

# Reset the time remaining counter before executing a command
sub _reset_timeout {
	my $self = shift;
	my $seconds = shift;
	if ($seconds == 0) {
		$self->{command_expire_time} = 0;
		return undef;
	}
	$self->{command_expire_time} = time + $seconds;
	return undef;
}

# Return the number of seconds left before the current
# operation should time out
sub _check_time_remaining {
	my $self = shift;
	return 0 if $self->{command_expire_time} == 0;
	my $remaining = $self->{command_expire_time} - time;
	$self->_fatal_error("Timeout while receiving data") if $remaining < 1;
	return $remaining;
}

# Fail if not connected anymore, due for example to a fatal error
sub _check_connected {
	my $self = shift;
	$self->_fatal_error("Not connected") unless defined($self->{socket});
}

# Fail if currently connected to a streaming data reader
sub _check_streaming {
	my $self = shift;
	$self->_vanilla_error("Connection is busy streaming") if $self->{streaming};
}

# Connects to the database server
sub _connect {
	my $self = shift;

	$self->_vanilla_error("Already connected") if defined($self->{socket});
	$self->_fatal_error("No host given") if length($self->{host}) == 0;
	$self->_fatal_error("No port given") if length($self->{port}) == 0;

	# Connect timeout.
	$self->_reset_timeout($self->{connect_timeout});

	my $socket;
	printf "Connecting to: %s:%d/tcp\n", $self->{host}, $self->{port} if $self->{debug} & 1;
	$socket = IO::Socket::INET->new(
		Proto    => 'tcp',
		PeerAddr => $self->{host},
		PeerPort => $self->{port},
		Timeout  => $self->_check_time_remaining
	) or $self->_fatal_error("Couldn't connect to $self->{host}:$self->{port}/tcp: $@");
	$socket->autoflush(1);
	$socket->timeout(TIMEOUT_GRANULARITY);
	$self->{socket} = $socket;
	$self->{io_select} = new IO::Select($self->{socket});

	$self->_reset_command_state;
}

# When a fatal error occurs, tear down TCP
# connection and set command state to indicate error.
sub _fatal_error {
	my $self = shift;
	my $msg = shift || '';

	$self->disconnect;

	$self->{error} = Net::Wire10::Error->new(-1, '', $msg) unless defined ($self->{error});
	$self->{error}->{message} = $msg if length($msg) > 0;

	die $self->{error}->{message};
}

# When a non-fatal error occurs, just throw it.
sub _vanilla_error {
	my $self = shift;
	my $msg = shift || '';

	$self->{error} = Net::Wire10::Error->new(-1, '', $msg) unless defined ($self->{error});
	$self->{error}->{message} = $msg if length($msg) > 0;

	die $self->{error}->{message};
}

# Receives data from the network and reassembles fragmented packets
sub _receive_packet_data {
	my $self = shift;
	my $socket = $self->{socket};
	my $io_select = $self->{io_select};
	my $data;

	while ($self->_check_time_remaining) {
		# Cancel if requested.
		$self->_fatal_error("Query cancelled") if $self->{cancelling};
		# Ask every second if there is data to be read.
		my $ready = $io_select->can_read(TIMEOUT_GRANULARITY);
		# IO::Select sometimes returns undef instead of an empty array.
		last if defined($ready) && (scalar($ready) != 0);
		# Work around a bug in alarm().
		sleep 0 if FIX_ALARM_BUG;
	}

	$socket->recv($data, STREAM_BUFFER_LENGTH, 0);
	# If select said data is available, but it
	# was not, it means the connection was lost.
	$self->_fatal_error("Lost connection") if length($data) == 0;
	$self->_dump_packet($data);

	$self->_alloc_packet_buffer(\$data);
	$self->_add_to_packet_buffer(\$data);
}

# If we have an idea of how much data is arriving, or there is not
# enough room for the recently received chunk, allocate more memory
sub _alloc_packet_buffer {
	my $self = shift;
	my $data = shift;

	my $buflen = length($self->{packet_buffer});
	my $goal = defined($self->{packet_goal}) ? $self->{packet_goal} + MACKET_HEADER_LENGTH : 0;
	my $goal2 = length($$data) + $self->{packet_read};
	$goal = $goal2 if $goal2 > $goal;
	if ($buflen < $goal ) {
		$self->{packet_buffer} .= ' ' x ($goal - $buflen);
	}
}

# Add received data to packet buffer
sub _add_to_packet_buffer {
	my $self = shift;
	my $data = shift;
	substr $self->{packet_buffer}, $self->{packet_read}, length($$data), $$data;
	$self->{packet_read} += length($$data);
}

# Queue the whole mackets received
sub _queue_mackets {
	my $self = shift;
	my $packet = \$self->{packet_buffer};

	# Receive more data if the macket header cannot be read
	return undef if $self->{packet_read} < MACKET_HEADER_LENGTH;

	# As long as we can read a macket header
	while ($self->{packet_read} > MACKET_HEADER_LENGTH) {
		# Get the length of the next macket, plus header length.
		if (! defined($self->{packet_goal})) {
			my $pos = 0;
			# Cached to avoid repeated decoding.
			$self->{packet_goal} = Net::Wire10::Util::decode_my_uint($$packet, \$pos, 3) + MACKET_HEADER_LENGTH;
		}
		return undef if $self->{packet_read} < $self->{packet_goal};
		print "Shifting a new macket totalling " . $self->{packet_goal} . " byte(s) into the macket queue.\n" if $self->{debug} & 1;
		my $macket = { buf => substr($$packet, 0, $self->{packet_goal}, '') };
		unshift(@{$self->{macket_queue}}, $macket);
		$self->{packet_read} -= $self->{packet_goal};
		$self->{packet_goal} = undef;
	}
}

# Gets the next macket in the queue
sub _next_macket {
	my $self = shift;
	my $macket;

	# Receive data if necessary
	while (scalar(@{$self->{macket_queue}}) == 0) {
		$self->_receive_packet_data;
		print "Bytes in receive buffer: " . $self->{packet_read} . "\n" if $self->{debug} & 1;
		$self->_queue_mackets;
	}

	# Return first queued macket
	$macket = pop(@{$self->{macket_queue}});
	$self->_check_received_macket_type($macket);
	return $macket;
}

# Fail if driver is getting something completely
# different from what it expected
sub _check_received_macket_type {
	my $self = shift;
	my $macket = shift;
	my $expected = $self->{expected_macket};
	my $msg = "";
	my $type = $self->_derive_received_macket_type($macket);

	$self->_dump_macket($macket);

	# If this is a error macket set error information and return,
	# regardless of what caller expects (error mackets are always expected).
	if ($type == MACKET_ERROR) {
		return undef;
	}

	if ($expected & MACKET_EOF) {
		return undef if ($type == MACKET_EOF);
		$msg .= "Expected EOF Macket, did not receive it!\n";
	}

	if ($expected & MACKET_HANDSHAKE) {
		return undef if ($type == MACKET_HANDSHAKE);
		$msg .= "Expected Handshake Initialization Macket, did not recieve it!\n";
	}

	if ($expected & MACKET_OK) {
		return undef if ($type == MACKET_OK);
		$msg .= "Expected OK Macket, did not receive it!\n";
	}

	if ($expected & MACKET_RESULT_SET_HEADER) {
		return undef if ($type == MACKET_RESULT_SET_HEADER);
		$msg .= "Expected Result Set Header Macket, dit not receive it!\n";
	}

	if ($expected & MACKET_COLUMN_INFO) {
		return undef if ($type == MACKET_COLUMN_INFO);
		$msg .= "Expected Column Info Macket, did not receive it!\n";
	}

	if ($expected & MACKET_ROW_DATA) {
		return undef if ($type == MACKET_ROW_DATA);
		$msg .= "Expected Row Data Macket, did not receive it!\n";
	}

	if ($expected & MACKET_MORE_DATA) {
		return undef if ($type == MACKET_MORE_DATA);
		$msg .= "Expected Fragmented Data Macket, did not receive it!\n";
	}

	$self->_fatal_error($msg);
}

# Lacking a proper indicator of message type in received mackets,
# try to figure out what type of macket it is based on circumstances
sub _derive_received_macket_type {
	my $self = shift;
	my $macket = shift;

	# If this is a error macket set error information and return
	if (ord(substr($macket->{buf}, MACKET_HEADER_LENGTH, 1)) == 0xff) {
		return $macket->{type} = MACKET_ERROR;
	}

	if ((length($macket->{buf}) < MACKET_HEADER_LENGTH + 9) && (ord(substr($macket->{buf}, MACKET_HEADER_LENGTH, 1)) == 0xFE)) {
		return $macket->{type} = MACKET_EOF;
	}

	if ((ord(substr($macket->{buf}, MACKET_HEADER_LENGTH, 1)) == 0) && (length($macket->{buf}) > MACKET_HEADER_LENGTH + 5)) {
		return $macket->{type} = MACKET_OK;
	}

	if ($self->_extract_macket_number($macket) == 0 && ! defined($self->{protocol_version})) {
		return $macket->{type} = MACKET_HANDSHAKE;
	}

	if (! defined($self->{no_of_columns})) {
		return $macket->{type} = MACKET_RESULT_SET_HEADER;
	}

	if ($self->{expected_macket} & MACKET_COLUMN_INFO) {
		return $macket->{type} = MACKET_COLUMN_INFO;
	}

	if ($self->{expected_macket} & MACKET_ROW_DATA) {
		return $macket->{type} = MACKET_ROW_DATA;
	}

	if ($self->{expected_macket} & MACKET_MORE_DATA) {
		return $macket->{type} = MACKET_MORE_DATA;
	}

	$self->_fatal_error("Unknown macket type received");
}

# Sends a message to the server
sub _send_mackets {
	my $self = shift;
	my $body = shift;
	my $nr = shift;
	my $type = shift;
	my $len = length($body);
	my $pos = 0;

	while ($len >= 0) {
		# The server terminates the connection if the fragmented mackets
		# (except the last, of course) are not 0xffffff bytes long.
		my $chunk_len = $len > 0xffffff ? 0xffffff : $len;
		$len = -1 if ($len == 0) or ($chunk_len < 0xffffff);
		$len -= $chunk_len;

		my $head_len = Net::Wire10::Util::encode_my_uint($chunk_len, 3);
		my $head_nr = Net::Wire10::Util::encode_my_uint($nr, 1);
		my $chunk = substr($body, $pos, $chunk_len);
		$type = MACKET_MORE_DATA if ($pos > 0);
		my $macket = { buf => $head_len . $head_nr . $chunk, type => $type };

		$self->_dump_macket($macket);

		my $socket = $self->{socket};
		die "Not connected" unless $socket;

		$socket->send($macket->{buf}, 0);
		$nr = ($nr + 1) % 256;
		$pos += $chunk_len;
	}
}

# Reads and interprets server handshake
sub _perform_handshake {
	my $self = shift;

	$self->{expected_macket} = MACKET_HANDSHAKE;

	my $macket = $self->_next_macket;

	# Server will send an error instead of greeting
	# if too many connection slots are filled.
	if ($macket->{type} == MACKET_ERROR) {
		$self->_parse_error_macket($macket, "Connect failed: ");
		$self->_fatal_error;
	}

	# Skip the macket header as it has been processed by _next_macket().
	my $i = MACKET_HEADER_LENGTH;
	printf "\n%s():\n", (caller(1))[3] if $self->{debug} & 1;
	# Protocol version
	$self->{protocol_version} = ord substr $macket->{buf}, $i, 1;
	printf "  -> Protocol Version: %d\n", $self->{protocol_version} if $self->{debug} & 1;

	# Quit if the protocol version does not match what the driver supports.
	if ($self->{protocol_version} != 10) {
		$self->_fatal_error("Only MySQL wire protocol 10 is supported at the moment\n");
	}

	$i += 1;
	# Server version
	my $string_end = index($macket->{buf}, "\0", $i);
	$self->_fatal_error("Could not decode server version.\n") if $string_end == -1;
	$string_end -= $i;
	$self->{server_version} = substr $macket->{buf}, $i, $string_end;
	printf "  -> Server version: %s\n", $self->{server_version} if $self->{debug} & 1;
	$i += $string_end + 1;
	my $left = length($macket->{buf}) - $i;
	$self->_fatal_error("Server handshake message truncated.\n") if $left < 44;
	# Server thread id
	$self->{server_thread_id} = Net::Wire10::Util::decode_my_uint($macket->{buf}, \$i, 4);
	printf "  -> Server thread id: %d\n", $self->{server_thread_id} if $self->{debug} & 1;
	# Scramble buff, 1st part
	$self->{salt} = substr $macket->{buf}, $i, 8;
	# Enables the use of old passwords
	$self->{salt_old} = $self->{salt};
	# The salt part is 8 bytes long and there is a one byte filler after
	$i += 8 + 1;
	# Server_capabilities is not used at the moment
	my $server_caps = Net::Wire10::Util::decode_my_uint($macket->{buf}, \$i, 2);
	printf "  -> Server capabilities (ignored): 0x%x\n", $server_caps if $self->{debug} & 1;
	# Server_language is not used at the moment
	my $server_lang = Net::Wire10::Util::decode_my_uint($macket->{buf}, \$i, 1);
	printf "  -> Server language (ignored): %d\n", $server_lang if $self->{debug} & 1;
	# Server_status is not used at the moment
	my $server_status = Net::Wire10::Util::decode_my_uint($macket->{buf}, \$i, 2);
	printf "  -> Server status (ignored): 0x%x\n", $server_status if $self->{debug} & 1;
	# 13 byte filler
	$i += 13;
	# Scramble buff, 2nd part
	# The MySQL protocol documentation says that this part is 13 bytes long,
	# but the last byte seems to be a filler.  So we only read 12 bytes.
	$self->{salt} .= substr $macket->{buf}, $i, 12;
	printf "  -> Salt: %s\n", $self->{salt} if $self->{debug} & 1;
	# Filler
	$i += 1;
}

# Sends authentication, waits for answer, resends
# response to challenge in old format if requested by server
sub _perform_authentication {
	my $self = shift;

	$self->_send_login_message;
	$self->{expected_macket} = MACKET_EOF + MACKET_OK;

	my $auth_result = $self->_next_macket;
	if ($auth_result->{type} == MACKET_EOF) {
		# Note: this EOF packet is truncated by 4 bytes compared
		#       to normal EOF packets, parsing it is skipped.
		$self->_send_old_password;
		$self->{expected_macket} = MACKET_OK;
		$auth_result = $self->_next_macket;
	}
	if ($auth_result->{type} == MACKET_OK) {
		$self->_parse_ok_macket($auth_result);
	}
	if ($auth_result->{type} == MACKET_ERROR) {
		$self->_parse_error_macket($auth_result, "Authentication failed: ");
		$self->_fatal_error;
	}
	print "Connected to database server\n" if $self->{debug} & 1;
}

# Sends new format client authentication
sub _send_login_message {
	my $self = shift;

	# Obligatory flags
	my $driver_flags =
		FLAG_LONG_PASSWORD +
		FLAG_LONG_FLAG +
		FLAG_CONNECT_WITH_DB +
		# Unsupported:  COMPRESS
		# Unsupported:  LOCAL_FILES
		FLAG_PROTOCOL_41 +
		# Unsupported:  SSL
		# Unsupported:  IGNORE_SIGPIPE
		FLAG_TRANSACTIONS +
		# Unsupported:  RESERVED
		FLAG_SECURE_CONNECTION +
		# Unsupported:  MULTI_STATEMENTS
		# Necessary to avoid server abort:  FLAG_MULTI_RESULTS
		FLAG_MULTI_RESULTS;
		# Unsupported:  SSL_VERIFY_SERVER_CERT
		# Unsupported:  REMEMBER_OPTIONS

	# Optional flags
	my $customizable_flags =
		FLAG_FOUND_ROWS +
		FLAG_NO_SCHEMA +
		FLAG_ODBC +
		FLAG_IGNORE_SPACE +
		FLAG_INTERACTIVE;

	my $flags = $driver_flags | ($customizable_flags & $self->{flags});
	my $body .= Net::Wire10::Util::encode_my_uint($flags, 4);
	# Max macket size.  Completely disregarded by the server,
	# which just overflows macket data onto other mackets when
	# sending and accepts overflowed data when receiving.
	# See also note in _send_mackets().
	$body .= Net::Wire10::Util::encode_my_uint(0x01000000, 4);
	# Character set; hardcoded to UTF-8, used in _parse_column_info_macket().
	$body .= chr(UTF8_GENERAL_CI);
	# 23 bytes filler.
	$body .= "\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0";
	# Null-terminated: user name
	$body .= $self->{user} . "\0";
	if (length($self->{password}) > 0) {
		$body .= "\x14" . Net::Wire10::Password->scramble($self->{password}, $self->{salt});
	} else {
		$body .= "\0";
	}
	if (defined($self->{database})) {
		# Documentation says there should be a filler here,
		# but other clients don't send that.
		#$body .= "\0";
		# Null-terminated: initial default database name.
		$body .= $self->{database};
		$body .= "\0";
	}
	$self->_send_mackets($body, 1, MACKET_AUTHENTICATE);
}

# Sends old format client authentication
sub _send_old_password {
	my $self = shift;
	my $body = Net::Wire10::Password32->scramble(
		$self->{password}, $self->{salt_old}, 1
	) . "\0";
	$self->_send_mackets($body, 3, MACKET_AUTHENTICATE);
}

# Execute a SQL command
sub _execute_query {
	my $self = shift;
	my $sql = shift;
	my $wantstream = shift;

	$self->_check_streaming;
	$self->_check_connected;
	$self->_reset_command_state;
	$self->_reset_timeout($self->{query_timeout});
	$self->{streaming} = 1 if $wantstream;

	my $iterator = Net::Wire10::Results->new($self);
	$self->{streaming_iterator} = $iterator if $wantstream;

	# The protocol is configured to always use UTF-8 during handskake, to
	# avoid messing with all the local character sets.  Therefore any input
	# string needs to be automatically converted if it is not already UTF-8.
	utf8::upgrade($sql) if defined($sql) and not utf8::is_utf8($sql);

	printf "Executing query: %s%s\n", substr($sql, 0, 100), length($sql) >= 100 ? " ..." : "" if $self->{debug} & 1;

	$self->_execute_command(COMMAND_QUERY, $sql, $iterator);
	return $iterator;
}

# Send a protocol command message
sub _execute_command {
	my $self = shift;
	my $command = shift;
	my $param = shift;
	my $iterator = shift;

	# Abort early if the driver is no longer connected.
	$self->_check_connected;

	# Strip the utf8 flag from the string,
	# otherwise the socket send() complains.
	Encode::_utf8_off($param);

	# Send the SQL command
	my $body = $command . $param;
	$self->_send_mackets($body, 0, MACKET_COMMAND);
	$self->{expected_macket} = MACKET_OK | MACKET_RESULT_SET_HEADER;

	# Receive the result from the database
	my $macket = $self->_next_macket;

	if ($macket->{type} == MACKET_ERROR) {
		$self->_detach_results($iterator);
		$self->_parse_error_macket($macket);
		$self->_vanilla_error;
	}
	if ($macket->{type} == MACKET_RESULT_SET_HEADER) {
		my $pos = MACKET_HEADER_LENGTH;
		$self->_parse_result_set_header_macket($macket, \$pos);
		$self->_retrieve_column_info($iterator);
		$self->_retrieve_results($iterator) unless $self->{streaming};
	}
	if ($macket->{type} == MACKET_OK) {
		$self->_parse_ok_macket($macket, $iterator);
		$self->_detach_results($iterator);
	}
}

# Reads and interprets result set header
sub _parse_result_set_header_macket {
	my $self = shift;
	my $macket = shift;
	my $pos = shift;

	$self->{no_of_columns} = $self->_decode_lcb_or_fail($macket->{buf}, $pos);
	printf "Number of columns: %d\n", $self->{no_of_columns} if $self->{debug} & 1;
	# Optionally the "extra" field is in the macket
	my $macket_length = length($macket->{buf}) - MACKET_HEADER_LENGTH;
	if ($macket_length - 1 > $pos) {
		my $extra = $self->_decode_lcb_or_fail($macket->{buf}, $pos);
		printf "Extra information (ignored): %d\n", $extra if $self->{debug} & 1;
	}
}

# Reads and stores error message, sql state and error code
sub _parse_error_macket {
	my $self = shift;
	my $macket = shift;
	my $extra = shift || '';

	if ($macket->{type} != MACKET_ERROR) {
		$self->_fatal_error("Expected error macket");
	}

	# skip macket header
	my $pos = MACKET_HEADER_LENGTH;
	# skip macket type
	$pos += 1;
	my $left = length($macket->{buf}) - $pos;
	$self->_fatal_error("Truncated error macket") if $left < 2;
	# error code
	my $code = Net::Wire10::Util::decode_my_uint($macket->{buf}, \$pos, 2);
	# Documentation says there always is a SQLSTATE marker here,
	# but that is not true.
	my $sqlstate_marker = substr($macket->{buf}, $pos, 1);
	my $sqlstate = '';
	if ($sqlstate_marker eq '#') {
		# skip SQL state marker
		$pos += 1;
		# read SQL state
		my $sqlstate = substr($macket->{buf}, $pos, 5);
		$pos += 5;
	}
	# message
	my $message = substr($macket->{buf}, $pos);
	Encode::_utf8_on($message);

	# create error info object
	$self->{error} = Net::Wire10::Error->new($code, $sqlstate, $extra . $message);
}

# Reads column info and EOF, possibly
# interrupted at any point by an error from the server
sub _retrieve_column_info {
	my $self = shift;
	my @queue = ();
	my $macket;
	my $iterator = shift;

	$self->{expected_macket} = MACKET_COLUMN_INFO | MACKET_EOF;
	# Put all field mackets in the selected queue
	do {
		$macket = $self->_next_macket;
		if ($macket->{type} == MACKET_ERROR) {
			$self->_parse_error_macket($macket, "Server reported error while reading column info: ");
			$self->_fatal_error;
		}
		if ($macket->{type} == MACKET_COLUMN_INFO) {
			my $column_info = $self->_parse_column_info_macket($macket);
			push(@queue, $column_info);
		}
	} until ($macket->{type} == MACKET_EOF);
	if ($self->{no_of_columns} ne scalar @queue) {
		$self->_fatal_error(sprintf(
			"Server reported %d columns, but sent %d",
			$self->{no_of_columns},
			scalar @queue
		));
	}
	$iterator->{column_info} = \@queue;
	$self->_parse_eof_macket($macket, $iterator);
}

# Reads all rows in result set
sub _retrieve_results {
	my $self = shift;
	my $iterator = shift;

	# Put all row data in the selected queue
	while ($self->_retrieve_row_data($iterator)) {}
}

# Reads row data or EOF, possibly
# interrupted at any point by an error from the server
sub _retrieve_row_data {
	my $self = shift;
	my $iterator = shift;

	$self->{expected_macket} = MACKET_ROW_DATA + MACKET_EOF;
	my $macket = $self->_next_macket;

	if ($macket->{type} == MACKET_ERROR) {
		$self->_parse_error_macket($macket, "Server reported error while reading row data: ");
		$self->_fatal_error;
	}
	if ($macket->{type} == MACKET_ROW_DATA) {
		# Note: The manual does not specify how the server fragments data.
		# One possibility would be that if the server sends data that fits
		# exactly on a macket boundary, then sends an additional completely
		# empty macket, to allow the client to assume that a maxed out macket
		# means that more fragments will follow.  Another possibility would
		# be that the server expects the client to deduce whether more mackets
		# are coming based on the contents of each macket, since fragmented
		# mackets only occur for data mackets which have an additional length
		# indicator (for the field data) inside the macket.  For the usual
		# lack of documentation, the method used below is a guess.
		my $nasty = length($macket->{buf});
		while ($nasty == MACKET_HEADER_LENGTH + 0xffffff) {
			# Glue together to form proper macket
			# (with invalid macket_length).
			printf "Fragmented macket found, retrieving one more fragment macket.\n" if $self->{debug} & 1;
			$self->{expected_macket} = MACKET_MORE_DATA;
			my $next_macket = $self->_next_macket;
			if ($macket->{type} == MACKET_ERROR) {
				$self->_parse_error_macket($macket, "Server reported error while reading more row data: ");
				$self->_fatal_error;
			}
			$nasty = length($next_macket->{buf});
			# Remove header from next fragment.
			substr($next_macket->{buf}, 0, MACKET_HEADER_LENGTH, "");
			# Concatenate macket contents.
			$macket->{buf} .= $next_macket->{buf};
		}
		# Remove header from first fragment.
		substr($macket->{buf}, 0, MACKET_HEADER_LENGTH, "");
		printf "Unshifting %d byte(s) of row data onto queue.\n", length($macket->{buf}) if $self->{debug} & 1;
		push(@{$iterator->{row_data}}, $macket->{buf});
		# More data may be available.
		return 1;
	}
	if ($macket->{type} == MACKET_EOF) {
		# Read EOF macket
		$self->_parse_eof_macket($macket, $iterator);
		$self->_detach_results($iterator);
		# No more data available.
		return 0;
	}
}

# Disconnect result set from driver, must only be done after all results have been read
sub _detach_results {
	my $self = shift;
	my $iterator = shift;
	$iterator->{wire}->{streaming_iterator} = undef;
	$iterator->{wire}->{streaming} = 0;
	$iterator->{wire} = undef;
}

# Reads and interprets OK, saving the number of affected rows,
# the insert id, and the server message.  Returns the number of
# affected rows
sub _parse_ok_macket {
	my $self = shift;
	my $macket = shift;
	my $iterator = shift;

	# Because affected rows is a Length Coded Binary
	# we use pointers to the position in the macket.
	# The position is updated in each method called
	# to ensure that the macket is read correctly.
	# First, skip macket header and macket type.
	my $pos = MACKET_HEADER_LENGTH + 1;
	printf "\n%s():\n", (caller(1))[3] if $self->{debug} & 1;
	# Affected rows
	my $affected = $self->_decode_lcb_or_fail($macket->{buf}, \$pos);
	printf "  -> Affected rows: %d\n", $affected if $self->{debug} & 1;
	# Insert id
	my $raw_id = $self->_skip_lcb_or_fail($macket->{buf}, \$pos);
	printf "  -> Insert id (decode postponed): %d byte(s)\n", length($raw_id) if $self->{debug} & 1;
	my $left = length($macket->{buf}) - $pos;
	$self->_fatal_error("Truncated OK macket") if $left < 4;
	# Server status
	my $status = Net::Wire10::Util::decode_my_uint($macket->{buf}, \$pos, 2);
	printf "  -> Server status flags (ignored): 0x%x\n", $status if $self->{debug} & 1;
	# Warning count
	my $warnings = Net::Wire10::Util::decode_my_uint($macket->{buf}, \$pos, 2);
	printf "  -> Warning count: %d\n", $warnings if $self->{debug} & 1;
	# Message
	my $message = substr($macket->{buf}, $pos);
	Encode::_utf8_on($message);
	printf "  -> Server message (ignored): %s\n", $message if $self->{debug} & 1;

	if (defined($iterator)) {
		$iterator->{no_of_affected_rows} = $affected;
		$iterator->{raw_insert_id} = $raw_id;
		$iterator->{warnings} = $warnings;
	}
}

# Reads and interprets EOF, saving relevant values.
sub _parse_eof_macket {
	my $self = shift;
	my $macket = shift;
	my $iterator = shift;

	my $pos = MACKET_HEADER_LENGTH + 1;
	my $left = length($macket->{buf}) - $pos;
	$self->_fatal_error("Truncated EOF macket") if $left < 4;
	printf "\n%s():\n", (caller(1))[3] if $self->{debug} & 1;
	# Warning count
	my $warnings = Net::Wire10::Util::decode_my_uint($macket->{buf}, \$pos, 2);
	printf "  -> Warning count: %d\n", $warnings if $self->{debug} & 1;
	# Server status
	my $status = Net::Wire10::Util::decode_my_uint($macket->{buf}, \$pos, 2);
	printf "  -> Server status flags: 0x%x\n", $status if $self->{debug} & 1;
	$self->_fatal_error("Query yielded multiple results (not supported)") if $status & STATUS_MORE_RESULTS_EXISTS;
	$iterator->{warnings} = $warnings if defined($iterator);
}

# Reads and interprets column information
sub _parse_column_info_macket {
	# Metadata always in UTF-8; see collation in _send_login_message().
	my $self = shift;
	my $macket = shift;

	# Skip the header.
	my $pos = MACKET_HEADER_LENGTH;
	# Catalog
	my $catalog = $self->_decode_string_or_fail($macket->{buf}, \$pos);
	Encode::_utf8_on($catalog);
	# DB
	my $db = $self->_decode_string_or_fail($macket->{buf}, \$pos);
	Encode::_utf8_on($db);
	# Table
	my $table = $self->_decode_string_or_fail($macket->{buf}, \$pos);
	Encode::_utf8_on($table);
	# Org_table
	my $orig_table = $self->_decode_string_or_fail($macket->{buf}, \$pos);
	Encode::_utf8_on($orig_table);
	# Name
	my $column = $self->_decode_string_or_fail($macket->{buf}, \$pos);
	Encode::_utf8_on($column);
	# Org_name
	my $orig_column = $self->_decode_string_or_fail($macket->{buf}, \$pos);
	Encode::_utf8_on($orig_column);
	# Filler
	$pos += 1;
	my $left = length($macket->{buf}) - $pos;
	$self->_fatal_error("Truncated column info macket") if $left < 10;
	# Charset number
	my $collation = Net::Wire10::Util::decode_my_uint($macket->{buf}, \$pos, 2);
	# Skip reading display length, because some server functions, eg. ASBINARY(),
	# sets this to 0xFFFFFFFF unsigned, which is too big for 32-bit Perl.
	$pos += 4;
	#my $display_length = Net::Wire10::Util::decode_my_uint($macket->{buf}, \$pos, 4);
	# Column data type
	my $data_type = Net::Wire10::Util::decode_my_uint($macket->{buf}, \$pos, 1);
	# Flags
	my $flags = Net::Wire10::Util::decode_my_uint($macket->{buf}, \$pos, 2);
	my $sane_flags = $flags & USEFUL_COLUMN_FLAGS;
	# Decimal scale (for DECIMAL or NUMERIC data types)
	my $decimal_scale = Net::Wire10::Util::decode_my_uint($macket->{buf}, \$pos, 1);
	# Filler
	$pos += 2;
	printf "\n%s():\n", (caller(1))[3] if $self->{debug} & 1;
	# Optionally the default field is available in the macket
	my $macket_length = length($macket->{buf}) - MACKET_HEADER_LENGTH;
	if ($macket_length - 1 > $pos) {
		$self->{extra} = $self->_decode_lcb_or_fail($macket->{buf}, \$pos);
		print "  -> Default (ignored): " . $self->{extra} if $self->{debug} & 1;
	}
	my $format =
		"  -> Catalog (ignored):  %s\n  -> Name:               %s\n  -> Object:             %s\n".
		"  -> Source Column:      %s\n  -> Source Table:       %s\n  -> Source Database:    %s\n".
		"  -> Collation:          %s\n  -> Data Type:          %s\n  -> Decimal Scale:      %s\n".
		"  -> Display (ignored):  Undeciphered\n  -> Flags:              0x%x\n";
	printf $format,
		$catalog, $column, $table,
		$orig_column, $orig_table, $db,
		$collation, $data_type, $decimal_scale,
		$flags
	if $self->{debug} & 1;
	return {
		name => $column,
		object => $table,
		orig_database => $db,
		orig_table => $orig_table,
		orig_column => $orig_column,
		collation => $collation,
		data_type => $data_type,
		decimal_scale => $decimal_scale,
		flags => $sane_flags
	};
}

# Verify the integrity of a row data macket by checking for each
# field value that it is contained within the macket's total length
sub _field_value_exceeds_buffer {
	my $self = shift;
	my $buf = shift;
	my $pos = shift;
	my $len = length($buf);
	while ($pos < $len) {
		Net::Wire10::Util::skip_string($buf, \$pos);
	}
	return $pos - $len;
}

# Given a macket, extract serial number from its header
sub _extract_macket_number {
	my $self = shift;
	my $macket = shift;

	return ord(substr($macket->{buf}, 3, 1));
}

# Given a macket, extract length from its header
sub _extract_macket_length {
	my $self = shift;
	my $macket = shift;
	my $pos = 0;

	return Net::Wire10::Util::decode_my_uint($macket->{buf}, \$pos, 3);
}

# Wrap the skip_lcb library function in an error handler,
# catching errors and sending them to the driver's _fatal_error()
sub _skip_lcb_or_fail {
	my $self = shift;
	my $buf = shift;
	my $pos = shift;

	my $result = eval {
		return Net::Wire10::Util::skip_lcb($buf, $pos);
	};
	$self->_fatal_error($@) if $@;
	return $result;
}

# Wrap the decode_lcb library function in an error handler,
# catching errors and sending them to the driver's _fatal_error()
sub _decode_lcb_or_fail {
	my $self = shift;
	my $buf = shift;
	my $pos = shift;

	my $result = eval {
		return Net::Wire10::Util::decode_lcb($buf, $pos);
	};
	$self->_fatal_error($@) if $@;
	return $result;
}

# Wrap the decode_string library function in an error handler,
# catching errors and sending them to the driver's _fatal_error()
sub _decode_string_or_fail {
	my $self = shift;
	my $buf = shift;
	my $pos = shift;

	my $result = eval {
		return Net::Wire10::Util::decode_string($buf, $pos);
	};
	$self->_fatal_error($@) if $@;
	return $result;
}

# Resets the command execution status
sub _reset_command_state {
	my $self = shift;
	# Disconnect streaming iterator.
	$self->_detach_results($self->{streaming_iterator}) if defined($self->{streaming_iterator});
	# Reset internal column counter.
	$self->{no_of_columns} = undef;
	# Reset error state.
	$self->{error} = undef;
	# Reset cancel flag.
	$self->{cancelling} = 0;
}

# Reset entire connection
sub _reset_connection_state {
	my $self = shift;
	$self->{protocol_version} = undef;
	$self->{server_version} = undef;
	$self->{salt} = '';
	$self->{packet_buffer} = '';
	$self->{packet_goal} = undef;
	$self->{packet_read} = 0;
	$self->{socket} = undef;
	$self->{io_select} = undef;
	$self->{expected_macket} = undef;
	$self->{macket_queue} = [];
	$self->{command_expire_time} = undef;
}

# Dumps the packet to standard output, useful for debugging
sub _dump_packet {
	my $self = shift;
	return unless $self->{debug} & 4;
	my $packet = shift;
	my $str = sprintf "\n%s():\n", (caller(1))[3];
	my $len = length($packet);
	my $skipped = 0;
	my $pos = -16;
	while ($packet =~ /(.{1,16})/sg) {
		$pos += 16;
		unless ($self->{debug} & 8) {
			if (($len > 528) && ($pos > 256) && ($len - $pos > 256)) {
				if (! $skipped) {
					print "\n\n" . ' ' x 25 . "... snip more data ...\n";
					$skipped = 1;
				}
				next;
			}
		}
		my $line = $1;
		$str .= join ' ', map {sprintf '%02X', ord $_} split //, $line;
		$str .= '   ' x (16 - length $line);
		$str .= '  ';
		$str .= join '', map {
			sprintf '%s', (/[\w\d\*\,\?\%\=\'\;\(\)\.-]/) ? $_ : '.'
		} split //, $line;
		print $str;
		$str = "\n";
	}
	print $str;
}

# Dumps the macket to standard output, useful for debugging.
sub _dump_macket {
	my $self = shift;
	return unless $self->{debug} & 2;
	my $macket = shift;
	my $str = sprintf "\n%s():\n", (caller(1))[3];
	$str .= sprintf "  -> serial: %d\n", Net::Wire10::_extract_macket_number($self, $macket);
	$str .= sprintf "  -> length: %d\n", Net::Wire10::_extract_macket_length($self, $macket);
	my $type = $macket->{type};
	$str .= sprintf "  -> type:   %s\n", $Net::Wire10::MACKET_NAMES{$type};
	$str .= sprintf "  -> data:   ";
	my $len = length($macket->{buf});
	my $skipped = 0;
	my $pos = -16;
	while ($macket->{buf} =~ /(.{1,16})/sg) {
		$pos += 16;
		unless ($self->{debug} & 16) {
			if (($len > 528) && ($pos > 256) && ($len - $pos > 256)) {
				if (! $skipped) {
					print "\n\n" . ' ' x 25 . "... snip more data ...\n";
					$skipped = 1;
				}
				next;
			}
		}
		my $line = $1;
		$str .= ' ' x 13 if substr($str, -1, 1) eq "\n";
		$str .= join ' ', map {sprintf '%02X', ord $_} split //, $line;
		$str .= '   ' x (16 - length $line);
		$str .= '  ';
		$str .= join '', map {
			sprintf '%s', (/[\w\d\*\,\?\%\=\'\;\(\)\.-]/) ? $_ : '.'
		} split //, $line;
		print $str;
		$str = "\n";
	}
	print $str;
}



package Net::Wire10::Results;

use strict;
use warnings;
use Encode;

# Constructor
sub new {
	my $class = shift;
	my $wire = shift;
	return bless {
		wire => $wire,
		column_info => [],
		row_data => []
	}, $class;
}

# Gets next row as an array
sub next_array {
	my $self = shift;
	my @result;
	my $row;
	my $wire = $self->{wire};

	# Note: A die() from this context often brings an application down.
	#       For disconnected result sets, row data could be integrity
	#       checked during query() to simplify error handling for
	#       applications.

	if (defined($wire)) {
		# In streaming mode, fetch a row
		return undef unless $wire->_retrieve_row_data($self);
	}

	# Return unless there is another row available
	return undef if scalar(@{$self->{row_data}}) == 0;

	$row = shift(@{$self->{row_data}});
	my $pos = 0;
	for (my $i = 1; $i <= scalar(@{$self->{column_info}}); $i++) {
		my $fieldvalue = eval {
			return Net::Wire10::Util::decode_string($row, \$pos);
		};
		$self->_abort($@) if $@;
		my $collation = $self->{column_info}->[$i - 1]->{"collation"};
		$Net::Wire10::UTF8_COLLATIONS{$collation} and Encode::_utf8_on($fieldvalue);
		push @result, $fieldvalue;
	}

	return \@result;
}

# Leave driver in a consistent state and then die.
sub _abort {
	my $self = shift;
	my $msg = shift;
	eval {
		$self->spool;
	};
	die $msg;
}

# Gets next row as a hash
sub next_hash {
	my $self = shift;
	my $row = $self->next_array;
	return undef unless defined $row;
	my %result = map { $_->{name} => shift(@{$row}) } @{$self->{column_info}};
	return \%result;
}

# Retrieve and store remaining rows
sub spool {
	my $self = shift;
	my $wire = $self->{wire};
	return undef unless defined($wire);
	$wire->_retrieve_results($self);
	return undef;
}

# Retrieve and dispose of remaining rows
sub flush {
	my $self = shift;
	while ($self->next_array) {};
	return undef;
}

# Get names or other information on every column
sub get_column_info {
	my $self = shift;
	my $what = shift;
	my @info = @{$self->{column_info}};
	return @info unless defined($what);
	return map { $_->{$what} } @info;
}

# Did the query return a set of results or not
sub has_results {
	my $self = shift;
	return scalar(@{$self->{column_info}}) ? 1 : 0;
}

# Get the number of affected rows
sub get_no_of_affected_rows {
	my $self = shift;
	return $self->{no_of_affected_rows};
}

# Get the number of selected rows
sub get_no_of_selected_rows {
	my $self = shift;
	return 0 unless defined $self->{row_data};
	return scalar(@{$self->{row_data}});
}

# Get the insert id
sub get_insert_id {
	my $self = shift;
	my $pos = 0;
	my $insert_id = eval {
		return Net::Wire10::Util::decode_my_uint($self->{raw_insert_id}, \$pos, length($self->{raw_insert_id}));
	};
	$self->_abort($@) if $@;
	return $insert_id;
}

# Get the number of warnings
sub get_warning_count {
	my $self = shift;
	return $self->{warnings};
}



package Net::Wire10::PreparedStatement;

use strict;
use warnings;
use utf8;
use Encode;

# Constructor
sub new {
	my $class = shift;
	my $wire = shift;
	my $sql = shift;

	# Input is either iso8859-1 or Unicode, handle as Unicode internally.
	utf8::upgrade($sql) if defined($sql) and not utf8::is_utf8($sql);

	# Find all "?" placeholders in query.
	my @tokens = Net::Wire10::Util::tokenize($sql);

	return bless {
		wire => $wire,
		tokens => \@tokens,
		params => [],
	}, $class;
}

# Set a parameter at a given index to a given value.
sub set_parameter {
	my $self = shift;
	my $index = shift;
	my $value = shift;
	my $datatype = shift || Net::Wire10::DATA_TEXT;

	# If input is iso8859-1 or Unicode, handle as Unicode.
	utf8::upgrade($value) if defined $value and not (($datatype == Net::Wire10::DATA_BINARY) or utf8::is_utf8($value));

	# Store for later.  Placeholders are numbered beginning with 1.
	@{$self->{params}}[$index - 1] = $value;

	return undef;
}

# Clear a parameter at a given index, or all parameters if no index given.
sub clear_parameter {
	my $self = shift;
	my @indices = @_;

	@{$self->{params}} = [] unless scalar @indices > 0;
	foreach my $index (@indices) {
		@{$self->{params}}[$index - 1] = undef;
	}

	return undef;
}

# Return the number of ? tokens in the prepared statement.
sub get_marker_count {
	my $self = shift;
	return scalar grep(/^\?$/, @{$self->{tokens}});
}

# See note about split and whitespace in Net::Wire10::Util.
my $whitespace = qr/^[\ \r\n\013\t\f]$/;

# Probably degrades performance quite a bit, necessary due to server bug:
# http://bugs.mysql.com/bug.php?id=1337
my $workaround_bug_1337 = qr/^[0-9]+$/;

# Assemble and execute prepared statement.
sub _execute {
	my $self = shift;
	my $wantstream = shift;
	my $wire = $self->{wire};

	# Replace tokens with parameters.
	my $prepared = '';
	my $i = 0;
	my $has_charset = 0;
	foreach my $token (@{$self->{tokens}}) {
		# Look for a ? token.
		if ($token ne '?') {
			$prepared .= $token;
			# Look for character set associations for upcoming ? tokens.
			my $first = substr($token, 0, 1);
			$has_charset = 0 if $first !~ $whitespace;
			$has_charset = 1 if $first eq '_';
			# Fast-forward to next token.
			next;
		}

		# Fetch and quote parameter, or add NULL for undef.
		my $value = $self->{params}->[$i++];
		unless (defined($value)) {
			$prepared .= 'NULL';
			next;
		}
		$value = Net::Wire10::Util::quote($value) if $value !~ $workaround_bug_1337;

		# If input is binary, tell the database server that it's binary.
		# It usually doesn't matter much because binary data is usually
		# used in a binary context anyway, but it fixes various uncommon
		# scenarios such as doing a CONVERT on a binary string.
		$prepared .= '_binary' unless utf8::is_utf8($value) or $has_charset;

		# If input is binary, make sure that Perl doesn't try to translate it
		# from iso8859-1 (which it isn't) to Unicode when the string is joined
		# with actual Unicode text to form a MySQL query, which is a character
		# array of mixed text and binary data.
		Encode::_utf8_on($value) unless utf8::is_utf8($value);

		# Add parameter in place of token.
		$prepared .= $value;
	}

	print "Assembled statement: " . $prepared . "\n" if $wire->{debug} & 1;
	return $wire->stream($prepared) if $wantstream;
	return $wire->query($prepared);
}

# Run the prepared statement and spool results.
sub query {
	my $self = shift;
	return $self->_execute(0);
}

# Run the prepared statement and return a result object for streaming.
sub stream {
	my $self = shift;
	return $self->_execute(1);
}



package Net::Wire10::Error;

use strict;
use warnings;

# Constructor
sub new {
	my $class = shift;
	my $code = shift;
	my $state = shift;
	my $msg = shift;

	return bless {
		code => $code,
		state => $state,
		message => $msg,
	}, $class;
}

# Get the error code
sub get_error_code {
	my $self = shift;
	return $self->{code};
}

# Get the error SQL state designation
sub get_error_state {
	my $self = shift;
	return $self->{state} ? $self->{state} : '';
}

# Get the error message
sub get_error_message {
	my $self = shift;
	return $self->{message};
}



package Net::Wire10::Util;

use strict;
use warnings;

use constant LCB_NULL       => 251;
use constant LCB_UINT16     => 252;
use constant LCB_UINT24     => 253;
use constant LCB_UINT64     => 254;
use constant UINT8_LENGTH   => 1;
use constant UINT16_LENGTH  => 2;
use constant UINT24_LENGTH  => 3;
use constant UINT32_LENGTH  => 4;
use constant UINT64_LENGTH  => 8;
my @LCB_LENGTHS             = ();
$LCB_LENGTHS[LCB_NULL]      = 0;
$LCB_LENGTHS[LCB_UINT16]    = UINT16_LENGTH;
$LCB_LENGTHS[LCB_UINT24]    = UINT24_LENGTH;
$LCB_LENGTHS[LCB_UINT64]    = UINT64_LENGTH;

# Encode a given uint into wire protocol format, return as string of given length.
sub encode_my_uint {
	my $uint = shift;
	my $wirelen = shift;

	return substr(pack('V', $uint), 0, $wirelen);
}

# Decode a wire protocol unsigned integer at given offset and length and advance buffer pointer.
sub decode_my_uint {
	my $buf = shift;
	my $pos = shift;
	my $nrbytes = shift;

	my @bytes = unpack('x' . $$pos . 'C' x $nrbytes, $buf);
	my $result = 0;
	for (my $i = 0; $i < $nrbytes; $i++) {
		$result += $bytes[$i] << ($i * 8);
	}
	$$pos += $nrbytes;

	return $result if $nrbytes <= Net::Wire10::MAX_UINT_SIZE;

	die 'Overflow while reading a large integer' if
		($nrbytes > Net::Wire10::MAX_UINT_SIZE + 1) or
		($bytes[-1] > 0x7f);

	return $result;
}

# Find the length of a LCB-coded unsigned integer and advance buffer pointer.
sub decode_lcb_width {
	my $buf = shift;
	my $pos = shift;
	my $len = length($buf);
	my $msg = 'Reached end of input while decoding Length Coded Binary';

	die $msg if $$pos >= $len;
	my $head = ord substr(
		$buf,
		$$pos,
		UINT8_LENGTH
	);

	return UINT8_LENGTH if $head < LCB_NULL;
	$$pos += UINT8_LENGTH;
	return undef if $head == LCB_NULL;
	my $bytes = $LCB_LENGTHS[$head];
	die $msg if $$pos + $bytes >= $len;
	return $bytes if $head <= LCB_UINT64;
	# If we end up here, first byte equals 255, which is invalid.
	die 'Invalid First-Byte in Length Coded Binary: 255';
}

# Skip over a LCB-coded unsigned integer without decoding it.
sub skip_lcb {
	my $buf = shift;
	my $pos = shift;

	my $bytes = decode_lcb_width($buf, $pos);
	my $my_uint = substr($buf, $$pos, $bytes);
	$$pos += $bytes;
	return $my_uint;
}

# Decode a wire protocol LCB-coded unsigned integer (length implicit) and advance buffer pointer.
sub decode_lcb {
	my $buf = shift;
	my $pos = shift;

	my $bytes = decode_lcb_width($buf, $pos);
	return undef unless defined $bytes;
	return decode_my_uint($buf, $pos, $bytes);
}

# Get string and seek to the position after it
sub decode_string {
	my $buf = shift;
	my $pos = shift;

	my $length = decode_lcb($buf, $pos);

	# Return some sort of a null indicator if the string is LCB_NULL.
	return undef unless defined $length;

	# Note:  Neither the result set header, nor the column info packets, has
	#        a field describing which character set the server has encoded the
	#        metadata such as column names in.  Also, the user can change the
	#        character set used by issuing various SQL statements.  In effect
	#        making it impossible to know at this point in the state machine
	#        which character set is being used without having either parsed user
	#        queries or having done a "SELECT character_set_results" automatically
	#        before each query is sent to the server.

	my $string = substr $buf, $$pos, $length;
	$$pos += $length;
	return $string;
}

# Skip string, seeking to the position after it
sub skip_string {
	my $buf = shift;
	my $pos = shift;

	my $length = decode_lcb($buf, $pos);

	# Do nothing if string is LCB_NULL.
	$$pos += $length if defined $length;

	return undef;
}

# Note: The manual does not specify how the server handles small comments
#       terminating in CR, NEL, VT, FF, LS or PS.  Only LF is recognized
#       below (a guess).
# Note: The manual states that small comments initiated with -- must be
#       followed by a whitespace or similar such that an arithmetic double
#       minus is not mistaken for a comment, but it does not specify exactly
#       which characters are permitted.  ASCII contol characters and the space
#       character are recognized below (a guess).
# Note: The manual does not indicate whether ? tokens are allowed inside
#       conditional code such as /*!...*/.  In the expression below
#       it is not allowed (a guess).
# Note: The manual does not indicate which ASCII and Unicode characters are
#       considered whitespace.  In the expression below, the ASCII characters
#       CR, LF, VT, HT, FF and 0x20 are recognized (a guess).
my $split = qr/
	# capture each part, which is either:
	(
		# small comment in double-dash or
		--[[:cntrl:]\ ].*(?:\n|\z) |
		# small comment in hash or
		\#.*(?:\n|\z) |
		# big comment in C-style or version-conditional code or
		\/\*(?:[^\*]|\*[^\/])*(?:\*\/|\*\z|\z) |
		# whitespace
		[\ \r\n\013\t\f]+ |
		# single-quoted literal text or
		'(?:[^'\\]*|\\(?:.|\n)|'')*(?:'|\z) |
		# double-quoted literal text or
		"(?:[^"\\]*|\\(?:.|\n)|"")*(?:"|\z) |
		# schema-quoted literal text or
		`(?:[^`]*|``)*(?:`|\z) |
		# else it is either sql speak or
		(?:[^'"`\?\ \r\n\013\t\f\#\-\/]|\/[^\*]|-[^-]|--(?=[^[:cntrl:]\ ]))+ |
		# bingo: a ? placeholder
		\?
	)
/x;

# Split a MySQL-specific SQL string into parts,
# looking for '?' prepared statement placeholders.
sub tokenize {
	my $query = shift;

	my @tokens = $query =~ m/$split/g;

	# Assertion to ensure that bugs get caught; it's a bug
	# if the above regexp does not capture all of the query.
	my $parsed = join('', @tokens);
	die "query parsing failed" if length($parsed) != length($query);

	return @tokens;
}

sub quote {
	my $string = shift;
	return 'NULL' unless defined $string;

	for ($string) {
		s/\\/\\\\/g;
		s/\0/\\0/g;
		s/\n/\\n/g;
		s/\r/\\r/g;
		s/'/\\'/g;
		s/"/\\"/g;
		s/\x1a/\\Z/g;
	}

	return "'$string'";
}

sub quote_identifier {
	my $identifier = shift;
	return 'NULL' unless defined $identifier;

	for ($identifier) {
		s/`/``/g;
	}

	return "`$identifier`";
}

sub quote_wildcards {
	my $string = shift;
	return undef unless defined $string;

	for ($string) {
		s/%/\\%/g;
		s/_/\\_/g;
	}

	return $string;
}



package Net::Wire10::Password;

#
# Note: Neither Digest::SHA nor Digest::SHA1 are pure perl modules.
#       Digest::SHA::PurePerl exists, but is somewhat slow.
#
# TODO: Make use of  if Digest::SHA1 is not available.

use strict;
use warnings;

BEGIN {
	package Net::Wire10::Password;

	# Use C implementation of SHA if available (currently much faster).
	eval {
		# Note: Digest::SHA1 is used rather than Digest::SHA because the
		#       former was in my tests slightly faster when processing
		#       small amounts of data (such as passwords).
		require Digest::SHA1;
		Digest::SHA1->import("sha1");
	};
	if ($@) {
		# Otherwise fall back to the PurePerl implementation of SHA.
		require Digest::SHA::PurePerl;
		Digest::SHA::PurePerl->import("sha1");
	}
}

sub scramble {
	my $class = shift;
	my $password = shift;
	my $hash_seed = shift;
	return '' unless $password;
	return '' if length $password == 0;
	return _make_scrambled_password($hash_seed, $password);
}

sub _make_scrambled_password {
	my $hash_seed = shift;
	my $password = shift;

	my $stage1 = sha1($password);
	my $stage2 = sha1($stage1);
	my $result = sha1($hash_seed . $stage2);

	return _my_crypt($result, $stage1);
}

sub _my_crypt {
	my $s1 = shift;
	my $s2 = shift;
	my $l = length($s1) - 1;
	my $result = '';
	for my $i (0..$l) {
		$result .= pack 'C', (unpack('C', substr($s1, $i, 1)) ^ unpack('C', substr($s2, $i, 1)));
	}
	return $result;
}



package Net::Wire10::Password32;

use strict;
use warnings;

sub scramble {
	my $class = shift;
	my $password = shift;
	my $hash_seed = shift;
	my $client_capabilities = shift;

	return '' unless $password;
	return '' if length $password == 0;

	my $hsl = length $hash_seed;
	my @out;
	my @hash_pass = _get_hash($password);
	my @hash_mess = _get_hash($hash_seed);

	my ($max_value, $seed, $seed2);
	my ($dRes, $dSeed, $dMax);
	if ($client_capabilities < 1) {
		$max_value = 0x01FFFFFF;
		$seed = _xor_by_long($hash_pass[0], $hash_mess[0]) % $max_value;
		$seed2 = int($seed / 2);
	} else {
		$max_value= 0x3FFFFFFF;
		$seed  = _xor_by_long($hash_pass[0], $hash_mess[0]) % $max_value;
		$seed2 = _xor_by_long($hash_pass[1], $hash_mess[1]) % $max_value;
	}
	$dMax = $max_value;

	for (my $i=0; $i < $hsl; $i++) {
		$seed  = int(($seed * 3 + $seed2) % $max_value);
		$seed2 = int(($seed + $seed2 + 33) % $max_value);
		$dSeed = $seed;
		$dRes = $dSeed / $dMax;
		push @out, int($dRes * 31) + 64;
	}

	if ($client_capabilities == 1) {
		# Make it harder to break
		$seed  = ($seed * 3 + $seed2  ) % $max_value;
		$seed2 = ($seed + $seed2 + 33 ) % $max_value;
		$dSeed = $seed;

		$dRes = $dSeed / $dMax;
		my $e = int($dRes * 31);
		for (my $i=0; $i < $hsl ; $i++) {
			$out[$i] ^= $e;
		}
	}
	return join '', map { chr $_ } @out;
}

sub _get_hash {
	my $password = shift;

	my $nr = 1345345333;
	my $add = 7;
	my $nr2 = 0x12345671;
	my $tmp;
	my $pwlen = length $password;
	my $c;

	for (my $i=0; $i < $pwlen; $i++) {
		my $c = substr $password, $i, 1;
		next if $c eq ' ' || $c eq "\t";
		my $tmp = ord $c;
		my $value = ((_and_by_char($nr, 63) + $add) * $tmp) + $nr * 256;
		$nr = _xor_by_long($nr, $value);
		$nr2 += _xor_by_long(($nr2 * 256), $nr);
		$add += $tmp;
	}
	return (_and_by_long($nr, 0x7fffffff), _and_by_long($nr2, 0x7fffffff));
}

sub _and_by_char {
	my $source = shift;
	my $mask   = shift;

	return $source & $mask;
}

sub _and_by_long {
	my $source = shift;
	my $mask = shift || 0xFFFFFFFF;

	return _cut_off_to_long($source) & _cut_off_to_long($mask);
}

sub _xor_by_long {
	my $source = shift;
	my $mask = shift || 0;

	return _cut_off_to_long($source) ^ _cut_off_to_long($mask);
}

sub _cut_off_to_long {
	my $source = shift;

	if ($] >= 5.006) {
		$source = $source % (0xFFFFFFFF + 1) if $source > 0xFFFFFFFF;
		return $source;
	}
	while ($source > 0xFFFFFFFF) {
		$source -= 0xFFFFFFFF + 1;
	}
	return $source;
}


1;
__END__