# Copyright (C) 2002 Hiroyuki OYAMA. Japan. All rights reserved.

# This program is free software: you can redistribute it and/or modify

# it under the terms of the GNU General Public License as published by

# the Free Software Foundation, either version 2 of the License, or

# (at your option) any later version.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# GNU General Public License for more details.

#

# You should have received a copy of the GNU General Public License

# along with this program.  If not, see <http://www.gnu.org/licenses/>.

package Net::MySQL;

use 5.004;

use IO::Socket;

use Carp;

use vars qw($VERSION $DEBUG);

use strict;

$VERSION = '0.09';

use constant COMMAND_SLEEP          => "\x00";

use constant COMMAND_QUIT           => "\x01";

use constant COMMAND_INIT_DB        => "\x02";

use constant COMMAND_QUERY          => "\x03";

use constant COMMAND_FIELD_LIST     => "\x04";

use constant COMMAND_CREATE_DB      => "\x05";

use constant COMMAND_DROP_DB        => "\x06";

use constant COMMAND_REFRESH        => "\x07";

use constant COMMAND_SHUTDOWN       => "\x08";

use constant COMMAND_STATISTICS     => "\x09";

use constant COMMAND_PROCESS_INFO   => "\x0A";

use constant COMMAND_CONNECT        => "\x0B";

use constant COMMAND_PROCESS_KILL   => "\x0C";

use constant COMMAND_DEBUG          => "\x0D";

use constant COMMAND_PING           => "\x0E";

use constant COMMAND_TIME           => "\x0F";

use constant COMMAND_DELAYED_INSERT => "\x10";

use constant COMMAND_CHANGE_USER    => "\x11";

use constant COMMAND_BINLOG_DUMP    => "\x12";

use constant COMMAND_TABLE_DUMP     => "\x13";

use constant COMMAND_CONNECT_OUT    => "\x14";

use constant DEFAULT_PORT_NUMBER => 3306;

use constant BUFFER_LENGTH       => 1460;

use constant DEFAULT_UNIX_SOCKET => '/tmp/mysql.sock';

sub new

{

	my $class = shift;

	my %args = @_;

	my $self = bless {

		hostname   => $args{hostname},

		unixsocket => $args{unixsocket} || DEFAULT_UNIX_SOCKET,

		port       => $args{port}       || DEFAULT_PORT_NUMBER,

		database   => $args{database},

		user       => $args{user},

		password   => $args{password},

		timeout    => $args{timeout}  || 60,

		'socket'   => undef,

		salt                 => '',

		protocol_version     => undef,

		client_capabilities  => 0,

		affected_rows_length => 0,

	}, $class;

	$self->debug($args{debug});

	$self->_initialize;

	return $self;

}

sub query

{

	my $self = shift;

	my $sql = join '', @_;

	my $mysql = $self->{socket};

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

}

sub create_database

{

	my $self = shift;

	my $db_name = shift;

	my $mysql = $self->{socket};

	return $self->_execute_command(COMMAND_CREATE_DB, $db_name);

}

sub drop_database

{

	my $self = shift;

	my $db_name = shift;

	my $mysql = $self->{socket};

	return $self->_execute_command(COMMAND_DROP_DB, $db_name);

}

sub close

{

	my $self = shift;

	my $mysql = $self->{socket};

	return unless $mysql->can('send');

	my $quit_message =

		chr(length(COMMAND_QUIT)). "\x00\x00\x00". COMMAND_QUIT;

	$mysql->send($quit_message, 0);

	$self->_dump_packet($quit_message) if Net::MySQL->debug;

	$mysql->close;

}

sub get_affected_rows_length

{

	my $self = shift;

	$self->{affected_rows_length};

}

sub get_insert_id

{

	my $self = shift;

	$self->{insert_id};

}

sub create_record_iterator

{

	my $self = shift;

	return undef unless $self->has_selected_record;

	my $record = Net::MySQL::RecordIterator->new(

		$self->{selected_record}

	);

	$self->{selected_record} = undef;

	$record->parse;

	return $record;

}

sub has_selected_record

{

	my $self = shift;

	$self->{selected_record} ? 1 : undef;

}

sub is_error

{

	my $self = shift;

	$self->{error_code} ? 1 : undef;

}

sub get_error_code

{

	my $self = shift;

	$self->{error_code};

}

sub get_error_message

{

	my $self = shift;

	$self->{server_message};

}

sub debug

{

	my $class = shift;

	$DEBUG = shift if @_;

	$DEBUG;

}

sub _connect

{

	my $self = shift;

	my $mysql;

	if ($self->{hostname}) {

		printf "Use INET Socket: %s %d/tcp\n", $self->{hostname}, $self->{port}

			if $self->debug;

		$mysql = IO::Socket::INET->new(

			PeerAddr => $self->{hostname},

			PeerPort => $self->{port},

			Proto    => 'tcp',

			Timeout  => $self->{timeout} || 60,

		) or croak "Couldn't connect to $self->{hostname}:$self->{port}/tcp: $@";

	}

	else {

		printf "Use UNIX Socket: %s\n", $self->{unixsocket} if $self->debug;

		$mysql = IO::Socket::UNIX->new(

			Type => SOCK_STREAM,

			Peer => $self->{unixsocket},

		) or croak "Couldn't connect to $self->{unixsocket}: $@";

	}

	$mysql->autoflush(1);

	$self->{socket} = $mysql;

}

sub _get_server_information

{

	my $self = shift;

	my $mysql = $self->{socket};

	my $message;

	$mysql->recv($message, BUFFER_LENGTH, 0);

	$self->_dump_packet($message)

		if Net::MySQL->debug;

	my $i = 0;

	my $packet_length = ord substr $message, $i, 1;

	$i += 4;

	$self->{protocol_version} = ord substr $message, $i, 1;

	printf "Protocol Version: %d\n", $self->{protocol_version}

		if Net::MySQL->debug;

	if ($self->{protocol_version} == 10) {

		$self->{client_capabilities} = 1;

	}

	++$i;

	my $string_end = index($message, "\0", $i) - $i;

	$self->{server_version} = substr $message, $i, $string_end;

	printf "Server Version: %s\n", $self->{server_version}

		if Net::MySQL->debug;

	$i += $string_end + 1;

	$self->{server_thread_id} = unpack 'v', substr $message, $i, 2;

	$i += 4;

	$self->{salt} = substr $message, $i, 8;

	#

	$i += 8+1;

	if (length $message >= $i + 1) {

		$i += 1;

	}

	if (length $message >= $i + 18) {

		# get server_language

		# get server_status

	}

	$i += 18 - 1;

	if (length $message >= $i + 12 - 1) {

		$self->{salt} .= substr $message, $i, 12;

	}

	printf "Salt: %s\n", $self->{salt} if Net::MySQL->debug;

}

sub _request_authentication

{

	my $self = shift;

	my $mysql = $self->{socket};

	$self->_send_login_message();

	my $auth_result;

	$mysql->recv($auth_result, BUFFER_LENGTH, 0);

	$self->_dump_packet($auth_result) if Net::MySQL->debug;

	if ($self->_is_error($auth_result)) {

		$mysql->close;

		if (length $auth_result < 7) {

			croak "Timeout of authentication";

		}

		croak substr $auth_result, 7;

	}

	print "connect database\n" if Net::MySQL->debug;

}

sub _send_login_message

{

	my $self = shift;

	my $mysql = $self->{socket};

	my $body = "\0\0\x01\x0d\xa6\03\0\0\0\0\x01".

		"\x21\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0\0".

		 join "\0",

		$self->{user},

		"\x14".

		Net::MySQL::Password->scramble(

			$self->{password}, $self->{salt}, $self->{client_capabilities}

		);

	$body .= $self->{database};

	$body .= "\0";

	my $login_message = chr(length($body)-3). $body;

	$mysql->send($login_message, 0);

	$self->_dump_packet($login_message) if Net::MySQL->debug;

}

sub _execute_command

{

	my $self = shift;

	my $command = shift;

	my $sql = shift;

	my $mysql = $self->{socket};

	my $message = pack('V', length($sql) + 1). $command. $sql;

	$mysql->send($message, 0);

	$self->_dump_packet($message) if Net::MySQL->debug;

	my $result;

	$mysql->recv($result, BUFFER_LENGTH, 0);

	$self->_dump_packet($result) if Net::MySQL->debug;

	$self->_reset_status;

	if ($self->_is_error($result)) {

		return $self->_set_error_by_packet($result);

	}

	elsif ($self->_is_select_query_result($result)) {

		return $self->_get_record_by_server($result);

	}

	elsif ($self->_is_update_query_result($result)){

		return $self->_get_affected_rows_information_by_packet($result);

	}

	else {

		croak 'Unknown Result: '. $self->_get_result_length($result). 'byte';

	}

}

sub _initialize

{

	my $self = shift;

	$self->_connect;

	$self->_get_server_information;

	$self->_request_authentication;

}

sub _set_error_by_packet

{

	my $self = shift;

	my $packet = shift;

	my $error_message = $self->_get_server_message($packet);

	$self->{server_message} = $error_message;

	$self->{error_code}     = $self->_get_error_code($packet);

	return undef;

}

sub _get_record_by_server

{

	my $self = shift;

	my $packet = shift;

	my $mysql = $self->{socket};

	$self->_get_column_length($packet);

	while ($self->_has_next_packet($packet)) {

		my $next_result;

		$mysql->recv($next_result, BUFFER_LENGTH, 0);

		$packet .= $next_result;

		$self->_dump_packet($next_result) if Net::MySQL->debug;

	}

	$self->{selected_record} = $packet;

}

sub _get_affected_rows_information_by_packet

{

	my $self = shift;

	my $packet = shift;

	$self->{affected_rows_length} = $self->_get_affected_rows_length($packet);

	$self->{insert_id} = $self->_get_insert_id($packet);

	$self->{server_message} = $self->_get_server_message($packet);

	return $self->{affected_rows_length};

}

sub _is_error

{

	my $self = shift;

	my $packet = shift;

	return 1 if length $packet < 4;

	ord(substr $packet, 4) == 255;

}

sub _is_select_query_result

{

	my $self = shift;

	my $packet = shift;

	return undef if $self->_is_error($packet);

	ord(substr $packet, 4) >= 1;

}

sub _is_update_query_result

{

	my $self = shift;

	my $packet = shift;

	return undef if $self->_is_error($packet);

	ord(substr $packet, 4) == 0;

}

sub _get_result_length

{

	my $self = shift;

	my $packet = shift;

	ord(substr $packet, 0, 1)

}

sub _get_column_length

{

	my $self = shift;

	my $packet = shift;

	ord(substr $packet, 4);

}

sub _get_affected_rows_length

{

	my $self = shift;

	my $packet = shift;

	my $pos = 5;

	return Net::MySQL::Util::get_field_length($packet, \$pos);

}

sub _get_insert_id

{

	my $self = shift;

	my $packet = shift;

	return ord(substr $packet, 6, 1) if ord(substr $packet, 6, 1) != 0xfc;

	unpack 'v', substr $packet, 7, 2;

}

sub _get_server_message

{

	my $self = shift;

	my $packet = shift;

	return '' if length $packet < 7;

	substr $packet, 7;

}

sub _get_error_code

{

	my $self = shift;

	my $packet = shift;

	$self->_is_error($packet)

		or croak "_get_error_code(): Is not error packet";

	unpack 'v', substr $packet, 5, 2;

}

sub _reset_status

{

	my $self = shift;

	$self->{insert_id}       = 0;

	$self->{server_message}  = '';

	$self->{error_code}      = undef;

	$self->{selected_record} = undef;

}

sub _has_next_packet

{

	my $self = shift;

	#substr($_[0], -1) ne "\xfe";

	return substr($_[0], -5) ne "\xfe\0\0\x22\x00";

}

sub _dump_packet {

    my $self = shift;

    my $packet = shift;

    my ($method_name) = (caller(1))[3];

    my $str = sprintf "%s():\n", $method_name;

    while ($packet =~ /(.{1,16})/sg) {

        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;

        $str .= "\n"; 

    }

    print $str;

}

package Net::MySQL::RecordIterator;

use strict;

use constant NULL_COLUMN           => 251;

use constant UNSIGNED_CHAR_COLUMN  => 251;

use constant UNSIGNED_SHORT_COLUMN => 252;

use constant UNSIGNED_INT24_COLUMN => 253;

use constant UNSIGNED_INT32_COLUMN => 254;

use constant UNSIGNED_CHAR_LENGTH  => 1;

use constant UNSIGNED_SHORT_LENGTH => 2;

use constant UNSIGNED_INT24_LENGTH => 3;

use constant UNSIGNED_INT32_LENGTH => 4;

use constant UNSIGNED_INT32_PAD_LENGTH => 4;

sub new

{

	my $class = shift;

	my $packet = shift;

	bless {

		packet   => $packet,

		position => 0,

		column   => [],

	}, $class;

}

sub parse

{

	my $self = shift;

	$self->_get_column_length;

	$self->_get_column_name;