view release on metacpan or  search on metacpan

lib/AnyEvent/Redis/Federated.pm  view on Meta::CPAN

	# populate node list
	$self->{nodes} = [keys %{$self->{config}->{nodes}}];

	if ($self->{debug}) {
		print "node list: ", join ', ', @{$self->{nodes}};
		print "\n";
	}

	# setup the addresses array
	foreach my $node (keys %{$self->{config}->{nodes}}) {
		if ($self->{config}->{nodes}->{$node}->{addresses}) {
			# shuffle the existing addresses array
			@{$self->{config}->{nodes}->{$node}->{addresses}} = shuffle(@{$self->{config}->{nodes}->{$node}->{addresses}});
			# and set the first to be our targeted server
			$self->{config}->{nodes}->{$node}->{address} = ${$self->{config}->{nodes}->{$node}->{addresses}}[-1];
		}
	}

	# setup the consistent hash
	my $set = Set::ConsistentHash->new;
	my @targets = map { $_, DEFAULT_WEIGHT } @{$self->{nodes}};
	$set->set_targets(@targets);
	$set->set_hash_func(\&_hash);
	$self->{set} = $set;
	$self->{buckets} = $self->{set}->buckets;

	$self->{idle_timeout} = 0 if not exists $self->{idle_timeout};

	print "config done.\n" if $self->{debug};
	bless $self, $class;

	# cache it for later use
	if ($self->{tag}) {
		$object_cache{$self->{tag}} = $self;
		weaken($object_cache{$self->{tag}});
	}

	return $self;
}

sub removeNode {
	my ($self, $node) = @_;
	$self->{set}->modify_targets($node => 0);
	$self->{buckets} = $self->{set}->buckets;
}

sub addNode {
	my ($self, $name, $ref) = @_;
	$self->{config}->{nodes}->{$name} = $ref;
	$self->{set}->modify_targets($name => DEFAULT_WEIGHT);
	$self->{buckets} = $self->{set}->buckets;
}

sub DESTROY {
}

sub _hash {
	return unpack("N", md5(shift));
}

sub commandTimeout {
	my ($self, $time) = @_;
	if (defined $time) {
		$self->{command_timeout} = $time;
	}
	return $self->{command_timeout};
}

sub queryAll {
	my ($self, $val) = @_;
	if (defined $val) {
		$self->{query_all} = $val;
	}
	return $self->{query_all};
}

sub nodeToHost {
	my ($self, $node) = @_;
	return $self->{config}->{nodes}->{$node}->{address};
}

sub keyToNode {
	my ($self, $key) = @_;
	my $node = $self->{buckets}->[_hash($key) % 1024];
	return $node;
}

sub isServerDown {
	my ($self, $server) = @_;
	return 1 if $self->{server_status}{"$server:down"};
 	return 0;
}

sub isServerUp {
	my ($self, $server) = @_;
	return 0 if $self->{server_status}{"$server:down"};
	return 1;
}

sub nextServer {
	my ($self, $server, $node) = @_;
	return $server unless $self->{config}->{nodes}->{$node}->{addresses};
	$self->{config}->{nodes}->{$node}->{address} = shift(@{$self->{config}->{nodes}->{$node}->{addresses}});
	push @{$self->{config}->{nodes}->{$node}->{addresses}}, $self->{config}->{nodes}->{$node}->{address};
	warn "redis server for $node changed from $server to $self->{config}->{nodes}->{$node}->{address} selected\n" if $self->{debug};
	return $self->{config}->{nodes}->{$node}->{address};
}

## return only on-line/up servers?

sub allServers {
	my ($self, $node) = @_;
	my $hosts = [ grep { $self->isServerUp($_) } @{$self->{config}->{nodes}->{$node}->{addresses}} ];
	return $hosts;
}

sub markServerUp {
	my ($self, $server) = @_;
	if ($self->{server_status}{"$server:down"}) {
		my $down_since = localtime($self->{server_status}{"$server:down_since"});
		delete $self->{server_status}{"$server:down"};