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 YAML qw/Dump/;
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;
        
        warn "## enqueue( $id, ", dump( @_ ), " )\n" if $self->debug;
        
        $self->{dq}->{$id}->enqueue_string( Dump( @_ ) );
}

=head2 dequeue

  my $job = $q->dequeue;
  my $dispatch = $job->dispatch;
  # after dispatch is processed
  $job->finish;

=cut

sub dequeue {
        my $self = shift;

        my $id = $self->id;

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

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

        return CWMP::Queue::Job->new({ job => $job });
}

package CWMP::Queue::Job;

use base qw/Class::Accessor/;
__PACKAGE__->mk_accessors( qw/
job
/ );

use YAML qw/LoadFile/;

sub dispatch {
        my $self = shift;
        my $path = $self->job->get_data_path || die "get_data_path?";
        return LoadFile( $path ) || die "can't read $path: $!";
}

sub finish {
        my $self = shift;
        $self->job->finish;
}

1;
1	dpavlin	194	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	dpavlin	197	use YAML qw/Dump/;
20	dpavlin	194	use Carp qw/confess/;
21
22			=head1 NAME
23
24			CWMP::Queue - implement commands queue for CPE
25
26			=head1 METHODS
27
28			=head2 new
29
30			my $obj = CWMP::Queue->new({
31			id => 'CPE_serial_number',
32			debug => 1
33			});
34
35			=cut
36
37			sub new {
38			my $class = shift;
39			my $self = $class->SUPER::new( @_ );
40
41			die "need id" unless $self->id;
42
43			warn "created ", __PACKAGE__, "(", dump( @_ ), ") object\n" if $self->debug;
44
45			my $dir = File::Spec->catfile('queue',$self->id);
46
47			if ( ! -e $dir ) {
48			mkpath $dir \|\| die "can't create $dir: $!";
49			print "created new queue $dir\n";
50			}
51
52			my $id = $self->id;
53
54			if ( ! defined( $self->{dq}->{$id} ) ) {
55			$self->{dq}->{$id} = IPC::DirQueue->new({
56			dir => $dir,
57			ordered => 1,
58			queue_fanout => 0,
59			});
60			warn "## created queue object for CPE $id path $dir\n" if $self->debug;
61			}
62
63			return $self;
64			}
65
66			=head2 enqueue
67
68			$q->enqueue(
69			'foo.bar.baz' => 42,
70			);
71
72			=cut
73
74			sub enqueue {
75			my $self = shift;
76
77			my $id = $self->id;
78	dpavlin	197
79			warn "## enqueue( $id, ", dump( @_ ), " )\n" if $self->debug;
80
81			$self->{dq}->{$id}->enqueue_string( Dump( @_ ) );
82	dpavlin	194	}
83
84			=head2 dequeue
85
86	dpavlin	197	my $job = $q->dequeue;
87			my $dispatch = $job->dispatch;
88			# after dispatch is processed
89			$job->finish;
90	dpavlin	194
91			=cut
92
93			sub dequeue {
94			my $self = shift;
95
96			my $id = $self->id;
97
98	dpavlin	197	my $job = $self->{dq}->{$id}->pickup_queued_job();
99			return unless defined $job;
100	dpavlin	194
101	dpavlin	197	warn "## dequeue( $id ) = ", dump( $job ), " )\n" if $self->debug;
102	dpavlin	194
103	dpavlin	197	return CWMP::Queue::Job->new({ job => $job });
104	dpavlin	194	}
105	dpavlin	197
106			package CWMP::Queue::Job;
107
108			use base qw/Class::Accessor/;
109			__PACKAGE__->mk_accessors( qw/
110			job
111			/ );
112
113			use YAML qw/LoadFile/;
114
115			sub dispatch {
116			my $self = shift;
117			my $path = $self->job->get_data_path \|\| die "get_data_path?";
118			return LoadFile( $path ) \|\| die "can't read $path: $!";
119			}
120
121			sub finish {
122			my $self = shift;
123			$self->job->finish;
124			}
125
126	dpavlin	194	1;