AnyEvent::STOMP::Client + AnyEvent::ForkManger = Временная ошибка

Question

Я пытаюсь написать процесс, который прослушивает ActiveMQ и, основываясь на сообщении, выходит и извлекает данные из веб-сервиса, выполняет некоторую обработку и затем передает данные процесса в другой веб-сервис. (REST/JSON)

Приведенный ниже модуль работает нормально, пока одно из вонких веб-сервисов, с которыми я общаюсь, не вернет ошибку. Я перепробовал много вещей, чтобы поймать ошибку, но пока безрезультатно. Как только происходит ошибка веб-сервиса, я получаю следующее сообщение:

необработанное исключение обратного вызова для события (MESSAGE, AnyEvent::STOMP::Client=HASH(0x3ad5e48), HASH(0x3a6bbb0) {"action":"созданный","data":{"id":40578737,"type":"alert","who":null},"guid":"ADCCEE0C-73A7-11E6-8084-74B346D1CA67","hostname":"myserver","pid":48632}): $fork_manager->start() should быть вызванным в процессе менеджера

Хорошо, я концептуально понимаю, что дочерний процесс пытается запустить другой процесс, и что менеджер ветвлений говорит, что нет. Но, учитывая модуль ниже, как правильно запустить новый процесс для обработки длительной обработки. Или почему дочерний процесс умирает, вызывая это исключение, и как я могу предотвратить это

Вот модуль (урезанный)

package consumer;

use AnyEvent::ForkManager;
use AnyEvent::STOMP::Client;
use JSON;
use Data::Dumper;
use v5.18;
use Moose;

sub run {
    my $self    = shift;
    my $pm      = AnyEvent::ForkManager->new(max_workers => 20);
    my $stomp   = AnyEvent::STOMP::Client->new();

    $stomp->connect();
    $stomp->on_connected(sub {
        my $stomp   = shift;
        $stomp->subscribe('/topic/test');
        say "Connected to STOMP";
    });

    $pm->on_start(sub {
        my ($pm,$pid,@params) = @_;
        say "Starting $pid worker";
    });

    $pm->on_finish(sub {
        my ($pm, $pid,@params) = @_;
        say "Finished $pid worker";
    });

    $pm->on_error(sub {
        say Dumper(\@_);
    });

    $stomp->on_message(sub {
        my ($stomp, $header, $body) = @_;
        my $href    = decode_json $body;
        $pm->start(cb => sub {
            my ($pm, @params) = @_;
            $self->process(@params);
        },
        args    => [ $href->{id}, $href->{data}->{type}, $href->{data}->{who} ],
        );
    });

    my $cv = AnyEvent->condvar;
    $cv->recv;
}

sub process {
    say "Processing ".Dumper(\@_);
    sleep 5;
    if ( int(rand(10)) < 5 ) {
        die "OOPS";  # this triggers the error message above
    }
    say "Done Processing $_[1]";
}

1;

Вот драйвер для модуля выше:

#!/usr/bin/env perl

use v5.18;
use lib '.';
use consumer;

my $c   = consumer->new();
$c->run;

Наконец, генератор трафика, который вы можете использовать, чтобы увидеть это в действии:

#!/usr/bin/env perl

use lib '../lib';
use lib '../../lib';
use v5.18;

use Data::Dumper;
use JSON;
use Net::STOMP::Client;

$ENV{'scot_mode'} = "testing";

my $stomp = Net::STOMP::Client->new(
    host    => "127.0.0.1",
    port    => 61613
);
$stomp->connect();

for (my $i = 1; $i < 1000000; $i++) {
    my $href    = {
        id      => $i,
        type    => "event",
        what    => "foo",
    };
    my $json    = encode_json $href;
    say "Sending ".Dumper($href);
    $stomp->send(
        destination => "/topic/test",
        body        => $json,
    );
}

$stomp->disconnect();

Answer 1

Я смог решить эту проблему с помощью Try::Catch и обернуть вызов self->process с помощью try catch:

$stomp->on_message(sub {
        my ($stomp, $header, $body) = @_;
        my $href    = decode_json $body;
        $pm->start(cb => sub {
            my ($pm, @params) = @_;
            try {
                $self->process(@params);
            }
            catch {
                # error handling stuff
            };
        },
        args    => [ $href->{id}, $href->{data}->{type}, $href->{data}->{who} ],
        );
    }
);