File Coverage

blib/lib/Search/Elasticsearch/CxnPool/Async/Simple/Sniff.pm

Criterion	Covered	Total	%
statement	12	67	17.9
branch	0	20	0.0
condition			n/a
subroutine	4	13	30.7
pod	0	3	0.0
total	16	103	15.5

package Search::Elasticsearch::CxnPool::Async::Simple::Sniff;

use Moo;

use Search::Elasticsearch::Role::Is_Async::Loader ();
use Search::Elasticsearch::Util qw(new_error);

use namespace::clean;

with 'Search::Elasticsearch::Role::CxnPool::Sniff',
     'Search::Elasticsearch::Role::Is_Async';


sub next_cxn {
	my ($self, $cb) = @_;

	if ($self->next_sniff <= time()) {
		$self->sniff(sub {
			$self->_next_cxn($cb);
		});
	}
	else {
		$self->_next_cxn($cb);
	}

	return;
}

sub _next_cxn {
	my ($self, $cb) = @_;

	my $cxns = $self->cxns;
	my $cnt  = @$cxns;

	while (0 < $cnt--) {
		my $cxn = $cxns->[$self->next_cxn_num];

		return $cb->($cxn) if $cxn->is_live;
	}

	local $@ = new_error('NoNodes', 'No nodes are available: [' . $self->cxns_seeds_str . ']');

	$cb->();
}

sub sniff {
	my ($self, $cb) = @_;

	my $cxns = $self->cxns;
	my (%seen, @skip);

	my $sub; $sub = sub {
		my $cnt = @$cxns;
		my $fnd;

		if ($cnt > keys(%seen)) {
			while ($cnt--) {
				my $cxn = $cxns->[$self->next_cxn_num];

				next if $seen{$cxn}++;

				if ($cxn->is_dead) {
					push(@skip, $cxn);
				}
				else {
					return $self->sniff_cxn($cxn, sub {
						return $cb->() if $_[0];
						$cxn->mark_dead();
						$sub->();
					});
				}
			}
		}

		if (my $cxn = shift(@skip)) {
			return $self->sniff_cxn($cxn, sub {
				return $cb->() if $_[0];
				$sub->();
			});
		}

		$self->logger->infof('No live nodes available. Trying seed nodes.');
		$self->_sniff_seed_nodes($cb);
	};

	$sub->();

	return;
}

sub _sniff_seed_nodes {
	my ($self, $cb) = @_;

	my $idx = 0;

	my $sub; $sub = sub {
		my $nods = $self->seed_nodes;

		return $cb->() if $idx >= @$nods;

		my $cxn = $self->cxn_factory->new_cxn($nods->[$idx++]);

		$self->sniff_cxn($cxn, sub {
			return $cb->() if $_[0];
			$sub->();
		});
	};

	$sub->();

	return;
}

sub sniff_cxn {
	my ($self, $cxn, $cb) = @_;

	$cxn->sniff(sub {
		$cb->($self->parse_sniff($cxn->protocol, $_[0]));
	});

	return;
}


1;


__END__