File Coverage

lib/Apache/Solr.pm

Criterion	Covered	Total	%
statement	134	301	44.5
branch	54	192	28.1
condition	32	79	40.5
subroutine	25	49	51.0
pod	25	27	92.5
total	270	648	41.6

# Copyrights 2012-2025 by [Mark Overmeer].
#  For other contributors see ChangeLog.
# See the manual pages for details on the licensing terms.
# Pod stripped from pm file by OODoc 2.03.
# This code is part of distribution Apache-Solr.  Meta-POD processed with
# OODoc into POD and HTML manual-pages.  See README.md
# Copyright Mark Overmeer.  Licensed under the same terms as Perl itself.

package Apache::Solr;{
our $VERSION = '1.11';
}


use warnings;
use strict;

use Apache::Solr::Tables;
use Log::Report    qw(solr);

use Scalar::Util   qw/blessed/;
use Encode         qw/encode/;
use Scalar::Util   qw/weaken/;

use URI            ();
use LWP::UserAgent ();
use MIME::Types    ();

use constant {
	LATEST_SOLR_VERSION => '9.8',  # newest support by this module
	ENETDOWN    => 100,   # error codes may not be available on all platforms
	ENETUNREACH => 101,   #    so cannot use Errno.
};

# overrule this when your host has a different unique field
our $uniqueKey  = 'id';
my  $mimetypes  = MIME::Types->new;
my  $http_agent;

sub _to_bool($)
{	my $b = shift;
	   !defined $b ? undef
	 : ($b && $b ne 'false' && $b ne 'off') ? 'true'
	 : 'false';
}


sub new(@)
{	my ($class, %args) = @_;
	if($class eq __PACKAGE__)
	{	my $format = delete $args{format} || 'XML';
		$format eq 'XML' || $format eq 'JSON'
			or panic "unknown communication format '$format' for solr";
		$class .= '::' . $format;
		eval "require $class"; panic $@ if $@;
	}
	(bless {}, $class)->init(\%args)
}

sub init($)
{	my ($self, $args) = @_;
	$self->server($args->{server});
	$self->{AS_core}     = $args->{core};
	$self->{AS_commit}   = exists $args->{autocommit} ? $args->{autocommit} : 1;
	$self->{AS_sversion} = $args->{server_version} || LATEST_SOLR_VERSION;
	$self->{AS_retry_wait} = $args->{retry_wait} // 5;  # seconds
	$self->{AS_retry_max}  = $args->{retry_max}  // 60;

	$http_agent = $self->{AS_agent} =
		$args->{agent} || $http_agent || LWP::UserAgent->new(keep_alive=>1);

	weaken $http_agent;
	$self;
}

#---------------

sub core(;$) { my $s = shift; @_ ? $s->{AS_core}   = shift : $s->{AS_core} }
sub autocommit(;$) { my $s = shift; @_ ? $s->{AS_commit} = shift : $s->{AS_commit} }
sub agent() {shift->{AS_agent}}
sub serverVersion() {shift->{AS_sversion}}


sub server(;$)
{	my ($self, $uri) = @_;
	$uri or return $self->{AS_server};
	$uri = URI->new($uri)
		unless blessed $uri && $uri->isa('URI');
	$self->{AS_server} = $uri;
}


#--------------------------

sub select(@)
{	my $self = shift;
	my $args = @_ && ref $_[0] eq 'HASH' ? shift : {};
	$self->_select($args, scalar $self->expandSelect(@_));
}
sub _select($$) {panic "not extended"}


sub queryTerms(@)
{	my $self  = shift;
	$self->_terms(scalar $self->expandTerms(@_));
}
sub _terms(@) {panic "not implemented"}

#-------------------------------------

sub addDocument($%)
{	my ($self, $docs, %args) = @_;
	$docs  = [ $docs ] if ref $docs ne 'ARRAY';

	my $sv = $self->serverVersion;

	my (%attrs, %params);
	$params{commit} = _to_bool(exists $args{commit} ? $args{commit} : $self->autocommit);

	if(my $cw = $args{commitWithin})
	{	if($sv lt '3.4') { $attrs{commit} = 'true' }
		else { $attrs{commitWithin} = int($cw * 1000) }
	}

	$attrs{overwrite} = _to_bool delete $args{overwrite}
		if exists $args{overwrite};

	foreach my $depr (qw/allowDups overwritePending overwriteCommitted/)
	{	if(exists $args{$depr})
		{	  if($sv ge '4.0') { $self->removed("add($depr)"); delete $args{$depr} }
			elsif($sv ge '1.0') { $self->deprecated("add($depr)") }
			else { $attrs{$depr} = _to_bool delete $args{$depr} }
		}
	}

	$self->_add($docs, \%attrs, \%params);
}


sub commit(%)
{	my ($self, %args) = @_;
	my $sv = $self->serverVersion;

	my %attrs;
	if(exists $args{waitFlush})
	{	  if($sv ge '4.0')
			 { $self->removed("commit(waitFlush)"); delete $args{waitFlush} }
		elsif($sv ge '1.4') { $self->deprecated("commit(waitFlush)") }
		else { $attrs{waitFlush} = _to_bool delete $args{waitFlush} }
	}

	$attrs{waitSearcher} = _to_bool delete $args{waitSearcher}
		if exists $args{waitSearcher};

	if(exists $args{softCommit})
	{	if($sv lt '4.0') { $self->ignored("commit(softCommit)") }
		else { $attrs{softCommit} = _to_bool delete $args{softCommit} }
	}

	if(exists $args{expungeDeletes})
	{	if($sv lt '1.4') { $self->ignored("commit(expungeDeletes)") }
		else { $attrs{expungeDeletes} = _to_bool delete $args{expungeDeletes} }
	}

	$self->_commit(\%attrs);
}
sub _commit($) {panic "not implemented"}


sub optimize(%)
{	my ($self, %args) = @_;
	my $sv = $self->serverVersion;

	my %attrs;
	if(exists $args{waitFlush})
	{	if($sv ge '4.0') { $self->removed("commit(waitFlush)"); delete $args{waitFlush} }
		elsif($sv ge '1.4') { $self->deprecated("optimize(waitFlush)") }
		else { $attrs{waitFlush} = _to_bool delete $args{waitFlush} }
	}

	$attrs{waitSearcher} = _to_bool delete $args{waitSearcher}
		if exists $args{waitSearcher};

	if(exists $args{softCommit})
	{	if($sv lt '4.0') { $self->ignored("optimize(softCommit)") }
		else { $attrs{softCommit} = _to_bool delete $args{softCommit} }
	}

	if(exists $args{maxSegments})
	{	if($sv lt '1.3') { $self->ignored("optimize(maxSegments)") }
		else { $attrs{maxSegments} = delete $args{maxSegments} }
	}

	$self->_optimize(\%attrs);
}
sub _optimize($) {panic "not implemented"}


sub delete(%)
{	my ($self, %args) = @_;

	my %attrs;
	$attrs{commit} = _to_bool(exists $args{commit} ? $args{commit} : $self->autocommit);

	if(exists $args{fromPending})
	{	$self->deprecated("delete(fromPending)");
		$attrs{fromPending}   = _to_bool delete $args{fromPending};
	}
	if(exists $args{fromCommitted})
	{	$self->deprecated("delete(fromCommitted)");
		$attrs{fromCommitted} = _to_bool delete $args{fromCommitted};
	}

	my @which;
	if(my $id = $args{id})
	{	push @which, map +(id => $_), ref $id eq 'ARRAY' ? @$id : $id;
	}
	if(my $q  = $args{query})
	{	push @which, map +(query => $_), ref $q  eq 'ARRAY' ? @$q  : $q;
	}
	@which or return;

	# JSON calls do not accept multiple ids at once (it seems in 4.0)
	my $result;
	if($self->serverVersion ge '1.4' && !$self->isa('Apache::Solr::JSON'))
	{	$result = $self->_delete(\%attrs, \@which);
	}
	else
	{	# old servers accept only one id or query per delete
		$result = $self->_delete(\%attrs, [splice @which, 0, 2]) while @which;
	}
	$result;
}
sub _delete(@) {panic "not implemented"}


sub rollback()
{	my $self = shift;
	$self->serverVersion ge '1.4'
		or error __x"Rollback not supported by solr server";

	$self->_rollback;
}


sub extractDocument(@)
{	my $self  = shift;

	$self->serverVersion ge '1.4'
		or error __x"extractDocument() requires Solr v1.4 or higher";

	my %p   = $self->expandExtract(@_);
	my $data;

	# expand* changes '_' into '.'
	my $ct  = delete $p{'content.type'};
	my $fn  = delete $p{file};
	$p{'resource.name'} ||= $fn if $fn && !ref $fn;

	$p{commit} = _to_bool $self->autocommit
		unless exists $p{commit};

	if(defined $p{string})
	{	# try to avoid copying the data, which can be huge
		$data = $ct =~ m!^text/!i
		  ? \encode(utf8 => (ref $p{string} eq 'SCALAR' ? ${$p{string}} : $p{string}))
		  : (ref $p{string} eq 'SCALAR' ? $p{string} : \$p{string} );

		delete $p{string};
	}
	elsif($fn)
	{	local $/;
		if(ref $fn eq 'GLOB') { $data = \<$fn> }
		else
		{	local *IN;
			open IN, '<:raw', $fn or fault __x"Cannot read document from {fn}", fn => $fn;
			$data = \;
			close IN or fault __x"Read error for document {fn}", fn => $fn;
			$ct ||= $mimetypes->mimeTypeOf($fn);
		}
	}
	else
	{	error __x"Extract requires document as file or string";
	}

	$self->_extract([%p], $data, $ct);
}

sub _extract($) { panic "not implemented" }

#-------------------------

sub _core_admin($@)
{	my ($self, $action, $params) = @_;
	$params->{core} ||= $self->core;

	my $endpoint = $self->endpoint('cores', core => 'admin', params => $params);
	my @params   = %$params;
	my $result   = Apache::Solr::Result->new(params => [ %$params ], endpoint => $endpoint, core => $self);

	$self->request($endpoint, $result);
	$result;
}


sub coreStatus(%)
{	my ($self, %args) = @_;
	$self->_core_admin('STATUS', \%args);
}


sub coreReload(%)
{	my ($self, %args) = @_;
	$self->_core_admin('RELOAD', \%args);
}


sub coreUnload($%)
{	my ($self, %args) = @_;
	$self->_core_admin('UNLOAD', \%args);
}

#--------------------------

sub _calling_sub()
{	for(my $i=0;$i <10; $i++)
	{	my $sub = (caller $i)[3];
		return $sub if !$sub || index($sub, 'Apache::Solr::') < 0;
	}
}

sub _simpleExpand($$$)
{	my ($self, $p, $prefix) = @_;
	my @p  = ref $p eq 'HASH' ? %$p : @$p;
	my $sv = $self->serverVersion;

	my @t;
	while(@p)
	{	my ($k, $v) = (shift @p, shift @p);
		$k =~ s/_/./g;
		$k = $prefix.$k if defined $prefix && index($k, $prefix)!=0;
		my $param   = $k =~ m/^f\.[^\.]+\.(.*)/ ? $1 : $k;

		my ($dv, $iv);
		if(($dv = $deprecated{$param}) && $sv ge $dv)
		{	my $command = _calling_sub;
			$self->deprecated("$command($param) since $dv");
		}
		elsif(($iv = $introduced{$param}) && $iv gt $sv)
		{	my $command = _calling_sub;
			$self->ignored("$command($param) introduced in $iv");
			next;
		}

		push @t, $k => $boolparams{$param} ? _to_bool($_) : $_
			for ref $v eq 'ARRAY' ? @$v : $v;
	}
	@t;
}


sub expandTerms(@)
{	my $self = shift;
	my $p = @_==1 ? shift : [@_];
	my @t = $self->_simpleExpand($p, 'terms.');
	wantarray ? @t : \@t;
}


sub _expand_flatten($$)
{	my ($self, $v, $prefix) = @_;
	my @l = ref $v eq 'HASH' ? %$v : @$v;
	my @s;
	push @s, $prefix.(shift @l) => (shift @l) while @l;
	@s;
}

sub expandExtract(@)
{	my $self = shift;
	my @p = @_==1 ? @{(shift)} : @_;
	my @s;
	while(@p)
	{	my ($k, $v) = (shift @p, shift @p);
		if(!ref $v || ref $v eq 'SCALAR')
			 { push @s, $k => $v }
		elsif($k eq 'literal' || $k eq 'literals')
			 { push @s, $self->_expand_flatten($v, 'literal.') }
		elsif($k eq 'fmap' || $k eq 'boost' || $k eq 'resource')
			 { push @s, $self->_expand_flatten($v, "$k.") }
		else { panic "unknown set '$k'" }
	}

	my @t = @s ? $self->_simpleExpand(\@s) : ();
	wantarray ? @t : \@t;
}


# probably more config later, currently only one column
# "also-per-field" means, not only $set.$more, but also f.$field.$set.$more
my %sets =   #also-per-field?
  ( expand  => [0]
  , facet   => [1]
  , hl      => [1]
  , mlt     => [0]
  , stats   => [0]
  , suggest => [0]
  , group   => [0]
  );

sub expandSelect(@)
{	my $self = shift;
	my @s;
	my (@flat, %seen_set);
	while(@_)
	{	my ($k, $v) = (shift, shift);
		$k =~ s/_/./g;
		my @p = split /\./, $k;

		# fields are $set.$more or f.$field.$set.$more
		my $per_field    = $p[0] eq 'f' && @p > 2;
		my ($set, $more) = $per_field ? @p[2,3] : @p[0,1];

		if(my $def = $sets{$set})
		{	$seen_set{$set} = 1;
			!$per_field || $def->[0]
			   or error __x"Set {set} cannot be used per field, in {field}", set => $set, field => $k;

			if(ref $v eq 'HASH')
			{	!$more or error __x"Field {field} is not simple for a set", field => $k;
				push @s, $self->_simpleExpand($v, "$k.");
			}
			elsif($more)    # skip $set=true for now
			{	push @flat, $k => $v;
			}
		}
		elsif(ref $v eq 'HASH')
		{	error __x"Unknown set {set}", set => $set;
		}
		else
		{	push @flat, $k => $v;
		}
	}
	push @flat, %seen_set;
	unshift @s, $self->_simpleExpand(\@flat);
	wantarray ? @s : \@s;
}


sub deprecated($)
{	my ($self, $msg) = @_;
	return if $self->{AS_depr_msg}{$msg}++;  # report only once
	warning __x"Deprecated solr {message}", message => $msg;
}


sub ignored($)
{	my ($self, $msg) = @_;
	return if $self->{AS_ign_msg}{$msg}++;  # report only once
	warning __x"Ignored solr {message}", message => $msg;
}


sub removed($)
{	my ($self, $msg) = @_;
	return if $self->{AS_rem_msg}{$msg}++;  # report only once
	warning __x"Removed solr {message}", message => $msg;
}


#------------------------

sub endpoint($@)
{	my ($self, $action, %args) = @_;
	my $core = $args{core} || $self->core;
	my $take = $self->server->clone;    # URI
	$take->path($take->path . (defined $core ? "/$core" : '') . "/$action");

	# make parameters ordered
	my $params = $args{params} || [];
	$params	= [ %$params ] if ref $params eq 'HASH';
	@$params or return $take;

	# remove paramers with undefined value
	my @p = @$params;
	my @params;
	while(@p)
	{	push @params, $p[0] => $p[1] if defined $p[1];
		shift @p, shift @p;
	}

	$take->query_form(@params) if @params;
	$take;
}


sub request($$;$$)
{	my ($self, $url, $result, $body, $body_ct) = @_;

	my $req;
	if($body)
	{	# request with 'form' payload
		$req       = HTTP::Request->new
		  ( POST => $url
		  , [ Content_Type        => $body_ct
			, Content_Disposition => 'form-data; name="content"'
			]
		  , (ref $body eq 'SCALAR' ? $$body : $body)
		  );
	}
	else
	{	# request without payload
		$req = HTTP::Request->new(GET => $url);
	}

	$result->request($req);

	my $resp;
	my $retries = $self->{AS_retry_max};
	my $wait    = $self->{AS_retry_wait};
	my $start   = time;

	while($retries--)
	{	$resp = $self->agent->request($req);
		$result->response($resp);
		my $dec = $result->decoded($self->decodeResponse($resp));

		last if $resp->is_success;

		if($resp->code==500)
		{	$! = ENETDOWN;  # HTTP(500) -> unix error
			alert __x"Solr request failed with {code}, {retries} retries left",
			code => $resp->code, retries => $retries, result => $result;
			sleep $wait if $wait && $retries;    # let remote settle a bit
			next;
		}

		error __x"Solr request failed with: {err}",
			err => ($result->solrError || $result->httpError), result => $result;
	}

	unless($resp->is_success)
	{	$! = $resp->code==500 ? ENETDOWN : ENETUNREACH;
		fault __x"Solr request failed after {elapse} seconds after {retries} retries",
			elapse => time - $start, retries => $self->{AS_retry_max} - $retries -1, result => $result;
	}

	$resp;
}

sub decodeResponse($) { undef }

#----------------------------------

1;