Guillaume Aubert > Queue-DBI-2.1.1 > Queue::DBI

Download:
Queue-DBI-2.1.1.tar.gz

Dependencies

Annotate this POD

Website

CPAN RT

Open  0
View/Report Bugs
Module Version: v2.1.1   Source   Latest Release: Queue-DBI-v2.6.1

NAME ^

Queue::DBI - A queueing module with an emphasis on safety, using DBI as a storage system for queued data.

VERSION ^

Version 2.1.1

SYNOPSIS ^

This module allows you to safely use a queueing system by preventing backtracking, infinite loops and data loss.

An emphasis of this distribution is to provide an extremely reliable dequeueing mechanism without having to use transactions.

        use Queue::DBI;
        my $queue = Queue::DBI->new(
                'queue_name'      => $queue_name,
                'database_handle' => $dbh,
                'cleanup_timeout' => 3600,
                'verbose'         => 1,
        );
        
        $queue->enqueue( $data );
        
        while ( my $queue_element = $queue->next() )
        {
                next
                        unless $queue_element->lock();
                
                eval {
                        # Do some work
                        process( $queue_element->{'email'} );
                };
                if ( $@ )
                {
                        # Something failed, we clear the lock but don't delete the record in the
                        # queue so that we can try again next time
                        $queue_element->requeue();
                }
                else
                {
                        # All good, remove definitively the element
                        $queue_element->success();
                }
        }
        
        # Requeue items that have been locked for more than 6 hours
        $queue->cleanup( 6 * 3600 );

METHODS ^

new()

Create a new Queue::DBI object.

        my $queue = Queue::DBI->new(
                'queue_name'        => $queue_name,
                'database_handle'   => $dbh,
                'cleanup_timeout'   => 3600,
                'verbose'           => 1,
                'max_requeue_count' => 5,
        );
        
        # Custom table names (optional).
        my $queue = Queue::DBI->new(
                'queue_name'                => $queue_name,
                'database_handle'           => $dbh,
                'cleanup_timeout'           => 3600,
                'verbose'                   => 1,
                'max_requeue_count'         => 5,
                'queues_table_name'         => $custom_queues_table_name,
                'queue_elements_table_name' => $custom_queue_elements_table_name,
        );

Parameters:

get_queue_id()

Returns the queue ID corresponding to the current queue object.

        my $queue_id = $self->get_queue_id();

count()

Returns the number of elements in the queue.

enqueue()

Adds a new element at the end of the current queue.

        my $queue_element_id = $queue->enqueue( $data );

The data passed can be a scalar or a reference to a complex data structure. There is no limitation on the type of data that can be stored as it is serialized for storage in the database.

next()

Retrieves the next element from the queue and returns it in the form of a Queue::DBI::Element object.

        my $queue_element = $queue->next();
        
        while ( my $queue_element = $queue->next() )
        {
                # [...]
        }

Additionally, for testing purposes, a list of IDs to use when trying to retrieve elements can be specified using 'search_in_ids':

        my $queue_item = $queue->next( 'search_in_ids' => [ 123, 124, 125 ] );

retrieve_batch()

Retrieves a batch of elements from the queue and returns them in an arrayref.

This method requires an integer to be passed as parameter to indicate the maximum size of the batch to be retrieved.

        my $queue_elements = $queue->retrieve_batch( 500 );
        
        foreach ( @$queue_elements )
        {
                # [...]
        }

Additionally, for testing purposes, a list of IDs to use when trying to retrieve elements can be specified using 'search_in_ids':

        my $queue_items = $queue->retrieve_batch(
                10,
                'search_in_ids' => [ 123, 124, 125 ],
        );

get_element_by_id()

Retrieves a queue element using a queue element ID, ignoring any lock placed on that element.

This method is mostly useful when doing a lock on an element and then calling success/requeue asynchroneously.

This method requires a queue element ID to be passed as parameter.

        my $queue_element = $queue->get_element_by_id( 123456 );

cleanup()

Requeue items that have been locked for more than the time in seconds specified as parameter.

Returns the items requeued so that a specific action can be taken on them.

        my $elements = $queue->cleanup( $time_in_seconds );
        foreach my $element ( @$elements )
        {
                # $element is a Queue::DBI::Element object
        }

purge()

Remove (permanently, caveat emptor!) queue elements based on how many times they've been requeued or how old they are.

        # Remove permanently elements that have been requeued more than 10 times.
        $queue->purge( max_requeue_count => 10 );
        
        # Remove permanently elements that were created over an hour ago.
        $queue->purge( lifetime => 3600 );

Important: locked elements are not purged even if they match the criteria, as they are presumed to be currently in process and purging them would create unexpected failures in the application processing them.

Also note that max_requeue_count and lifetime cannot be combined.

ACCESSORS ^

get_max_requeue_count()

Return how many times an element can be requeued before it is ignored when retrieving elements.

        my $max_requeue_count = $queue->get_max_requeue_count();

set_max_requeue_count()

Set the number of time an element can be requeued before it is ignored when retrieving elements. Set it to undef to disable the limit.

        # Don't keep pulling the element if it has been requeued more than 5 times.
        $queue->set_max_requeue_count( 5 );+
        
        # Retry without limit.
        $queue->set_max_requeue_count( undef );

get_lifetime()

Return how old an element can be before it is ignored when retrieving elements.

        # Find how old an element can be before the queue will stop retrieving it.
        my $lifetime = $queue->get_lifetime();

set_lifetime()

Set how old an element can be before it is ignored when retrieving elements.

Set it to undef to reset Queue::DBI back to its default behavior of retrieving elements without time limit.

        # Don't pull queue elements that are more than an hour old.
        $queue->set_lifetime( 3600 );
        
        # Pull elements without time limit.
        $queue->set_lifetime( undef );

get_verbose()

Return the verbosity level, which is used in the module to determine when and what type of debugging statements / information should be warned out.

See set_verbose() for the possible values this function can return.

        warn 'Verbose' if $queue->get_verbose();
        
        warn 'Very verbose' if $queue->get_verbose() > 1;

set_verbose()

Control the verbosity of the warnings in the code:

        $queue->set_verbose(1); # turn on verbose information
        
        $queue->set_verbose(2); # be extra verbose
        
        $queue->set_verbose(0); # quiet now!

DEPRECATED METHODS ^

create_tables()

Please use create_tables() in Queue::DBI::Admin instead.

Here is an example that shows how to refactor your call to this deprecated function:

        # Load the admin module.
        use Queue::DBI::Admin;
        
        # Create the object which will allow managing the queues.
        my $queues_admin = Queue::DBI::Admin->new(
                database_handle => $dbh,
        );
        
        # Create the tables required by Queue::DBI to store the queues and data.
        $queues_admin->create_tables(
                drop_if_exist => $boolean,
                sqlite        => $boolean,
        );

lifetime()

Please use get_lifetime() and set_lifetime() instead.

verbose()

Please use get_verbose() and set_verbose() instead.

max_requeue_count()

Please use get_max_requeue_count() and set_max_requeue_count() instead.

INTERNAL METHODS ^

get_dbh()

Returns the database handle used for this queue.

        my $dbh = $queue->get_dbh();

get_queues_table_name()

Returns the name of the table used to store queue definitions.

        my $queues_table_name = $queue->get_queues_table_name();

get_queue_elements_table_name()

Returns the name of the table used to store queue definitions.

        my $queue_elements_table_name = $queue->get_queue_elements_table_name();

AUTHOR ^

Guillaume Aubert, <aubertg at cpan.org>.

BUGS ^

Please report any bugs or feature requests to bug-queue-dbi at rt.cpan.org, or through the web interface at http://rt.cpan.org/NoAuth/ReportBug.html?Queue=Queue-DBI. I will be notified, and then you'll automatically be notified of progress on your bug as I make changes.

SUPPORT ^

You can find documentation for this module with the perldoc command.

        perldoc Queue::DBI

You can also look for information at:

ACKNOWLEDGEMENTS ^

Thanks to ThinkGeek (http://www.thinkgeek.com/) and its corporate overlords at Geeknet (http://www.geek.net/), for footing the bill while I write code for them!

Thanks to Jacob Rose <jacob at thinkgeek.com>, who wrote the first queueing module at ThinkGeek http://www.thinkgeek.com and whose work provided the inspiration to write this full-fledged queueing system.

Thanks to Jamie McCarthy for the locking mechanism improvements in version 1.1.0.

Thanks to Sergey Bond for suggesting many features added in version 1.8.x (lifetime constraint, purge() function, get/set functions cleanup).

COPYRIGHT & LICENSE ^

Copyright 2009-2012 Guillaume Aubert.

This program is free software: you can redistribute it and/or modify it under the terms of the GNU General Public License version 3 as published by the Free Software Foundation.

This program is distributed in the hope that it will be useful, but WITHOUT ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for more details.

You should have received a copy of the GNU General Public License along with this program. If not, see http://www.gnu.org/licenses/

syntax highlighting: