view release on metacpan or  search on metacpan

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

            unshift(@$events, $event);
            if ($event->{time} && $event->{interval}) {
                $event->{time} += $event->{interval}; # reschedule the event
                $self->log({level=>5},"Event Rescheduled: time=$time [$event->{time}, every $event->{interval}] $event->{method}().\n");
                if ($time_of_next_event == 0 || $event->{time} < $time_of_next_event) {
                    $time_of_next_event = $event->{time};
                }
            }
            else {
                $self->log({level=>5},"Event Removed: time=$time [$event->{time}, every $event->{interval}] $event->{method}().\n");
                splice(@$scheduled_events, $i, 1); # remove the (one-time) event
                $event->{scheduled} = 0;
            }
        }
        else {
            if ($time_of_next_event == 0 || $event->{time} < $time_of_next_event) {
                $time_of_next_event = $event->{time};
            }
        }
    }
    &App::sub_exit($time_of_next_event) if ($App::trace);
    return($time_of_next_event);
}

# NOTE: send_event() is similar to call(). I ought to resolve this.
sub send_event {
    &App::sub_entry if ($App::trace);
    my ($self, $event) = @_;
    my $method = $event->{method};
    my @args = $event->{args} ? @{$event->{args}} : ();
    my $name = $event->{name};
    my $service_type = $event->{service_type};
    $service_type = "SessionObject" if (!$service_type && $name);
    my (@results);
    if ($name) {
        my $service = $self->service($service_type, $name);
        $self->log({level=>3},"Send Event: $service_type($name).$method(@args)\n");
        @results = $service->$method(@args);
    }
    else {
        $self->log({level=>3},"Send Event: $method(@args)\n");
        @results = $self->$method(@args);
    }
    &App::sub_exit(@results) if ($App::trace);
    if (wantarray()) {
        return(@results);
    }
    else {
        if ($#results == -1) {
            return(undef);
        }
        elsif ($#results == 0) {
            return($results[0]);
        }
        else {
            return(\@results);
        }
    }
}

# NOTE: The baseline context implements the API for asynchronous events
#       in a simplistic, sequential way.
#       It merely sends the event, then sends the callback event.
#       See App::Context::Server for a context that spawns processes which
#       execute the event.  When the process exits, the callback_event is fired.
#       See App::Context::Cluster for a context that sends a message to an
#       available cluster node for executing.  When the node reports back that
#       it has completed the task, the callback_event is fired.

sub send_async_event {
    &App::sub_entry if ($App::trace);
    my ($self, $event, $callback_event) = @_;
    my $event_token = $self->send_async_event_in_process($event, $callback_event);
    &App::sub_exit($event_token) if ($App::trace);
    return($event_token);
}

sub send_async_event_in_process {
    &App::sub_entry if ($App::trace);
    my ($self, $event, $callback_event) = @_;
    my $errnum = 0;
    my $errmsg = "";
    my $event_token = "local-$$";
    my ($returnval);
    eval {
        $returnval = $self->send_event($event);
    };
    if ($@) {
        $errmsg = $@;
        $errnum = 1;
        $self->log("ERROR: send_async_event_now() $event->{name}.$event->{method} : $errmsg\n");
    }
    if ($callback_event) {
        $callback_event->{args} = [] if (! $callback_event->{args});
        push(@{$callback_event->{args}}, {event_token => $event_token, returnval => $returnval, errnum => $errnum, errmsg => $errmsg});
        $self->send_event($callback_event);
    }
    &App::sub_exit($event_token) if ($App::trace);
    return($event_token);
}

=head2 wait_for_event()

    * Signature: $self->wait_for_event($event_token)
    * Param:     $event_token     string
    * Return:    void
    * Throws:    App::Exception
    * Since:     0.01

    Sample Usage: 

    $self->wait_for_event($event_token);

The wait_for_event() method is called when an asynchronous event has been
sent and no more processing can be completed before it is done.

=cut

sub wait_for_event {
    &App::sub_entry if ($App::trace);
    my ($self, $event_token) = @_;
    &App::sub_exit() if ($App::trace);
}

# NOTE: This send_message() and send_async_message() can be on the App::Context
#       class to allow a program in any context to send this kind of message.
#       (The only downside is a dependency on IO::Socket::INET.)
sub send_async_message {
    &App::sub_entry if ($App::trace);
    my ($self, $host, $port, $message, $await_return_value, $timeout, $server_close) = @_;
    my $pid = $self->fork();
    if (!$pid) {   # running in child
        $self->send_message($host, $port, $message, $await_return_value, $timeout, $server_close);
        $self->exit(0);
    }
    &App::sub_exit() if ($App::trace);
}

# NOTE: $messages that start with "RV-" wait for a return value.
#       $messages that start with "SC-" force the server to close the socket first
#       This is to help manage which system has the sockets lingering in TIME_WAIT state.
# Here is the truth table for $await_return_value, $server_close
#       $await_return_value  $server_close =         client         +        server     
#       -------------------  -------------   ----------------------   ---------------------
#                 0                0              write/close              read/close
#                 0                1            write/read/close           read/close
#                 1                0         write/read/write/close   read/write/read/close
#                 1                1            write/read/close         read/write/close
sub send_message {
    &App::sub_entry if ($App::trace);
    my ($self, $host, $port, $message, $await_return_value, $timeout, $server_close) = @_;
    my $verbose = $self->{verbose};

    if (!$port && $host =~ /^([^:]+):([0-9]+)$/) {
        $host = $1;
        $port = $2;
    }

    my $send_socket = IO::Socket::INET->new(
        PeerAddr  => $host,
        PeerPort  => $port,
        Proto     => "tcp",
        Type      => SOCK_STREAM,
        ReuseAddr => 1,
    );
    my ($send_fd);
    $send_fd = fileno($send_socket) if ($send_socket);
    $self->log({level=>3},"($send_fd) send_message($host, $port, $message)\n");

    my $response = "";
    my $rv = $await_return_value ? "RV-" : "";
    my $sc = $server_close ? "SC-" : "";
    if ($send_socket) {
        eval {