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(
        'CommandToDispatch', {
                'foo.bar.baz' => 42,
        }
  );

=cut

sub enqueue {
        my $self = shift;

        my $id = $self->id;
        my ( $dispatch, $args ) = @_;

        warn "## enqueue( $dispatch with ", dump( $args ), " ) for $id\n" if $self->debug;
        
        $self->{dq}->{$id}->enqueue_string( Dump({ dispatch => $dispatch, args => $args }) );
}

=head2 dequeue

  my $job = $q->dequeue;
  my ( $dispatch, $args ) = $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 for $id = ", dump( $job ), " )\n" if $self->debug;

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

=head2 dq

Accessor to C<IPC::DirQueue> object

  my $dq = $q->dq;

=cut

sub dq {
        my $self = shift;
        return $self->{dq}->{$self->id};
}

package CWMP::Queue::Job;

=head1 CWMP::Queue::Job

Single queued job

=cut

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

use YAML qw/LoadFile/;
use Data::Dump qw/dump/;

=head2 dispatch

  my ( $dispatch, $args ) = $job->dispatch;

=cut

sub dispatch {
        my $self = shift;
        my $path = $self->job->get_data_path || die "get_data_path?";
        my $data = LoadFile( $path ) || die "can't read $path: $!";
        warn "## dispatch returns ",dump($data),"\n" if $self->debug;
        return ( $data->{dispatch}, $data->{args} );
}

=head2 finish

Finish job and remove it from queue

  $job->finish;

=cut

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

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 YAML qw/Dump/;
20	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	'CommandToDispatch', {
70	'foo.bar.baz' => 42,
71	}
72	);
73
74	=cut
75
76	sub enqueue {
77	my $self = shift;
78
79	my $id = $self->id;
80	my ( $dispatch, $args ) = @_;
81
82	warn "## enqueue( $dispatch with ", dump( $args ), " ) for $id\n" if $self->debug;
83
84	$self->{dq}->{$id}->enqueue_string( Dump({ dispatch => $dispatch, args => $args }) );
85	}
86
87	=head2 dequeue
88
89	my $job = $q->dequeue;
90	my ( $dispatch, $args ) = $job->dispatch;
91	# after dispatch is processed
92	$job->finish;
93
94	=cut
95
96	sub dequeue {
97	my $self = shift;
98
99	my $id = $self->id;
100
101	my $job = $self->{dq}->{$id}->pickup_queued_job();
102	return unless defined $job;
103
104	warn "## dequeue for $id = ", dump( $job ), " )\n" if $self->debug;
105
106	return CWMP::Queue::Job->new({ job => $job, debug => $self->debug });
107	}
108
109	=head2 dq
110
111	Accessor to C<IPC::DirQueue> object
112
113	my $dq = $q->dq;
114
115	=cut
116
117	sub dq {
118	my $self = shift;
119	return $self->{dq}->{$self->id};
120	}
121
122	package CWMP::Queue::Job;
123
124	=head1 CWMP::Queue::Job
125
126	Single queued job
127
128	=cut
129
130	use base qw/Class::Accessor/;
131	__PACKAGE__->mk_accessors( qw/
132	job
133	debug
134	/ );
135
136	use YAML qw/LoadFile/;
137	use Data::Dump qw/dump/;
138
139	=head2 dispatch
140
141	my ( $dispatch, $args ) = $job->dispatch;
142
143	=cut
144
145	sub dispatch {
146	my $self = shift;
147	my $path = $self->job->get_data_path \|\| die "get_data_path?";
148	my $data = LoadFile( $path ) \|\| die "can't read $path: $!";
149	warn "## dispatch returns ",dump($data),"\n" if $self->debug;
150	return ( $data->{dispatch}, $data->{args} );
151	}
152
153	=head2 finish
154
155	Finish job and remove it from queue
156
157	$job->finish;
158
159	=cut
160
161	sub finish {
162	my $self = shift;
163	$self->job->finish;
164	}
165
166	1;