view release on metacpan or  search on metacpan

lib/App/Context/ClusterNode.pm  view on Meta::CPAN

                method       => $3,
            );
            my $args = $4;
            $event{args} = $self->{rpc_serializer}->deserialize($args) if ($args ne "");
            my $event_token = $self->send_async_event({method => "process_async_event", args => [\%event],});
            $event{event_token} = $event_token;
            $return_value = "ASYNC-EVENT-TOKEN:$event_token\n";
        }
        elsif ($msg =~ /^CONTROLLER-UP:/) {
            my $controller_host = $self->{controller_host};
            my $controller_port = $self->{controller_port};
            my $node_host       = $self->{host};
            my $node_port       = $self->{port};
            $self->send_async_event({
                method => "send_async_message",
                args => [ $controller_host, $controller_port, "NODE-UP:$node_host:$node_port" ],
            });
            $return_value = "OK";
        }
        elsif ($msg =~ /^ABORT-ASYNC-EVENT:(.*)/) {
            my $event_token = $1;
            $self->abort_async_event($event_token);
            $return_value = "OK";
        }
        else {
            $return_value = "ERROR: unknown [$msg]";
        }
    }
    &App::sub_exit($return_value) if ($App::trace);
    return($return_value);
}

# Can be overridden to provide customized processing.
sub process_custom_msg {
    &App::sub_entry if ($App::trace);
    my ($self, $msg) = @_;
    my $return_value = "";
    &App::sub_exit($return_value) if ($App::trace);
    return($return_value);
}

sub state {
    &App::sub_entry if ($App::trace);
    my ($self) = @_;

    my $datetime = time2str("%Y-%m-%d %H:%M:%S", time());
    my $state = "Cluster Node: $self->{host}:$self->{port}  procs[$self->{num_procs}/$self->{max_procs}:max]  async_events[$self->{num_async_events}/$self->{max_async_events}:max]\n[$datetime]\n";
    $state .= "\n";
    $state .= $self->_state();

    &App::sub_exit($state) if ($App::trace);
    return($state);
}

sub _state {
    &App::sub_entry if ($App::trace);
    my ($self) = @_;

    my $state = "";

    $state .= $self->SUPER::_state();

    &App::sub_exit($state) if ($App::trace);
    return($state);
}

sub process_async_event {
    &App::sub_entry if ($App::trace);
    my ($self, $event) = @_;
    my ($results);
    eval {
        $results = $self->send_event($event);
    };
    if ($@) {
        $results = $@;
    }
    $results =~ s/\s*\n\s*/ /gs;  # we can't have multi-line result text
    my $results_txt = $self->{rpc_serializer}->serialize($results);
    my $msg = "ASYNC-EVENT-RESULTS:$event->{event_token}:$results_txt";
    $self->send_message($self->{controller_host}, $self->{controller_port}, $msg);
    &App::sub_exit() if ($App::trace);
}

sub finish_killed_async_event {
    &App::sub_entry if ($App::trace);
    my ($self, $event) = @_;
    my $results_txt = $self->{rpc_serializer}->serialize("Subrequest Killed");
    my $msg = "ASYNC-EVENT-RESULTS:$event->{event_token}:$results_txt";
    $self->send_async_message($self->{controller_host}, $self->{controller_port}, $msg);
    &App::sub_exit() if ($App::trace);
}

1;