summaryrefslogtreecommitdiff
diff options
context:
space:
mode:
Diffstat (limited to '')
-rw-r--r--lib/presque/Service.pm43
1 files changed, 43 insertions, 0 deletions
diff --git a/lib/presque/Service.pm b/lib/presque/Service.pm
new file mode 100644
index 0000000..fd18ace
--- /dev/null
+++ b/lib/presque/Service.pm
@@ -0,0 +1,43 @@
+package presque::Service;
+
+use Moose;
+extends 'Tatsumaki::Service';
+with 'presque::Role::Queue::Names';
+
+has redis => (is => 'rw', isa => 'Object', required => 1);
+
+sub start {
+ my $self = shift;
+ my $t;
+ $t = AE::timer 0, 1, sub {
+ scalar $t;
+ $self->redis->smembers(
+ 'QUEUESET',
+ sub {
+ my $queues = shift;
+ foreach my $q (@$queues) {
+ $self->_check_delayed_queue($q);
+ }
+ }
+ );
+ };
+}
+
+sub _check_delayed_queue {
+ my ($self, $queue_name) = @_;
+
+ my $dkey = $self->_queue_delayed($queue_name);
+
+ $self->redis->zrangebyscore(
+ $dkey, 0, time,
+ sub {
+ my $keys = shift;
+ foreach my $k (@$keys) {
+ $self->redis->zrem($dkey, $k);
+ $self->redis->lpush($self->_queue($queue_name), $k);
+ }
+ }
+ );
+}
+
+1;