lib/CWMP/Queue.pm

   1 package CWMP::Queue;
   2
   3 use strict;
   4 use warnings;
   5
   6
   7 use base qw/Class::Accessor/;
   8 __PACKAGE__->mk_accessors( qw/
   9 id
  10 debug
  11
  12 / );
  13
  14 #use Carp qw/confess/;
  15 use Data::Dump qw/dump/;
  16 use File::Spec;
  17 use File::Path qw/mkpath/;
  18 use IPC::DirQueue;
  19 use Carp qw/confess/;
  20
  21 =head1 NAME
  22
  23 CWMP::Queue - implement commands queue for CPE
  24
  25 =head1 METHODS
  26
  27 =head2 new
  28
  29   my $obj = CWMP::Queue->new({
  30         id => 'CPE_serial_number',
  31         debug => 1
  32   });
  33
  34 =cut
  35
  36 sub new {
  37         my $class = shift;
  38         my $self = $class->SUPER::new( @_ );
  39
  40         die "need id" unless $self->id;
  41
  42         warn "created ", __PACKAGE__, "(", dump( @_ ), ") object\n" if $self->debug;
  43
  44         my $dir = File::Spec->catfile('queue',$self->id);
  45
  46         if ( ! -e $dir ) {
  47                 mkpath $dir || die "can't create $dir: $!";
  48                 print "created new queue $dir\n";
  49         }
  50
  51         my $id = $self->id;
  52
  53         if ( ! defined( $self->{dq}->{$id} ) ) {
  54                 $self->{dq}->{$id} = IPC::DirQueue->new({
  55                         dir => $dir,
  56                         ordered => 1,
  57                         queue_fanout => 0,
  58                 });
  59                 warn "## created queue object for CPE $id path $dir\n" if $self->debug;
  60         }
  61
  62         return $self;
  63 }
  64
  65 =head2 enqueue
  66
  67   $q->enqueue(
  68         'foo.bar.baz' => 42,
  69   );
  70
  71 =cut
  72
  73 sub enqueue {
  74         my $self = shift;
  75
  76         my $id = $self->id;
  77         my $data = {@_} || confess "need data";
  78
  79         warn "## enqueue( $id, ", dump( $data ), " )\n" if $self->debug;
  80
  81         $self->{dq}->{$id}->enqueue_string( $id, $data );
  82 }
  83
  84 =head2 dequeue
  85
  86   my $data = $q->dequeue;
  87
  88 =cut
  89
  90 sub dequeue {
  91         my $self = shift;
  92
  93         my $id = $self->id;
  94
  95         my $data = $self->{dq}->{$id}->pickup_queued_job();
  96         return unless defined $data;
  97
  98         warn "## dequeue( $id ) = ", dump( $data ), " )\n" if $self->debug;
  99
 100         return $data->{metadata};
 101 }
 102 1;