File Coverage

blib/lib/jQluster/Server.pm

Criterion	Covered	Total	%
statement	55	63	87.3
branch	12	18	66.6
condition	3	5	60.0
subroutine	12	13	92.3
pod	4	4	100.0
total	86	103	83.5

package jQluster::Server;
use 5.10.0;
use strict;
use warnings;
use Carp;
use Data::UUID;

our $VERSION = "0.03";

sub new {
    my ($class, %args) = @_;
    my $self = bless {
        id_generator => Data::UUID->new,
        registry => {},
        uids_for_remote_node_id => {},
        logger => $args{logger} // sub { },
    }, $class;
    return $self;
}

sub _generate_message_id {
    my ($self) = @_;
    return $self->{id_generator}->create_str();
}

sub _log {
    my ($self, $level, $msg) = @_;
    $self->{logger}->($level, $msg);
}

sub register {
    my ($self, %args) = @_;
    foreach my $key (qw(unique_id message sender)) {
        croak "$key parameter is mandatory" if not defined $args{$key};
    }
    foreach my $msg_key (qw(message_id from message_type)) {
        if(!defined($args{message}{$msg_key})) {
            croak "The register message does not have $msg_key field. Something is wrong.";
        }
    }
    if($args{message}{message_type} ne "register") {
        croak "Message type is $args{message}{message_type}, not 'register'. Something is wrong.";
    }
    my %reg_entry = (
        unique_id => $args{unique_id},
        sender => $args{sender},
        remote_node_id => $args{message}{from}
    );
    if(exists $self->{registry}{$reg_entry{unique_id}}) {
        croak "Duplicate registration for unique ID: $reg_entry{unique_id}";
    }
    $self->{registry}{$reg_entry{unique_id}} = \%reg_entry;
    $self->{uids_for_remote_node_id}{$reg_entry{remote_node_id}}{$reg_entry{unique_id}} = 1;
    $self->_log(info => "Accept registration: unique_id = $reg_entry{unique_id}, remote_node_id = $reg_entry{remote_node_id}");

    $self->distribute({
        message_id => $self->_generate_message_id(),
        message_type => "register_reply",
        from => undef, to => $reg_entry{remote_node_id},
        body => { error => undef, in_reply_to => $args{message}{message_id} }
    });
}

sub unregister {
    my ($self, $unique_id) = @_;
    my $entry = delete $self->{registry}{$unique_id};
    return if !defined($entry);
    delete $self->{uids_for_remote_node_id}{$entry->{remote_node_id}}{$entry->{unique_id}};
    $self->_log(info => "Unregister: unique_id = $unique_id, remote_node_id = $entry->{remote_node_id}");
}

my %REPLY_MESSAGE_TYPE_FOR = (
    select_and_get => "select_and_get_reply",
    select_and_listen => "select_and_listen_reply"
);

sub _try_reply_error_to {
    my ($self, $orig_message, $error) = @_;
    my $reply_message_type = $REPLY_MESSAGE_TYPE_FOR{$orig_message->{message_type}};
    if(!defined($reply_message_type)) {
        $self->_log("error", "Unknown message type: $orig_message->{message_type}: cannot reply to it.");
        return;
    }
    $self->distribute({
        message_id => $self->_generate_message_id(),
        message_type => $reply_message_type,
        from => undef, to => $orig_message->{from},
        body => { error => $error, in_reply_to => $orig_message->{message_id} }
    });
}

sub distribute {
    my ($self, $message) = @_;
    my $to = $message->{to};
    if(!defined($to)) {
        return;
    }
    my $uid_map = $self->{uids_for_remote_node_id}{$to};
    if(!defined($uid_map) || !%$uid_map) {
        $self->_try_reply_error_to($message, "Target remote node ($to) does not exist.");
        return;
    }
    foreach my $uid (keys %$uid_map) {
        my $entry = $self->{registry}{$uid};
        if(!defined($entry)) {
            $self->_log("error", "UID registry has a key for $uid, but it does not map to an entry object. Something is wrong.");
            next;
        }
        $entry->{sender}->($message);
    }
}

1;

__END__