view release on metacpan or  search on metacpan

lib/AWS/Lambda/Bootstrap.pm  view on Meta::CPAN

package AWS::Lambda::Bootstrap;
use 5.026000;
use utf8;
use strict;
use warnings;
use HTTP::Tiny;
use JSON::XS qw/decode_json encode_json/;
use Try::Tiny;
use AWS::Lambda;
use AWS::Lambda::Context;
use AWS::Lambda::ResponseWriter;
use Scalar::Util qw(blessed);
use Exporter 'import';

our @EXPORT = ('bootstrap');

sub bootstrap {
    my $handler = shift;
    my $bootstrap = AWS::Lambda::Bootstrap->new(
        handler => $handler,
    );
    $bootstrap->handle_events;
}

sub new {
    my $proto = shift;
    my $class = ref $proto || $proto;
    my %args;
    if (@_ == 1 && ref $_[0] eq 'HASH') {
        %args = %{$_[0]};
    } else {
        %args = @_;
    }

    my $api_version = '2018-06-01';
    my $env_handler = $args{handler} // $ENV{'_HANDLER'} // die '$_HANDLER is not found';
    my ($handler, $function) = split(/[.]/, $env_handler, 2);
    my $runtime_api = $args{runtime_api} // $ENV{'AWS_LAMBDA_RUNTIME_API'} // die '$AWS_LAMBDA_RUNTIME_API is not found';
    my $task_root = $args{task_root} // $ENV{'LAMBDA_TASK_ROOT'} // die '$LAMBDA_TASK_ROOT is not found';
    my $max_workers = $args{max_workers} // $ENV{'AWS_LAMBDA_MAX_CONCURRENCY'} // 0;
    die "max_workers must be a non-negative integer, got: $max_workers" 
        unless $max_workers =~ /^\d+$/ && $max_workers >= 0;
    my $self = bless +{
        task_root      => $task_root,
        handler        => $handler,
        function_name  => $function,
        runtime_api    => $runtime_api,
        api_version    => $api_version,
        next_event_url => "http://${runtime_api}/${api_version}/runtime/invocation/next",
        max_workers    => $max_workers,
        http           => HTTP::Tiny->new(
            # XXX: I want to disable timeout, but it seems HTTP::Tiny does not support it.
            # So, I set a long timeout.
            timeout => 365*24*60*60, # 365 days
        ),
    }, $class;
    return $self;
}

sub handle_events {
    my $self = shift;
    $self->_init or return;

    if ($self->{max_workers} > 0) {
        require Parallel::Prefork;
        my $pm = Parallel::Prefork->new({
            max_workers => $self->{max_workers},
            trap_signals => {
                TERM => 'TERM',
                HUP  => 'TERM',
            },
        });
        while ($pm->signal_received ne 'TERM') {
            $pm->start and next;
            $self->_handle_events;
            $pm->finish;
        }
        $pm->wait_all_children;
    } else {
        $self->_handle_events
    }
}

sub _handle_events {
    my $self = shift;
    while(1) {
        $self->handle_event;
    }
}

sub _init {
    my $self = shift;
    if (my $func = $self->{function}) {
        return $func;
    }

    my $task_root = $self->{task_root};
    my $handler = $self->{handler};
    my $name = $self->{function_name};
    return try {
        package main;
        require "${task_root}/${handler}.pl";
        my $f = main->can($name) // die "handler $name is not found";
        $self->{function} = $f;
    } catch {
        $self->lambda_init_error($_);
        $self->{function} = sub {};
        undef;
    };
}

sub handle_event {
    my $self = shift;