lib/CWMP/Queue.pm

package CWMP::Queue;

use strict;
use warnings;


use base qw/Class::Accessor/;
__PACKAGE__->mk_accessors( qw/
id
debug

/ );

#use Carp qw/confess/;
use Data::Dump qw/dump/;
use File::Spec;
use File::Path qw/mkpath/;
use IPC::DirQueue;
use Carp qw/confess/;

=head1 NAME

CWMP::Queue - implement commands queue for CPE

=head1 METHODS

=head2 new

  my $obj = CWMP::Queue->new({
        id => 'CPE_serial_number',
        debug => 1
  });

=cut

sub new {
        my $class = shift;
        my $self = $class->SUPER::new( @_ );

        die "need id" unless $self->id;

        warn "created ", __PACKAGE__, "(", dump( @_ ), ") object\n" if $self->debug;

        my $dir = File::Spec->catfile('queue',$self->id);

        if ( ! -e $dir ) {
                mkpath $dir || die "can't create $dir: $!";
                print "created new queue $dir\n";
        }

        my $id = $self->id;

        if ( ! defined( $self->{dq}->{$id} ) ) {
                $self->{dq}->{$id} = IPC::DirQueue->new({
                        dir => $dir,
                        ordered => 1,
                        queue_fanout => 0,
                });
                warn "## created queue object for CPE $id path $dir\n" if $self->debug;
        }

        return $self;
}

=head2 enqueue

  $q->enqueue(
        'foo.bar.baz' => 42,
  );

=cut

sub enqueue {
        my $self = shift;

        my $id = $self->id;
        my $data = {@_} || confess "need data";

        warn "## enqueue( $id, ", dump( $data ), " )\n" if $self->debug;

        $self->{dq}->{$id}->enqueue_string( $id, $data );
}

=head2 dequeue

  my $data = $q->dequeue;

=cut

sub dequeue {
        my $self = shift;

        my $id = $self->id;

        my $data = $self->{dq}->{$id}->pickup_queued_job();
        return unless defined $data;

        warn "## dequeue( $id ) = ", dump( $data ), " )\n" if $self->debug;

        return $data->{metadata};
}
1;
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;