new(%opts)register(@alias)debug_warn($level,@debug_lines)last_id()mark_as($type)process($callback)leave()ignore(@list)fail($message)halt()is_ignored($id)reset()acquire()release()error()consume($callback)proceed($passes)sweep()runstats()
Data::Consumer - Repeatedly consume a data resource in a robust way
Version 0.09
use Data::Consumer;
my $consumer = Data::Consumer->new(
type => $consumer_name,
unprocessed => $unprocessed,
working => $working,
processed => $processed,
failed => $failed,
max_passes => $num_or_undef,
max_process => $num_or_undef,
max_elapsed => $seconds_or_undef,
);
$consumer->consume( sub {
my $id = shift;
print "processed $id\n";
} );
It is a common requirement to need to process a feed of items of some sort in a robust manner. Such a feed might be records that are inserted into a table, or files dropped in a delivery directory. Writing a script that handles all the edge cases, like getting "stuck" on a failed item, and manages things like locking so that the script can be parallelized can be tricky and is certainly repetitive.
The aim of the Data::Consumer manpage is to provide a framework to allow writing such consumer type scripts as easy as writing a callback that processes each item. The framework handles the rest.
The basic idea is that one need only use, or in the case of a feed type not already supported, define a the Data::Consumer manpage subclass which implements a few reasonably well defined primitive methods which handle the required tasks, and then the the Data::Consumer manpage methods use those to provide a DWIMily consistant interface to the end consumer.
Currently the Data::Consumer manpage is distributed with two subclasses,
the Data::Consumer::MySQL manpage for handling records in a MySQL db (using the
MySQL GET_LOCK() function), and the Data::Consumer::Dir manpage for handling
a drop directory scenario (like for FTP or a mail directory).
Once a resource type has been defined as a the Data::Consumer manpage subclass the use pattern is to construct the subclass with the appropriate arguments, and then call consume with a callback.
The consumer pattern is where code wants to consume an 'atomic' resource piece by piece. The consuming codes doesnt really want to worry much about how they got the piece a task that should be handled by the framework. The consumer subclasses assume that the resource can be modeled as a queue (that there is some ordering principle by which they can be processed in a predictable sequence). The consume pattern in full glory is something very close to the following following pseudo code. The items marked with asterixs are where user callbacks may be invoked:
DO
RESET TO THE BEGINNING OF THE QUEUE
WHILE QUEUE NOT EMPTY AND CAN *PROCEED*
ACQUIRE NEXT ITEM TO PROCESS FROM QUEUE
MARK AS 'WORKING'
*PROCESS* ITEM
IF PROCESSING FAILED
MARK AS 'FAILED'
OTHERWISE
MARK AS 'PROCESSED'
SWEEP UP ABANDONDED 'WORKING' ITEMS AND MARK THEM AS 'FAILED'
UNTIL WE CANNOT *PROCEED* OR NOTHING WAS PROCESSED
RELEASE ANY LOCKS STILL HELD
This implies that each item potentially has four states: unprocessed,
working, processed and failed. In a database these might be
values in a field, in a drop directory scenario these would be different
directories, but with all of them they would normally be supplied as
values to the the Data::Consumer manpage subclass being created.
the Data::Consumer manpage can be used with any resource type that can be modeled
as a queue, supports some form of advisory locking mechanism, and
provides a way to discriminate between at least the unprocessed and
processed state.
The routines that must be defined for a new consumer type are new(),
reset(), acquire(), release(), and _mark_as(),
_do_callback(), and possibly _fix_sweeper().
It is almost for sure that a subclass will need to override the default constructor. All the Data::Consumer manpage objects are blessed hashes, and in fact you should always call the parents classes constructor first with:
my $self= $class->SUPER::new();
This routine is used to reset the objects internal state so the next call to acquire will return the first available item in the queue.
This routine is to find and in some way lock the next item in the queue. It should ensure
that it call is_ignored() on each item to verify the item has not been requested to be
ignored.
This routine is to release any held locks in the object.
This routine is called to "mark" an item as a particular state. It should be able to handle user supplied values. For instance the Data::Consumer::MySQL manpage implements this as an update statement that maps user supplied values to the consumer state names.
Possible states are: unprocessed, working, processed,
failed.
This routine is used to call the user supplied callback with the correct
arguments. What arguments are appropriate for the callback are context
dependent on the type of class. For instance in the Data::Consumer::MySQL manpage
calls the callback with the arguments ($consumer, $id, $dbh) whereas
the Data::Consumer::Dir manpage calls the callback with the arguments
($consumer, $filespec, $filehandle, $filename). The point is that the
end user should be passed the arguments that make sense, not necessarily
the same thing for each consumer type.
the Data::Consumer manpage has support for a "sweep-up" mode for handling things
that were abandonded in the working state and moving them to the
failed state. This is mostly a sanity check to prevent things like
powerfailures and segfaults leaving an item in a partially or
unprocessed state without it being properly marked as failed. In order
to do this the the Data::Consumer manpage subclass actually creates a private
near clone of itself. _fix_sweeper() is called in case the normal
modifications done by the base class routine are not sufficient for a
subclass. As an example the Data::Consumer::MySQL manpage provides this hook
while the Data::Consumer::Dir manpage does not.
Every well-behaved the Data::Consumer manpage subclass should include the functional equivalent of the following code in its .pm file:
use base 'Data::Consumer';
__PACKAGE__->register();
This will ensure that it can be properly loaded by
Data::Consumer->new(type=>$shortname).
It is also normal for a the Data::Consumer manpage subclass to provide special
methods as needed. For instance Data::Consumer::Dir->fh() and
Data::Consumer::MySQL->dbh().
new(%opts)Constructor. Normally the Data::Consumer manpage's constructor is not called
directly, instead the constructor of a subclass is used. However to
make it easier to have a data driven load process the Data::Consumer manpage
accepts the type argument which should specify the the short name of
the subclass (the part after Data::Consumer::) or the full name of
the subclass.
Thus
Data::Consumer->new(type=>'MySQL',%args);
is exactly equivalent to calling
Data::Consumer::MySQL->new(%args);
except that the former will automatically require or use the appropriate module and the latter necessitates that you do so yourself.
Every the Data::Consumer manpage subclass constructor supports the following arguments on top of any that are subclass specific. Additionally some arguments are universally used, but have different meaning depending on the subclass.
How to tell if the item is unprocessed.
How this argument is interpreted depends on the the Data::Consumer manpage subclass involved.
How to tell if the item is currently being worked on.
How this argument is interpreted depends on the the Data::Consumer manpage subclass involved.
How to tell if the item has already been worked on.
How this argument is interpreted depends on the the Data::Consumer manpage subclass involved.
How to tell if processing failed while handling the item.
How this argument is interpreted depends on the the Data::Consumer manpage subclass involved.
Normally consume() will loop through the data set until it is
exhausted. By setting this parameter you can control the maximum number
of iterations, for instance setting it to 1 will result in a single
pass through the data per invocation. If 0 (or any other false value)
is treated as meaning "loop until exhausted".
Maximum number of items to process per invocation.
If set to a false value there is no limit.
Maximum number of failed process attempts that may occur before consume will stop. If set to a false value there is no limit. Setting this to 1 will cause processing to stop after the first failure.
Maximum amount of time that may have elapsed when starting a new
process. If more than this value has elapsed then no further processing
occurs. If 0 (or any false value) then there is no time limit.
This is a callback that may be used to control the looping process in
consume via the proceed() method. See the documentation of
consume() and proceed()
If this parameter is true, and there are four modes defined
(unprocessed, working, processed, failed) then consume will
perform a "sweep up" after every pass, which is responsible for moving
"abandonded" files from the working directory (such as from a previous
process that segfaulted during processing). Generally this should
not be necessary.
register(@alias)Used by subclasses to register themselves as a the Data::Consumer manpage subclass and register any additional aliases that the class may be identified as.
Will throw an exception if any of the aliases are already associated to a different class.
When called on a subclass in list context returns a list of the subclasses registered aliases,
If called on the Data::Consumer manpage in list context returns a list of all alias class mappings.
debug_warn($level,@debug_lines)If Debug is enabled and above $level then print @debug_lines to
STDOUT in a specific format that includes the class name of the
caller and process id.
last_id()Returns the identifier for the last item acquired.
Returns undef if acquire has never been called or if the last attempt to acquire data failed because none was available.
mark_as($type)Mark an item as a particular type if the object defines that type.
Allowed types are unprocessed, working, processed, failed
process($callback)Marks the current item as working and processes it using the
$callback. If the $callback dies then the item is marked as
failed, otherwise the item is marked as processed once the
$callback returns. The return value of the $callback is ignored.
$callback will be called with at least two arguments, the first being
the $consumer object itself, and the second being an identifier for the
current record. Normally additional, likely to be useful, arguments are
provided as well, on a per subclass basis. For example
the Data::Consumer::MySQL manpage will pass in the consumer object, the id of the to
be processed record, and a copy of the consumers database handle as well for
convenience. On the other hand the Data::Consumer::Dir manpage will pass in the
consumer object, followed by a filespecification for the file to be
processed, an open filehandle to the file, and the filename itself (with
no path).
The callback may call the methods 'leave', 'ignore', 'fail', and 'halt' on the consumer object before returning, typically by doing something like
return $consumer->ignore;
this allows the callback to send specific signals to consume, specifically
leave : return the item to the unprocessed state after the callback returns.
ignore : return the item to the unprocessed state after the callback returns
and never attempt to process it again with this consumer object.
fail : same result as dieing in a callback, except without throwing an exception
in the situation where there might be $SIG{__DIE__} hooks to worry about.
halt : stop the consume() process after this has been executed
For further details always consult the relevent subclasses documentation for
process()
leave()Sometimes its useful to defer processing. This method when called from within a consume/process callback will result in the item being marked as 'unprocessed' after the callback returns (so long as it does not die).
Typically this is invoked as
return $consumer->leave;
from withing a consume/process callback.
Returns $consumer. Will die if not 'unprocessed' state is defined.
ignore(@list)This can used to cause acquire to ignore each item in @list.
If @list is empty then it is assumed it is being called from
within consume/process and and marks the currently acquired item
as ignored and calls $consumer->leave().
Returns $consumer. Will die if not 'unprocessed' state is defined.
fail($message)Same as doing die($message) from within a consume/process callback except
that no exception is thrown (no $SIG{__DIE__} callbacks are invoked) and
the error is defered until the callback actually returns.
Typically used as
return $consumer->fail;
from within a consumer() callback.
Returns the $consumer object.
halt()Causes consume() to halt processing and exit once
the callback returns. Typically invoked like
return $consumer->halt;
or
return $consumer->fail->halt;
Returns the consumer object.
is_ignored($id)Returns true if an item has been set to be ignored. If $id is omitted defaults to last_id
reset()Reset the state of the object.
acquire()Aquire an item to be processed.
returns an identifier to be used to identify the item acquired.
release()Release any locks on the currently held item.
Normally there is no need to call this directly.
error()Calls the error callback if the user has provided one, otherwise
calls confess(). Probably not all that useful for an end user.
consume($callback)Consumes a data resource until it is exhausted using acquire(),
process(), and release() as appropriate. Normally this is the main
method used by external processes.
Before each attempt to acquire a new resource, and once at the end of
each pass consume will call proceed() to determine if it can do so.
The user may hook into this by specifying a callback in the constructor.
This callback will be executed with no args when it is in the inner loop
(per item), and with the number of passes at the end of each pass
(starting with 1).
proceed($passes)Returns true if the conditions specified at construction time are
satisfied and processing may proceed. Returns false otherwise.
If the user has specified a proceed callback in the constructor then
this will be executed before any other rules are applied, with a
reference to the current $object, a reference to the runstats, and if
being called at the end of pass with the number of passes.
If this callback returns true then the other rules will be applied,
and only if all other conditions from the constructor are satisfied
will proceed() itself return true.
sweep()If the user has specified both a working and a failed state then
this routine will move all lockable working items and change them to
the failed state. This is to catch catastrophic failures where
unprocessed items are left in the working state. Presumably this is a
rare case.
runstats()Returns a reference to a hash of statistics about the last (or currently running) execution of consume.
Yves Orton, <YVES at cpan.org>
Please report any bugs or feature requests to
bug-data-consumer at rt.cpan.org, or through the web interface at
http://rt.cpan.org/NoAuth/ReportBug.html.
I will be notified, and then you'll automatically be notified of progress on your bug as I make changes.
You can find documentation for this module with the perldoc command.
perldoc Data::Consumer
You can also look for information at:
Igor Sutton <IZUT@cpan.org> for ideas, testing and support
Copyright 2008 Yves Orton, all rights reserved.
This program is free software; you can redistribute it and/or modify it under the same terms as Perl itself.