view release on metacpan or  search on metacpan

lib/Argon/Client.pm  view on Meta::CPAN

    ->Declare(1)
    ->Out;

  $self->queue('Argon::Task', [$code, $args], $cb);
}


sub async ($\[&$]\@) {
  my ($self, $code_ref, $args) = @_;
  my $cv = AnyEvent->condvar;
  $self->process($code_ref, $args, $cv);
  tie my $async, 'Argon::Async', $cv;
  return $async;
}

sub cleanup {
  my $self = shift;
  $self->closed->();
  $self->channel(undef);

  my $error = 'Remote host was disconnected before task completed';

  foreach my $id ($self->msg_ids) {
    my $info = $self->get_msg($id);
    my $cb   = $info->{cb} or next;
    my $msg  = $info->{orig};
    $cb->($msg->error($error));
  }
}

sub _ready { shift->ready->() }

sub _error {
  my ($self, $error) = @_;
  log_error '[%s] %s', $self->addr, $error;
  $self->cleanup;
}

sub _close {
  my ($self) = @_;
  log_debug '[%s] Remote host disconnected', $self->addr;
  $self->cleanup;
}

sub _notify {
  my ($self, $msg) = @_;

  if ($self->has_msg($msg->id)) {
    my $info = $self->del_msg($msg->id);

    if ($msg->denied && $info->{retry}) {
      my $copy  = $info->{orig}->copy;
      my $intvl = $info->{intvl}->();
      log_debug 'Retrying message in %0.2fs: %s', $intvl, $info->{orig}->explain;

      $self->add_msg($copy->id, {
        orig  => $copy,
        cb    => $info->{cb},
        intvl => $info->{intvl},
        retry => 1,
        timer => AnyEvent->timer(after => $intvl, cb => K('send', $self, $copy)),
      });

      return;
    }

    if ($info->{cb}) {
      $info->{cb}->($msg);
    }
    else {
      $self->notify->($msg);
    }
  }
  else {
    $self->notify->($msg);
  }
}

__PACKAGE__->meta->make_immutable;

1;

__END__

=pod

=encoding UTF-8

=head1 NAME

Argon::Client - Client-side connection class for Argon systems

=head1 VERSION

version 0.18

=head1 SYNOPSIS

  use Argon::Client;
  use AnyEvent;

  my $cv = AnyEvent->condvar;

  my $ar = Argon::Client->new(
    host   => 'some.host.net',
    port   => 1234,
    retry  => 1,
    opened => $cv,
    ready  => sub{},
    failed => sub{},
    closed => sub{},
    notify => sub{},
  );

  $cv->recv;

  while (my $task = get_next_task) {
    $ar->process($task->class, $task->args, \&task_complete);
  }

  my $result = $ar->async(sub{ ... });
  if ($result eq 'fnord') {
    ...
  }

=head1 DESCRIPTION

Provides the client connection to an L<Argon> network.

=head1 ATTRIBUTES

=head2 host

The hostname of the L<Argon::Manager> serving as the entry point for the
Argon network.

=head2 port

The port number for the L<Argon::Manager>.

=head2 retry

By default, when the network is at capacity, new tasks may be rejected, causing
L<Argon::Message/result> to croak. If C<retry> is set, the C<Argon::Client>
will instead retry the task on a logarithmic backoff timer until the task is
accepted by the manager.

=head2 opened

A code ref that is triggered when the connection is initially opened.

=head2 ready

A code ref that is triggered when the connection has been opened and the
client is ready to begin sending tasks.

=head2 failed

A code ref that is triggered when the connection fails. The value of C<$!> is
passed as an argument.

=head2 closed

A code ref that is triggered when the connection to the remote host is
closed.

=head2 notify

When tasks are created without a callback (see L<Argon::Client/process>),
the C<notify> callback is used in its place. The L<Argon::Message> reply
is passed as an argument.

=head1 METHODS

=head2 ping

Pings the L<Argon::Manager> and calls the supplied callback with the manager's
reply.

  $ar->ping(sub{ my $reply = shift; ... });

=head2 queue

Queues a task with the Ar manager. Accepts the name of a class accessible to
the workers defining a C<new> and C<run> method, an array of arguments to be
passed to C<new>, and an optional code ref to be called when the task is
complete. If not supplied, the L<Argon::Client/notify> method will be called in
its place.

  $ar->queue('Task::Class', $args_list, sub{
    my $reply = shift;
    ...
  });

=head2 process

If the Ar workers were started with C<--allow-eval> and if the client process
itself has C<$Argon::ALLOW_EVAL> set to a true value, a code ref may be passed
in place of a task class. The code ref will be serialized using
L<Data::Dump::Streamer> and has limited support for closures.

  $ar->process(sub{ ... }, $args_list, sub{
    my $reply = shift;
    ...
  });