abonbot/lib/rabbit_async_rec.pm

#!/usr/bin/perl

# Реконнект для rabbit_async
# Юрий Жиловец, 06.04.2018

package rabbit_async_rec;

use Modern::Perl;
use Mojo::IOLoop;

use rabbit_async;

use constant SCALE => 2;
use constant MIN_TIMEOUT => 1;
use constant MAX_TIMEOUT => 30;

sub new
{
  my $class = shift;
  my $arg = shift;
  my $callback = shift;
  
  my $self = bless {
    timeout => 0,
    arg => $arg,
    callback => $callback,
    delayed => [],
  };
  
  $arg->{on_close} = sub { $self->_reconnect_later };
  $arg->{on_error} = sub { say STDERR "RabbitMQ error: " . shift };
  $arg->{on_connect_error} = sub { 
    unless ($self->{timeout})
    {
      # если ошибка произошла при старте, это фатально
      say STDERR "RabbitMQ connect error: ", @_;
      exit();
    }
    else
    {
      say STDERR "RabbitMQ error: ",@_;
      $self->_reconnect_later;
    }
  };
  
  $self->_connect;
            
  return $self;
}

sub _reconnect_later
{
  my $self = shift;
  
  # событие запросто может придти второй раз на соединении, которое теоретически уже закрылось
  $self->{rabbit}->{on_close} = $self->{rabbit}->{on_error} = sub {};
  $self->{rabbit} = undef;
  
  $self->{timeout} = $self->_calc_timeout;
  say STDERR "rabbit_async_rec: will try to reconnect in $self->{timeout} sec...";
    
  Mojo::IOLoop->timer($self->{timeout} => sub { $self->_connect });    
}

sub _connect
{
  my $self = shift;

  say STDERR "rabit_async_rec: connecting to ", $self->{arg}->{host};
  
  my $rabbit;
  
  my $cb = sub
  {
    $self->{rabbit} = $rabbit;

    foreach (@{ $self->{delayed} })
    {
      my $method = $_->{method};
      $self->{rabbit}->$method(@{ $_->{args} });
    }
  
    $self->{delayed} = [];
    
    $self->{callback}->();
  };
  
  $rabbit = new rabbit_async($self->{arg}, $cb);
}

sub _calc_timeout
{
  my $self = shift;
  
  my $timeout = MIN_TIMEOUT;
  if ($self->{timeout})
  {
    $timeout = $self->{timeout} * SCALE;
    $timeout = MAX_TIMEOUT if $timeout > MAX_TIMEOUT;
  }

  return $timeout;  
}

sub AUTOLOAD
{
  my $self = shift or return undef;
  my @args = @_;

  ( my $method = $rabbit_async_rec::AUTOLOAD ) =~ s{.*::}{};
  
  if ($self->{rabbit})
  {
    # делегируем запросы настоящему модулю
    $self->{rabbit}->$method(@args);
  }
  else
  {
    # запоминаем запрос для последующего выполнения
    push @{ $self->{delayed} }, { method=>$method, args=>\@args };
  }
}

1;