AnyEvent-Gearman

 view release on metacpan or  search on metacpan

lib/AnyEvent/Gearman/Client.pm  view on Meta::CPAN

You should set at least one job_server.

=item prefix => 'Str',

Sets the namespace / prefix for the function names. This is useful for sharing job servers between different applications or different instances of the same application (different development sandboxes for example).

The namespace is currently implemented as a simple tab separated concatenation of the prefix and the function name.

=back

=head2 add_task($function, $workload, %callbacks)

Start new job and wait results in C<%callbacks>

    $gearman->add_task(
        $function => $workload,
        on_complete => sub {
            my $result = $_[1],
        },
        on_fail => sub {
            # job failled
        },
    );

C<$function> is a worker function name, and C<$workload> is a data that will be passed to worker.

C<%callbacks> is set of callbacks called by job events. Available callbacks are:

=over 4

=item on_complete => $cb->($self, $result)

Called when the job is completed. C<$result> is some results data which is set by C<< $job->complete($result) >> in worker.

=item on_fail => $cb->($self, $reason)

Called when the job is failed. C<$reason> is empty if its threw by worker. I don't know why but gearman spec say so. Considering to use C<on_warning> below for some failing notify.

lib/AnyEvent/Gearman/Client.pm  view on Meta::CPAN

=item on_created => $cb->($self)

Called when the servers reports that the task was created successfully.
Updates the Task object with the server assigned C<job_handle>.

=back

You should to set C<on_complete> and C<on_fail> at least.


=head2 add_task_bg($function, $workload, %callbacks)

Starts a new background job. The parameters are the same as
L<add_task($function, $workload, %callbacks)|add_task()>, but the only
callback that is called is C<on_created>.

    $gearman->add_task_bg(
        $function => $workload,
        on_created => sub {
            my ($task) = @_;
        },
    );

lib/AnyEvent/Gearman/Connection.pm  view on Meta::CPAN

    isa      => 'Object',
    weak_ref => 1,
);

has handler => (
    is      => 'rw',
    isa     => 'Maybe[AnyEvent::Handle]',
    clearer => 'clear_handler',
);

has on_connect_callbacks => (
    is      => 'rw',
    isa     => 'ArrayRef',
    default => sub { [] },
);

has dead_time => (
    is      => 'rw',
    isa     => 'Int',
    default => 0,
);

lib/AnyEvent/Gearman/Connection.pm  view on Meta::CPAN

                    my @undone = @{ $self->_need_handle },
                                 values %{ $self->_job_handles };
                    $_->event('on_fail') for @undone;

                    $self->_need_handle([]);
                    $self->_job_handles({});
                    $self->mark_dead;
                },
            );
            $self->handler( $handle );
            $_->() for map { $_->[0] } @{ $self->on_connect_callbacks };
        }
        else {
            warn sprintf("Connection failed: %s", $!);
            $self->mark_dead;
            $_->() for map { $_->[1] } @{ $self->on_connect_callbacks };
        }

        $self->on_connect_callbacks( [] );
    };

    weaken $self;
    $self->_con_guard($g);

    $self;
}

sub connected {
    !!shift->handler;
}

sub add_on_ready {
    my ($self, $cb, $eb) = @_;

    if ($self->connected) {
        $cb->();
    }
    else {
        push @{ $self->on_connect_callbacks }, [ $cb, $eb ];
        $self->connect;
    }
}

sub mark_dead {
    my ($self) = @_;
    $self->dead_time( time + 10 );
    $self->clear_handler;
}

t/02_client_worker.t  view on Meta::CPAN

        
        on_created  => sub { $cbs{on_created}++  },
        on_data     => sub { $cbs{on_data}++     },
        on_status   => sub { $cbs{on_status}++   },
        on_warning  => sub { $cbs{on_warning}++  },
        on_complete => sub { $cbs{on_complete}++ },
        on_fail     => sub { $cbs{on_fail}++     },
    );
    
    is $cv->recv, 'bg job done: pick me!';
    cmp_deeply(\%cbs, { on_created => 1 }, 'proper set of callbacks executed');
}

my $child = fork;
if (!defined $child) {
    die "fork failed: $!";
}
elsif ($child == 0) {
    my $server = Gearman::Server->new( port => $port );
    Danga::Socket->EventLoop;
    exit;