File Coverage

blib/lib/DBD/Wire10.pm

Criterion	Covered	Total	%
statement	36	370	9.7
branch	0	158	0.0
condition	0	29	0.0
subroutine	12	43	27.9
pod	1	1	100.0
total	49	601	8.1

# TODO: Find out why DBI croaks under PerlSvc when DBD::Wire10 and Net::Wire10
#       files are in UTF-8 (with BOM), this prevents using Unicode for the POD docs.

# TODO: Put in a place where people expect to find this driver, maybe alias
#       under package names matching the supported database systems?
#       (DBD::MySQL::Wire10, DBD::Sphinx::Wire10, DBD::Drizzle::Wire10 etc)

package DBD::Wire10;

use strict;
use warnings;
use DBI;
use vars qw($VERSION $err $errstr $state $drh);

$VERSION = '1.08';
$err = 0;
$errstr = '';
$state = undef;
$drh = undef;

our $methods_already_installed = 0;

sub driver {
	return $drh if $drh;

	my $class = shift;
	my $attr  = shift;
	$class .= '::dr';

	# TODO: core->ping() + core->connect() useful as a generic mechanism
	unless ($methods_already_installed++) {
		eval {
			my $method = "DBI::db::reconnect";
			my $file = "__FILENAME__";
			my $info = {};
			DBI->_install_method($method, $file, $info);
		};
		warn "Failed to register reconnect method: $@" if $@;
	}

	$drh = DBI::_new_drh($class, {
		Name        => 'Wire10',
		Version     => $VERSION,
	}, {});
	return $drh;
}

sub CLONE {
	undef $drh;
}

# TODO: Is there a shrinkwrapped function to do this
sub _parse_dsn {
	my $class = shift;
	my ($dsn, $args) = @_;
	my ($hash, $var, $val);
	return undef if ! defined $dsn;

	while (length $dsn) {
		if ($dsn =~ /([^:;]*)[:;](.*)/) {
			$val = $1;
			$dsn = $2;
		} else {
			$val = $dsn;
			$dsn = '';
		}
		if ($val =~ /([^=]*)=(.*)/) {
			$var = $1;
			$val = $2;
			if ($var eq 'hostname' || $var eq 'host') {
				$hash->{host} = $val;
			} elsif ($var eq 'db' || $var eq 'dbname') {
				$hash->{database} = $val;
			} else {
				$hash->{$var} = $val;
			}
		} else {
			for $var (@$args) {
				if (! defined($hash->{$var})) {
					$hash->{$var} = $val;
					last;
				}
			}
		}
	}
	return $hash;
}

sub _parse_dsn_host {
	my ($class, $dsn) = @_;
	my $hash = $class->_parse_dsn($dsn, ['host', 'port']);
	return ($hash->{host}, $hash->{port});
}



package DBD::Wire10::dr;

use strict;
use warnings;
use Net::Wire10;

# Note: rather undocumented, for now blindly hoping that 0 means auto-detect.
$DBD::Wire10::dr::imp_data_size = 0;

sub connect {
	my $drh = shift;
	my ($dsn, $user, $password, $attrhash) = @_;

	my $data_source_info = DBD::Wire10->_parse_dsn(
		$dsn, ['database', 'host', 'port'],
	);
	$user     ||= '';
	$password ||= '';

	my $dbh = DBI::_new_dbh($drh, { Name => $dsn });
	eval {
		# See note in take_imp_data().
		my $wire = delete $attrhash->{dbi_imp_data};
		unless (defined $wire) {
			$wire = Net::Wire10->new(
				host            => $data_source_info->{host},
				port            => $data_source_info->{port},
				database        => $data_source_info->{database},
				user            => $user,
				password        => $password,
				debug           => $attrhash->{wire10_debug} || undef,
				connect_timeout => $attrhash->{wire10_connect_timeout} || undef,
				query_timeout   => $attrhash->{wire10_query_timeout} || undef,
			);
			$wire->connect;
		};
		$dbh->STORE('wire10_driver_dbh', $wire);
		$dbh->STORE('wire10_thread_id', $wire->{server_thread_id});
		$dbh->STORE('wire10_server_version', $wire->{server_version});
	};
	if ($@) {
		$dbh->DBI::set_err(-1, $@);
		return undef;
	}
	$dbh->STORE('Active', 1);
	return $dbh;
}

sub data_sources {
	return ("DBI:Wire10:");
}



package DBD::Wire10::db;

use strict;
use warnings;

$DBD::Wire10::db::imp_data_size = 0;

sub quote {
	my $dbh = shift;
	my ($statement, $type) = @_;
	return Net::Wire10::Util::quote($statement);
}

sub quote_identifier {
	my $dbh = shift;
	my $name = shift;
	return Net::Wire10::Util::quote_identifier($name);
}

sub prepare {
	my $dbh = shift;
	my ($statement, @attribs) = @_;
	my $wire = $dbh->FETCH('wire10_driver_dbh');

	my $sth;
	eval {
		$sth = DBI::_new_sth($dbh, {Statement => $statement});
		DBD::Wire10::st::_constructor($sth, $wire, $statement);
	};
	if ($@) {
		$dbh->DBI::set_err(-1, $@);
		return undef;
	}
	return $sth;
}

sub STORE {
	my $dbh = shift;
	my ($key, $value) = @_;

	if ($key =~ /^AutoCommit$/) {
		my $wire = $dbh->FETCH('wire10_driver_dbh');
		eval {
			$wire->query("SET AUTOCOMMIT=".$value);
		};
		if ($@) {
			die $@ unless $wire->is_connected;
			if ($dbh->FETCH('Warn')) {
				warn "Server does not allow setting AUTOCOMMIT: $@";
			}
		}
		# Can't store as AutoCommit via SUPER::STORE, not sure why.
		$dbh->STORE('wire10_autocommit', $value);
		return 1;
	}

	if ($key =~ /^(?:wire10_connect_timeout)$/) {
		my $wire = $dbh->FETCH('wire10_driver_dbh');
		$wire->{connect_timeout} = $value;
		return 1;
	}

	if ($key =~ /^(?:wire10_query_timeout)$/) {
		my $wire = $dbh->FETCH('wire10_driver_dbh');
		$wire->{query_timeout} = $value;
		return 1;
	}

	if ($key =~ /^(?:wire10_debug)$/) {
		my $wire = $dbh->FETCH('wire10_driver_dbh');
		$wire->{debug} = $value;
		return 1;
	}

	if ($key =~ /^(?:wire10_.*)$/) {
		$dbh->{$key} = $value;
		return 1;
	}

	return $dbh->SUPER::STORE($key, $value);
}

sub FETCH {
	my $dbh = shift;
	my $key = shift;

	if ($key =~ /^(?:wire10_connect_timeout)$/) {
		my $wire = $dbh->FETCH('wire10_driver_dbh');
		return $wire->{connect_timeout};
	}
	if ($key =~ /^(?:wire10_query_timeout)$/) {
		my $wire = $dbh->FETCH('wire10_driver_dbh');
		return $wire->{query_timeout};
	}
	if ($key =~ /^(?:wire10_debug)$/) {
		my $wire = $dbh->FETCH('wire10_driver_dbh');
		return $wire->{debug};
	}

	if ($key =~ /^AutoCommit$/) {
		# See comment in STORE.
		return $dbh->FETCH('wire10_autocommit');
	}

	return $dbh->{$key} if $key =~ /^(?:wire10_.*)$/;
	return $dbh->SUPER::FETCH($key);
}

sub commit {
	my $dbh = shift;

	if ($dbh->FETCH('AutoCommit')) {
		if ($dbh->FETCH('Warn')) {
			warn 'Commit ineffective while AutoCommit is on';
		}
	}

	my $wire = $dbh->FETCH('wire10_driver_dbh');
	$wire->query("COMMIT");

	return 1;
}

sub rollback {
	my $dbh = shift;

	if ($dbh->FETCH('AutoCommit')) {
		if ($dbh->FETCH('Warn')) {
			warn 'Rollback ineffective while AutoCommit is on';
		}
	}

	my $wire = $dbh->FETCH('wire10_driver_dbh');
	$wire->query("ROLLBACK");

	return 1;
}

sub ping {
	my $dbh = shift;
	my $wire = $dbh->FETCH('wire10_driver_dbh');

	eval {
		$wire->ping;
	};

	my $error = $wire->get_error_info;
	if ($error) {
		$dbh->DBI::set_err($error->get_error_code || -1, $error->get_error_message, $error->get_error_state);
	} elsif ($@) {
		$dbh->DBI::set_err(-1, $@);
	}

	return $wire->is_connected;
}

sub reconnect {
	my $dbh = shift;
	my $wire = $dbh->FETCH('wire10_driver_dbh');

	if ($wire->is_connected) {
		eval {
			$wire->ping;
		};
	}
	# ping() also sets is_connected, unnecessary to check ping return value.
	if (not $wire->is_connected) {
		eval {
			$wire->connect;
			# The below is copy/paste from the drh connect() call.
			$dbh->STORE('wire10_thread_id', $wire->{server_thread_id});
			$dbh->STORE('wire10_server_version', $wire->{server_version});
			$dbh->STORE('Active', 1);
			$dbh->STORE('AutoCommit', $dbh->FETCH('AutoCommit'));
		};
		# Return 0 on failure.
		my $error = $wire->get_error_info;
		if ($error) {
			$dbh->DBI::set_err($error->get_error_code || -1, $error->get_error_message, $error->get_error_state);
			return 0;
		} elsif ($@) {
			$dbh->DBI::set_err(-1, $@);
			return 0;
		}
		# Return 1 if connection was reestablished.
		return 1;
	}
	# Return -1 if nothing besides a protocol ping was done.
	return -1;
}

sub disconnect {
	my $dbh = shift;
	my $wire = $dbh->FETCH('wire10_driver_dbh');
	$wire->disconnect if defined $wire;
	$dbh->STORE('wire10_thread_id', undef);
	$dbh->STORE('Active', 0);
	return 1;
}

sub DESTROY {
	my $dbh = shift;
	$dbh->disconnect if $dbh->FETCH('Active');
	$dbh->SUPER::DESTROY;
}

sub last_insert_id {
	my $dbh = shift;
	return $dbh->FETCH('wire10_insertid')
}

# TODO: Support more get_info properties as needed.
sub get_info {
	my $dbh = shift;
	my $type = shift;
	# 17: SQL_DBMS_NAME
	# Difficult to return something intelligent here, the server
	# only reports a version, not a daemon name in the handshake.
	return 'Wire10' if $type == 17;
	# 18: SQL_DBMS_VER
	return $dbh->FETCH('wire10_server_version') if $type == 18;
	# 29: SQL_IDENTIFIER_QUOTE_CHAR
	return '`' if $type == 29;
	# 41: SQL_CATALOG_NAME_SEPARATOR
	return '.' if $type == 41;
	# 114: SQL_CATALOG_LOCATION
	# According to MSDN, 0 means "catalog not supported" which is accurate.
	# (The server happily accepts, discards and prints a catalog named
	#  'def', though.)
	return 0 if $type == 114;
	# Return undef for unknown and unsupported attributes.
	return undef;
}

sub take_imp_data {
	my $dbh = shift;

	# Finish any active statements (important if streaming enabled).
	for my $sth (@{$dbh->{ChildHandles} || []}) {
		next unless $sth;
		$sth->finish if $sth->{Active};
	}

	# Take out core driver and remove reference to it.
	my $wire = $dbh->FETCH('wire10_driver_dbh');
	$dbh->STORE('wire10_driver_dbh', undef);

	# Remove reference to dbh from drh, probably also destroys dbh.
	$dbh->SUPER::take_imp_data;

	# Note: It would be nice to serialize or tie the core such that
	#       it can be be shared among interpreters running in different
	#       processes or threads.  Unfortunately, neither of the available
	#       modules, Storable and threads::shared, seem to be able to
	#       cope with socket handles:
	#
	#       Storable error: "Can't store GLOB items"
	#       threads::shared error: "Invalid value for shared scalar"
	#
	#       For now, we just return the core driver and expect that the
	#       caller serialize and deserialize the object if the caller needs
	#       to use it from a different context.

	# Return the core driver.
	return $wire;
}



package DBD::Wire10::st;

use strict;
use warnings;
use DBI qw(:sql_types);

$DBD::Wire10::st::imp_data_size = 0;

# TODO: Find out if DBI already calls a DBD st constructor somewhere
sub _constructor {
	my $sth = shift;
	my $wire = shift;
	my $sql = shift;

	my $ps = $wire->prepare($sql);

	# Store driver handle and prepared statement for later.
	$sth->STORE('wire10_driver_sth', $wire);
	$sth->STORE('wire10_prepared', $ps);
	$sth->STORE('NUM_OF_PARAMS', $ps->get_marker_count);
}

sub bind_param {
	my $sth = shift;
	my ($index, $value, $attr) = @_;
	my $binary = _test_for_binary_flag($sth, $attr);
	my $ps = $sth->FETCH('wire10_prepared');
	$ps->set_parameter($index, $value, $binary);
	return 1;
}

sub _test_for_binary_flag {
	my $sth = shift;
	my $attr = shift;
	return 0 unless defined $attr;
	my $binary = Net::Wire10::DATA_BINARY;
	my $text = Net::Wire10::DATA_TEXT;
	my $sqltype;
	# May be undefined.
	$sqltype = $attr if ref($attr) eq '';
	$sqltype = $attr->{TYPE} if ref($attr) eq 'HASH';
	if (defined $sqltype) {
		return $binary if $sqltype == SQL_BINARY;
		return $binary if $sqltype == SQL_VARBINARY;
		return $binary if $sqltype == SQL_LONGVARBINARY;
		return $binary if $sqltype == SQL_BLOB;
	}
	# For testing Oracle-based code with MySQL.
	# ORA_BLOB is 113, defined in Oracle.h.
	my $oratype;
	$oratype = $attr->{ora_type} if ref($attr) eq 'HASH';
	return $binary if defined $oratype and $oratype == 113;
	return $text;
}

sub execute {
	my $sth = shift;
	my @new_params = @_;
	my $dbh = $sth->{Database};
	my $wire = $sth->FETCH('wire10_driver_sth');
	my $ps = $sth->FETCH('wire10_prepared');

	unless (defined($ps)) {
		$sth->DBI::set_err(-1, "execute without prepare");
		return undef;
	}

	if (scalar(@new_params) > 0) {
		$ps->clear_parameter;
		my $i = 1;
		foreach my $p (@new_params) {
			$ps->set_parameter($i++, $p, 0);
		}
	}

	my $rowcount = eval {
		$sth->finish;
		my $stream_results = $sth->FETCH('wire10_streaming') || 0;
		my $res = $stream_results ? $ps->stream : $ps->query;

		die if $wire->get_error_info;

		$sth->STORE('wire10_warning_count', $res->get_warning_count);
		# For backward compatibility and/or do(), store in dbh too.
		my $dbh = $sth->{Database};
		$dbh->STORE('wire10_warning_count', $res->get_warning_count);

		if ($res->has_results) {
			$sth->{wire10_iterator} = $res;
			my @names = $res->get_column_info("name");
			$sth->STORE('NUM_OF_FIELDS', scalar @names);
			$sth->STORE('NAME', [@names]);
			my @flags = $res->get_column_info("flags");
			my @nullable = map { ! $_ & Net::Wire10::COLUMN_NOT_NULL } @flags;
			$sth->STORE('NULLABLE', [@nullable]);
			# DBI docs says this is important for bind_columns and bind_cols.
			$sth->STORE('Active', 1);
			# Note: Emulate DBD-MySQL by not resetting insertid in dbh (only sth).
			$sth->STORE('wire10_insertid', undef);
			$sth->{wire10_rows} = $res->get_no_of_selected_rows;
			return $res->get_no_of_selected_rows;
		} else {
			$sth->{wire10_iterator} = undef;
			$sth->STORE('NUM_OF_FIELDS', undef);
			$sth->STORE('NAME', undef);
			$sth->STORE('NULLABLE', undef);
			my $insertid;
			eval {
				$insertid = $res->get_insert_id;
			};
			if ($@) {
				# If the insert_id is too big for this Perl to handle,
				# extract it using an alternate method.
				my $res = $wire->query('SELECT LAST_INSERT_ID()');
				$insertid = $res->next_array()->[0] || 0;
			}
			$sth->STORE('wire10_insertid', $insertid);
			# For backward compatibility and/or do(), store in dbh too.
			$dbh->STORE('wire10_insertid', $insertid);
			$sth->{wire10_rows} = $res->get_no_of_affected_rows;
			return $res->get_no_of_affected_rows;
		}
	};

	my $error = $wire->get_error_info;
	if ($error) {
		$sth->DBI::set_err($error->get_error_code || -1, $error->get_error_message, $error->get_error_state);
		return undef;
	} elsif ($@) {
		$sth->DBI::set_err(-1, $@);
		return undef;
	}

	return $rowcount ? $rowcount : '0E0';
}

sub cancel {
	my $sth = shift;
	my $wire = $sth->FETCH('wire10_driver_sth');

	eval {
		$wire->cancel;
	};

	if ($@) {
		$sth->DBI::set_err(-1, $@);
		return undef;
	}

	return 1;
}

sub finish {
	my $sth = shift;
	my $dbh = $sth->{Database};
	# If in streaming mode, flush remaining results.
	my $iterator = $sth->{wire10_iterator};
	$iterator->spool if defined $iterator;
	$sth->{wire10_iterator} = undef;
	$sth->STORE('Active', 0);
	$sth->SUPER::finish;
}

sub fetchrow_arrayref {
	my $sth = shift;

	my $iterator = $sth->FETCH('wire10_iterator');
	unless ($iterator) {
		if ($sth->FETCH('Warn')) {
			warn 'fetch() without execute(), previous execute() failed, executed query does not have results, or last row was already fetched';
		}
		return undef;
	}

	my $row = undef;
	eval {
		$row = $iterator->next_array;
	};
	if ($@) {
		$sth->DBI::set_err(-1, $@);
		return undef;
	}
	if (! $row) {
		$sth->finish;
		return undef;
	}

	if ($sth->FETCH('ChopBlanks')) {
		map {s/\s+$//} @$row;
	}

	return $sth->_set_fbav($row);
}

# required alias for fetchrow_arrayref
*fetch = \&fetchrow_arrayref;

sub rows {
	my $sth = shift;
	my $rows = $sth->FETCH('wire10_rows');
	return $rows unless $rows == -1;
	return $sth->SUPER::rows;
}

sub FETCH {
	my $sth = shift;
	my $key = shift;

	return $sth->{NAME} if $key eq 'NAME';
	return $sth->{NULLABLE} if $key eq 'NULLABLE';
	return $sth->{$key} if $key =~ /^wire10_/;
	return $sth->SUPER::FETCH($key);
}

sub STORE {
	my $sth = shift;
	my ($key, $value) = @_;

	if ($key eq 'NAME') {
		$sth->{NAME} = $value;
		return 1;
	}
	if ($key eq 'NULLABLE') {
		$sth->{NULLABLE} = $value;
		return 1;
	}
	if ($key =~ /^wire10_/) {
		$sth->{$key} = $value;
		return 1;
	}

	return $sth->SUPER::STORE($key, $value);
}

sub DESTROY {
	my $sth = shift;
	$sth->finish if $sth->FETCH('Active');
	$sth->SUPER::DESTROY;
}



1;
__END__