File Coverage

lib/Neo4j/Driver/Transaction.pm

Criterion	Covered	Total	%
statement	124	132	93.9
branch	56	68	82.3
condition	9	11	81.8
subroutine	25	26	96.1
pod	1	2	100.0
total	215	239	90.3

use v5.14;
use warnings;

package Neo4j::Driver::Transaction 1.02;
# ABSTRACT: Logical container for an atomic unit of work


use Carp qw(croak);
our @CARP_NOT = qw(
	Neo4j::Driver::Session
	Neo4j::Driver::Session::Bolt
	Neo4j::Driver::Session::HTTP
);
use Scalar::Util qw(blessed);

use Neo4j::Driver::Result;


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

	my $events = $session->{driver}->{events};
	my $transaction = {
		cypher_params_v2 => $session->{cypher_params_v2},
		net => $session->{net},
		mode => $mode,
		unused => 1,  # for HTTP only
		closed => 0,
		return_stats => 1,
		error_handler => sub { $events->trigger(error => shift) },
	};

	return bless $transaction, $class;
}


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

	croak 'Transaction already closed' unless $self->is_open;

	croak sprintf 'The %s->{return_graph} feature was removed', __PACKAGE__ if $self->{return_graph};

	my @statements;
	if (ref $query eq 'ARRAY') {
		croak 'Call run() with a single query statement only';
		# Consider using the private internal method _run_multiple() if you really have to
	}
	elsif ($query) {
		@statements = ( $self->_prepare($query, @parameters) );
	}
	else {
		@statements = ();
	}

	my @results = $self->{net}->_run($self, @statements);

	my $result = $results[0] // Neo4j::Driver::Result->new;
	return $result;
}


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

	if (ref $query) {
		croak 'Query cannot be unblessed reference' unless blessed $query;
		# REST::Neo4p::Query->query is not part of the documented API
		$query = '' . $query->query if $query->isa('REST::Neo4p::Query');
	}

	my $params;
	if (ref $parameters[0] eq 'HASH') {
		$params = $parameters[0];
	}
	elsif (@parameters) {
		croak 'Query parameters must be given as hash or hashref' if ref $parameters[0];
		croak 'Odd number of elements in query parameter hash' if scalar @parameters % 2 != 0;
		$params = {@parameters};
	}

	if ($self->{cypher_params_v2} && defined $params) {
		my @params_quoted = map {quotemeta} keys %$params;
		my $params_re = join '|', @params_quoted, map {"`$_`"} @params_quoted;
		$query =~ s/\{($params_re)}/\$$1/g;
	}

	my $statement = [$query, $params // {}];
	return $statement;
}




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

use Carp qw(croak);
use Feature::Compat::Try;


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

	croak "Concurrent transactions are unsupported in Bolt (there is already an open transaction in this session)" if $self->{net}->{active_tx};

	try {
		$self->{bolt_txn} = $self->{net}->_new_tx($self);
	}
	catch ($e) {
		die $e if $e !~ m/\bprotocol version\b/i;  # Bolt v1/v2
	}
	$self->{net}->{active_tx} = 1;
	$self->run('BEGIN') unless $self->{bolt_txn};
	return $self;
}


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

	croak "Concurrent transactions are unsupported in Bolt (there is already an open transaction in this session)" if $self->{net}->{active_tx};

	$self->{net}->{active_tx} = 1;  # run() requires an active tx
	my $results;
	try {
		$results = $self->run($query, @parameters);
	}
	catch ($e) {
		$self->{net}->{active_tx} = 0;
		croak $e;
	}
	$self->{net}->{active_tx} = 0;

	return $results;
}


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

	croak 'Transaction already closed' unless $self->is_open;
	croak 'Use `return` to commit a managed transaction' if $self->{managed};

	if ($self->{bolt_txn}) {
		$self->{bolt_txn}->commit;
	}
	else {
		$self->run('COMMIT');
	}
	$self->{closed} = 1;
	$self->{net}->{active_tx} = 0;
}


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

	croak 'Transaction already closed' unless $self->is_open;
	croak 'Explicit rollback of a managed transaction' if $self->{managed};

	if ($self->{bolt_txn}) {
		$self->{bolt_txn}->rollback;
	}
	else {
		$self->run('ROLLBACK');
	}
	$self->{closed} = 1;
	$self->{net}->{active_tx} = 0;
}


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

	return 0 if $self->{closed};  # what is closed stays closed
	return $self->{net}->{active_tx};
}




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

use Carp qw(croak);

# use 'rest' in place of broken 'meta', see neo4j #12306
my $RESULT_DATA_CONTENTS = ['row', 'rest'];


sub _run_multiple {
	my ($self, @statements) = @_;

	croak 'Transaction already closed' unless $self->is_open;

	return $self->{net}->_run( $self, map {
		croak '_run_multiple() expects a list of array references' unless ref eq 'ARRAY';
		croak '_run_multiple() with empty statements not allowed' unless $_->[0];
		$self->_prepare(@$_);
	} @statements );
}


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

	my $statement = $self->SUPER::_prepare($query, @parameters);
	my ($cypher, $parameters) = @$statement;

	my $json = { statement => '' . $cypher };
	$json->{resultDataContents} = $RESULT_DATA_CONTENTS;
	$json->{includeStats} = \1 if $self->{return_stats};
	$json->{parameters} = $parameters if %$parameters;

	return $json;
}


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

	# no-op for HTTP
	return $self;
}


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

	$self->{transaction_endpoint} = $self->{commit_endpoint};
	$self->{transaction_endpoint} //= URI->new( $self->{net}->{endpoints}->{new_commit} )->path;

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


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

	croak 'Use `return` to commit a managed transaction' if $self->{managed};

	$self->_run_autocommit;
}


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

	croak 'Transaction already closed' unless $self->is_open;
	croak 'Explicit rollback of a managed transaction' if $self->{managed};

	$self->{net}->_request($self, 'DELETE') if $self->{transaction_endpoint};
	$self->{closed} = 1;
}


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

	return 0 if $self->{closed};
	return 1 if $self->{unused};
	return $self->{net}->_is_active_tx($self);
}


1;

__END__