File Coverage

lib/Neo4j/Driver/Session.pm

Criterion	Covered	Total	%
statement	92	92	100.0
branch	22	22	100.0
condition	16	16	100.0
subroutine	26	26	100.0
pod	5	6	100.0
total	161	162	100.0

use v5.14;
use warnings;

package Neo4j::Driver::Session 1.02;
# ABSTRACT: Context of work for database interactions


use Carp qw(croak);
our @CARP_NOT = qw(
	Neo4j::Driver
);
use Feature::Compat::Try;
use List::Util qw(min);
use Scalar::Util qw(blessed);
use Time::HiRes ();

use Neo4j::Driver::Net::Bolt;
use Neo4j::Driver::Net::HTTP;
use Neo4j::Driver::Transaction;
use Neo4j::Error;


sub new {
	# uncoverable pod (private method)
	my ($class, $driver) = @_;

	return Neo4j::Driver::Session::Bolt->new($driver) if $driver->config('uri')->scheme eq 'bolt';
	return Neo4j::Driver::Session::HTTP->new($driver);
}


# Connect and get ServerInfo (via Bolt HELLO or HTTP Discovery API),
# then determine the default database name for Neo4j >= 4.
sub _connect {
	my ($self, $database) = @_;

	my $neo4j_version = $self->server->agent;  # ensure contact with the server has been made
	$self->{cypher_params_v2} = 0 if $neo4j_version =~ m{^Neo4j/2\.};  # no conversion required

	$database //= $self->server->_default_database($self->{driver});
	$self->{net}->_set_database($database);
	return $self;
}


sub begin_transaction {
	my ($self) = @_;

	return $self->new_tx->_begin;
}


sub run {
	my ($self, $query, @parameters) = @_;

	return $self->new_tx->_run_autocommit($query, @parameters);
}


sub _execute {
	my ($self, $mode, $func) = @_;

	croak sprintf "%s->execute_%s() requires subroutine ref", __PACKAGE__, lc $mode unless ref $func eq 'CODE';

	$self->{retry_sleep} //= 1;
	my $time_stop = Time::HiRes::time
		+ ($self->{driver}->config('max_transaction_retry_time') // 30);  # seconds
	my $tries = 0;
	while () {
		my $tx = $self->new_tx($mode);
		$tx->{error_handler} = sub { die shift };

		try {
			$tx->_begin;
			$tx->{managed} = 1;  # Disallow commit() in $func
			wantarray ? (my @r = $func->($tx)) : (my $r = $func->($tx));
			$tx->{managed} = 0;
			$tx->commit;

			return unless defined wantarray;
			$r = $r[0] if wantarray;
			warnings::warnif closure => "Result object may not be valid outside the transaction function"
				if blessed $r && $r->isa('Neo4j::Driver::Result');

			return wantarray ? @r : $r;
		}
		catch ($e) {
			# The tx may or may not already be closed; we need to make sure
			$tx->{managed} = 0;
			try { $tx->rollback } catch ($f) {}

			# Never retry non-Neo4j errors
			croak $e unless blessed $e && $e->isa('Neo4j::Error');

			if (! $e->is_retryable || Time::HiRes::time >= $time_stop) {
				$self->{driver}->{events}->trigger( error => $e );
				return;  # in case the event handler doesn't die
			}
		}

		Time::HiRes::sleep min
			$self->{retry_sleep} * (1 << $tries++),
			$time_stop - Time::HiRes::time;
	}
}


sub execute_read {
	my ($self, $func) = @_;

	return $self->_execute( READ => $func );
}


sub execute_write {
	my ($self, $func) = @_;

	return $self->_execute( WRITE => $func );
}


sub server {
	my ($self) = @_;

	my $server_info = $self->{driver}->{server_info};
	return $server_info if defined $server_info;
	return $self->{driver}->{server_info} = $self->{net}->_server;
}




package # private
        Neo4j::Driver::Session::Bolt;
use parent -norequire => 'Neo4j::Driver::Session';


sub new {
	my ($class, $driver) = @_;

	return bless {
		cypher_params_v2 => $driver->config('cypher_params'),
		driver => $driver,
		net => Neo4j::Driver::Net::Bolt->new($driver),
	}, $class;
}


sub new_tx {
	return Neo4j::Driver::Transaction::Bolt->new(@_);
}




package # private
        Neo4j::Driver::Session::HTTP;
use parent -norequire => 'Neo4j::Driver::Session';


sub new {
	my ($class, $driver) = @_;

	return bless {
		cypher_params_v2 => $driver->config('cypher_params'),
		driver => $driver,
		net => Neo4j::Driver::Net::HTTP->new($driver),
	}, $class;
}


sub new_tx {
	return Neo4j::Driver::Transaction::HTTP->new(@_);
}


1;

__END__