Skip to content
This repository

HTTPS clone URL

Subversion checkout URL

You can clone with HTTPS or Subversion.

Download ZIP
Browse code

add (de)serialize settings.

  • Loading branch information...
commit 7899641eed611c211cb7fb508401e03672f8a52a 1 parent ac558ee
Atsushi Kobayashi authored

Showing 3 changed files with 148 additions and 9 deletions. Show diff stats Hide diff stats

  1. +40 8 lib/Jonk.pm
  2. +5 1 lib/Jonk/Job.pm
  3. +103 0 t/serialize.t
48 lib/Jonk.pm
@@ -23,6 +23,7 @@ sub new {
23 23 table_name => $table_name,
24 24 functions => $functions,
25 25 driver => $driver,
  26 + has_func => scalar(keys %{$functions}) ? 1 : 0,
26 27
27 28 _errstr => undef,
28 29
@@ -68,7 +69,9 @@ sub _parse_functions {
68 69 if (not defined $functions->[$i+1]) {$i++ }
69 70 elsif (ref $functions->[$i+1]) {$value = $functions->[++$i]}
70 71
71   - $value ||= +{grab_for => $default_grab_for};
  72 + $value->{grab_for} ||= $default_grab_for;
  73 + $value->{serializer} ||= ($opts->{default_serializer} || sub {$_[0]});
  74 + $value->{deserializer} ||= ($opts->{default_deserializer} || sub {$_[0]});
72 75
73 76 $funcs->{$func} = $value;
74 77 }
@@ -102,9 +105,10 @@ sub insert {
102 105 local $self->{dbh}->{RaiseError} = 1;
103 106 local $self->{dbh}->{PrintError} = 0;
104 107
  108 + my $serializer = $self->{functions}->{$func}->{serializer} || sub {$_[0]};
105 109 my $sth = $self->{dbh}->prepare_cached($self->{insert_query});
106 110 $sth->bind_param(1, $func);
107   - $sth->bind_param(2, $arg, _bind_param_attr($self->{driver}));
  111 + $sth->bind_param(2, $serializer->($arg), _bind_param_attr($self->{driver}));
108 112 $sth->bind_param(3, $self->{insert_time_callback}->());
109 113 $sth->bind_param(4, $opt->{run_after}||0);
110 114 $sth->bind_param(5, $opt->{priority} ||0);
@@ -192,7 +196,7 @@ sub lookup_job {
192 196 sub find_job {
193 197 my ($self, $opts) = @_;
194 198
195   - unless (keys %{$self->{functions}}) {
  199 + unless ($self->{has_func}) {
196 200 Carp::croak('missin find_job functions.');
197 201 }
198 202
@@ -223,7 +227,7 @@ sub _delete {
223 227 sub _failed {
224 228 my ($self, $job_id, $opt) = @_;
225 229
226   - my $retry_delay = $self->_server_unixitime + ($opt->{retry_delay} || 60);
  230 + my $retry_delay = $self->_server_unixitime + (defined($opt->{retry_delay}) ? $opt->{retry_delay} : 60);
227 231 try {
228 232 my $sth = $self->{dbh}->prepare_cached($self->{failed_query});
229 233 $sth->execute($retry_delay, $job_id);
@@ -288,6 +292,26 @@ $dbh is database handle.
288 292
289 293 Key word of job which this Jonk instance looks for.
290 294
  295 +=over 4
  296 +
  297 +=item * $options->{functions} = [qw/worker_key worker_key2/]
  298 +
  299 +can set *worker_key* at arrayref.
  300 +
  301 +=item * $options->{functions} = ['worker_key' => {grab_for => 5}],
  302 +
  303 +can set worker_key's grab_for setting by hash-ref.
  304 +
  305 +=item * $options->{functions} = ['worker_key' => {serializer => \&serialize_code, deserializer => \&deserialize_code}],
  306 +
  307 +can set worker_key's (de)serializer code setting by hash-ref.
  308 +
  309 +=item * $options->{functions} = ['worker_key' => {serializer => \&serialize_code, deserializer => \&deserialize_code}, 'worker_key2'],
  310 +
  311 +can mix worker settings.
  312 +
  313 +=back
  314 +
291 315 =item * $options->{table_name}
292 316
293 317 specific job table name.
@@ -300,6 +324,18 @@ specific lookup job record size.
300 324
301 325 Default 50.
302 326
  327 +=item * $options->{default_serializer}
  328 +
  329 +global serializer setting.
  330 +
  331 +=item * $options->{default_deserializer}
  332 +
  333 +global deserializer setting.
  334 +
  335 +=item * $options->{default_grab_for}
  336 +
  337 +global grab_for setting.
  338 +
303 339 =back
304 340
305 341 =head2 my $job_id = $jonk->insert($func, $arg);
@@ -317,10 +353,6 @@ specific your worker funcname.
317 353
318 354 job argument data.
319 355
320   -serialize is not done in Jonk.
321   -
322   -Please pass data that does serialize if it is necessary.
323   -
324 356 =back
325 357
326 358 =head2 my $job = $jonk->lookup_job($job_id);
6 lib/Jonk/Job.pm
@@ -48,7 +48,11 @@ sub aborted {
48 48
49 49 sub id { $_[0]->{job}->{id} }
50 50 sub func { $_[0]->{job}->{func} }
51   -sub arg { $_[0]->{job}->{arg} }
  51 +sub raw_arg { $_[0]->{job}->{arg} }
  52 +sub arg {
  53 + my $self = shift;
  54 + $self->{_jonk}->{functions}->{$self->func}->{deserializer}->($self->{job}->{arg});
  55 +}
52 56 sub enqueue_time { $_[0]->{job}->{enqueue_time} }
53 57 sub grabbed_until { $_[0]->{job}->{grabbed_until} }
54 58 sub retry_cnt { $_[0]->{job}->{retry_cnt} }
103 t/serialize.t
... ... @@ -0,0 +1,103 @@
  1 +use t::Utils;
  2 +use Test::More;
  3 +use DBI;
  4 +use Jonk;
  5 +use Storable ();
  6 +
  7 +my $dbh = t::Utils->setup;
  8 +
  9 +subtest '(de)serializer set each functions' => sub {
  10 + my $serialized_arg;
  11 + my $jonk = Jonk->new($dbh, {
  12 + functions => [
  13 + 'MyWorker' => {
  14 + serializer => sub {$serialized_arg = Storable::nfreeze($_[0])},
  15 + deserializer => sub {Storable::thaw($_[0])}
  16 + }
  17 + ]
  18 + });
  19 +
  20 + subtest 'job completed' => sub {
  21 + my $job_id = $jonk->insert('MyWorker', {chars => 'bar'});
  22 + ok $job_id;
  23 +
  24 + my $job = $jonk->lookup_job($job_id);
  25 + is $job->id, $job_id;
  26 + is_deeply $job->arg, {chars => 'bar'};
  27 + is $job->raw_arg, $serialized_arg;
  28 +
  29 + $job->completed;
  30 +
  31 + ok not $jonk->errstr;
  32 + };
  33 +
  34 + subtest 'job failed and retry' => sub {
  35 +
  36 + my $job_id = $jonk->insert('MyWorker', {chars => 'bar'});
  37 + ok $job_id;
  38 +
  39 + my $job = $jonk->lookup_job($job_id);
  40 + is $job->id, $job_id;
  41 + is_deeply $job->arg, {chars => 'bar'};
  42 + is $job->raw_arg, $serialized_arg;
  43 +
  44 + $job->failed({retry_delay => 0});
  45 +
  46 + $job = $jonk->lookup_job($job_id);
  47 + is $job->id, $job_id;
  48 + is_deeply $job->arg, {chars => 'bar'};
  49 + is $job->raw_arg, $serialized_arg;
  50 +
  51 + $job->completed;
  52 +
  53 + ok not $jonk->errstr;
  54 + };
  55 +};
  56 +
  57 +subtest '(de)serializer set global' => sub {
  58 + my $serialized_arg;
  59 + my $jonk = Jonk->new($dbh, {
  60 + functions => [qw/MyWorker/],
  61 + default_serializer => sub {$serialized_arg = Storable::nfreeze($_[0])},
  62 + default_deserializer => sub {Storable::thaw($_[0])},
  63 + });
  64 +
  65 + subtest 'job completed' => sub {
  66 + my $job_id = $jonk->insert('MyWorker', {chars => 'bar'});
  67 + ok $job_id;
  68 +
  69 + my $job = $jonk->lookup_job($job_id);
  70 + is $job->id, $job_id;
  71 + is_deeply $job->arg, {chars => 'bar'};
  72 + is $job->raw_arg, $serialized_arg;
  73 +
  74 + $job->completed;
  75 +
  76 + ok not $jonk->errstr;
  77 + };
  78 +
  79 + subtest 'job failed and retry' => sub {
  80 +
  81 + my $job_id = $jonk->insert('MyWorker', {chars => 'bar'});
  82 + ok $job_id;
  83 +
  84 + my $job = $jonk->lookup_job($job_id);
  85 + is $job->id, $job_id;
  86 + is_deeply $job->arg, {chars => 'bar'};
  87 + is $job->raw_arg, $serialized_arg;
  88 +
  89 + $job->failed({retry_delay => 0});
  90 +
  91 + $job = $jonk->lookup_job($job_id);
  92 + is $job->id, $job_id;
  93 + is_deeply $job->arg, {chars => 'bar'};
  94 + is $job->raw_arg, $serialized_arg;
  95 +
  96 + $job->completed;
  97 +
  98 + ok not $jonk->errstr;
  99 + };
  100 +};
  101 +
  102 +done_testing;
  103 +

0 comments on commit 7899641

Please sign in to comment.
Something went wrong with that request. Please try again.