__PACKAGE__->attr(server => '127.0.0.1:4730');
__PACKAGE__->attr(ioloop => sub { Mojo::IOLoop->new });
__PACKAGE__->attr(error => undef);
-__PACKAGE__->attr(timeout => 300);
+__PACKAGE__->attr(timeout => 5);
__PACKAGE__->attr(encoding => 'UTF-8');
__PACKAGE__->attr(
on_error => sub {
}
);
+ warn "# using gearman server $address:$port\n";
+
return $self;
}
}
my $packet_type = {
+ TEXT => -1, # fake type for text protocol
+
CAN_DO => 1,
PRE_SLEEP => 4,
JOB_ASSIGN => 11,
WORK_COMPLETE => 13,
+ WORK_FAIL => 14,
ECHO_REQ => 16,
ECHO_RES => 17,
my $response;
my $cb = sub {
my ( $self, $data ) = @_;
- my ( $type, @data ) = $self->parse_packet($data);
- warn "# <<<< ", $nr2type->{$type}, " ",dump(@data);
+
+ if ( substr($data,0,1) ne "\x00" ) {
+ $self->res( $data );
+ $self->stop;
+ return;
+ }
+
+ my ( $type, $handle, @data ) = $self->parse_packet($data);
+ warn "# <<<< ", $nr2type->{$type}, " $handle ",dump(@data);
if ( $type == $packet_type->{JOB_CREATED} ) {
push @{ $self->{_cb_queue} }, sub {
my ( $self,$data ) = @_;
-warn "# WORK_COMPLETE ",dump $data;
my ( $type, $handle, $out ) = $self->parse_packet($data);
- die "not WORK_COMPLETE" unless $type == $packet_type->{WORK_COMPLETE};
+warn "# <<<< ",$nr2type->{$type}, " ",dump $data;
+ if ( $type == $packet_type->{WORK_COMPLETE} ) {
+ warn "WORK_COMPLETE $handle ", dump $out;
+ } elsif ( $type == $packet_type->{WORK_FAIL} ) {
+ warn "WORK_FAIL $handle ", dump $out;
+ }
$self->res( $out );
$self->stop;
};
+ $self->ioloop->timer( $self->timeout => sub {
+ my $self = shift;
+ warn "TIMEOUT $handle ", $self->timeout, "s for result";
+ $self->stop;
+ });
} elsif ( $type == $packet_type->{NO_JOB} ) {
$self->req( 'PRE_SLEEP' );
+ $self->stop;
} elsif ( $type == $packet_type->{JOB_ASSIGN} ) {
- my ( $handle, $function, $workload ) = @data;
+ my ( $function, $workload ) = @data;
my $callback = $self->{_function}->{$function};
die "no $function callback" unless ref $callback eq 'CODE';
- warn "# calling $data callback $callback";
my $out = $callback->( $workload );
- warn "# === ",dump $out;
+ warn "## $data $callback = ", dump $out;
$self->req( 'WORK_COMPLETE', $handle, $out );
$self->req( 'GRAB_JOB' );
} elsif ( $type == $packet_type->{NOOP} ) {
# $data .= "\0" if $data;
my $len = length($data);
- my $message = pack("a4NN", "\0REQ", $packet_type->{$type}, length $data ) . $data;
+ my $message = $type ne 'TEXT'
+ ? pack("a4NN", "\0REQ", $packet_type->{$type}, length $data ) . $data
+ : "$data\r\n"
+ ;
warn "# >>>> $type ",dump($message);
my $mqueue = $self->{_message_queue} ||= [];
if ( $type eq 'CAN_DO' ) {
$self->{_function}->{$data} = $callback;
+ $self->res( $callback );
warn "# installed $data callback $callback";
- $self->req( 'GRAB_JOB' );
+ } elsif ( $type =~ m/^(ECHO_REQ|SUBMIT_JOB|GRAB_JOB|TEXT)$/ ) { # sync commands
+ $self->start;
}
-
-
-# $self->start;
$self->res;
}
sub _on_read {
my ($self, $ioloop, $id, $data) = @_;
+ warn "<<<< _on_read ",dump( $id, $data );
+
+ $data = $self->{message}->{$id} .= $data;
+ my ($magic, $type, $len) = unpack( "a4NN", $data );
+
+ if ( substr($data,0,1) ne "\x00" ) {
+ return if $data !~ s/\.[\n\r]$//s;
+ } elsif ( length $data < $len ) {
+ warn "# _on_read incomplete message";
+ return;
+ }
+
+ warn "#### data ",dump($data);
+
my $cb = shift @{$self->{_cb_queue}};
if ($cb) {
- Mojo::Util::decode($self->encoding, $data) if $data;
+# Mojo::Util::decode($self->encoding, $data) if $data; # FIXME
warn "# on read callback with ", dump($data);
$cb->($self, $data);
} else {
warn "no callback";
}
+ delete $self->{message}->{$id};
+
# Reset error after callback dispatching
$self->error(undef);
}