POE::Component::Schedule: Request for feature
[email protected] (Deven Parekh)
| Newsgroups | perl.poe |
|---|---|
| Message-ID | <[email protected]> |
Hello Oliver,
I saw your other mail that you were re-factoring the
POE::Component::Schedule.
I was tempted to reply to that but did not want to pollute that thread.
First of all this is a great module. Thanks.
I have a feature request for this module for your consideration.
I was wondering if a capability could be defined to plug in a
custom iterator to use (instead of DateTime::Set). I was interested in
Algorithm::TokenBucket as the iterator.
Thanks in advance for the consideration.
Regards
Deven.
p.s: I gave it an attempt and came up with a version that works. Basically
your module + some modification to _schedule and minors ones to
_client_event
and of course the iterator type check in add. But was wondering if some
how the
two can be combined? I attached the experimental module an example
script that
uses it and the output of that script. I guess i can try subclass from
POCO::Schedule
and just overload these three routines.
p.p.s: This is based on the original version of this module before
re-factoring.
output.txt
(text/plain, 542 B)
./example_1.pl start 1258751264.06958 tick 1258751265.74366 ( 00:00:01.674042 ) tick 1258751267.41223 ( 00:00:03.342612 ) tick 1258751269.0807 ( 00:00:05.011077 ) tick 1258751270.7492 ( 00:00:06.679584 ) tick 1258751272.41771 ( 00:00:08.348090 ) tick 1258751274.08622 ( 00:00:10.016599 ) tick 1258751275.75471 ( 00:00:11.685086 ) tick 1258751277.42326 ( 00:00:13.353643 ) tick 1258751279.09176 ( 00:00:15.022136 ) tick 1258751280.76028 ( 00:00:16.690656 ) tick 1258751282.42877 ( 00:00:18.359147 ) tick 1258751284.09731 ( 00:00:20.027685 )
ATkB.pm
(application/octet-stream, 7.2 KB)
package POE::Component::Schedule::ATkB;
use 5.008;
our $VERSION = '0.03';
use strict;
use warnings;
use POE;
my $Singleton;
my $ID_Sequence = 'a'; # sequence is 'a', 'b', ..., 'z', 'aa', 'ab', ...
my %Schedule_Ticket; # Hash helps remember alarm id for cancel.
#
# crank up the schedule session
#
sub spawn {
my $class = shift;
my %arg = @_;
if ( !defined $Singleton ) {
$Singleton = POE::Session->create(
inline_states => {
_start => sub {
my ($k) = $_[KERNEL];
$k->alias_set( $arg{'Alias'} || $class );
$k->sig( 'SHUTDOWN', 'shutdown' );
},
schedule => \&_schedule,
client_event => \&_client_event,
cancel => \&_cancel,
shutdown => sub {
#print "# $class shutdown\n";
my $k = $_[KERNEL];
# FIXME We are removing too much here!
$k->alarm_remove_all();
$k->sig_handled();
},
_stop => sub {
#print "# $class _stop\n";
$Singleton = undef;
},
},
)->ID;
}
}
#
# schedule the next event
# ARG0 is a client session,
# ARG1 is the client event name,
# ARG2 is a DateTime::Set iterator
# ARG3 is an schedule ticket
# ARG4 .. $#_ are arguments to the client event
#
#--- Overload add and schedule
#
## schedule the next event
## ARG0 is the schedule ticket
##
sub _schedule {
my ( $k, $s, $e, $ds, $tix, @arg ) = @_[ KERNEL, ARG0 .. $#_ ];
my @ds_state = $ds->state();
my $burst_size = $ds_state[1];
if ($ds->conform($burst_size) ) {
#--- It is time to call the event
for (my $i = 0; $i < $burst_size;$i++) {
$k->yield( 'client_event', $s, $e, $ds, $tix, @arg );
}
$ds->count($burst_size);
}
#--- How long before we do another check
my $n;
$n = $ds->until($burst_size);
$Schedule_Ticket{$tix} = $poe_kernel->delay_set( 'schedule', $n, $s, $e, $ds, $tix, @arg );
}
#
# handle a client event and schedule the next one
# ARG0 is a client session,
# ARG1 is the client event name,
# ARG2 is a DateTime::Set iterator
# ARG3 is an schedule ticket
# ARG4 .. $#_ are arguments to the client event
#
sub _client_event {
my ( $k, $s, $e, $ds, $tix, @arg ) = @_[ KERNEL, ARG0 .. $#_ ];
$k->post( $s, $e, @arg );
}
#
# cancel an alarm
#
sub _cancel {
my ( $k, $id ) = @_[ KERNEL, ARG0 ];
$k->alarm_remove($id);
}
sub add {
my $class = shift;
my $ticket = $ID_Sequence++; # get the next ticket;
my ( $session, $event, $iterator, @args ) = @_;
$iterator->isa('Algorithm::TokenBucket')
or die __PACKAGE__ . "->add: third arg must be a Algorithm::TokenBucket";
$class->spawn unless $Singleton;
$poe_kernel->post( $poe_kernel->ID_id_to_session($Singleton),
'schedule', $session, $event, $iterator, $ticket, @args, );
$Schedule_Ticket{$ticket} = ();
return bless \$ticket, ref $class || $class;
}
sub delete {
my $self = shift;
my $ticket = $$self;
$poe_kernel->post(
$poe_kernel->ID_id_to_session($Singleton),
'cancel', $Schedule_Ticket{$ticket},
);
delete $Schedule_Ticket{$ticket};
}
sub DESTROY {
$_[0]->delete if exists $Schedule_Ticket{${$_[0]}};
}
{
no warnings;
*new = \&add;
}
1;
__END__
=head1 NAME
POE::Component::Schedule - Schedule POE events using DateTime::Set iterators
=head1 SYNOPSIS
use POE qw(Component::Schedule);
use DateTime::Set;
$s1 = POE::Session->create(
inline_states => {
_start => sub {
$_[HEAP]{sched} = POE::Component::Schedule->add(
$_[SESSION], Tick => DateTime::Set->from_recurrence(
after => DateTime->now,
before => DateTime->now->add(seconds => 3)
recurrence => sub {
return $_[0]->truncate( to => 'second' )->add( seconds => 1 )
},
),
);
},
Tick => sub {
print 'tick ', scalar localtime, "\n";
},
remove_sched => sub {
# Three ways to remove a schedule
# The first one is only for API compatibility with POE::Component::Cron
$_[HEAP]{sched}->delete;
$_[HEAP]{sched} = undef;
delete $_[HEAP]{sched};
}
_stop => sub {
print "_stop\n";
},
},
);
=head1 DESCRIPTION
This component encapsulates a session that sends events to client sessions
on a schedule as defined by a DateTime::Set iterator.
=head1 POE::Component::Schedule METHODS
=head2 spawn(Alias => I<name>)
No need to call this in normal use, add() and new() all crank
one of these up if it is needed. Start up a PoCo::Schedule. Returns a
handle that can then be added to.
=head2 add()
my $sched = POE::Component::Schedule->add(
$session_object,
$event_name,
$DateTime_Set_iterator,
@event_args
);
Add a set of events to the schedule. The C<$session_object> and C<$event_name> are passed
to POE without even checking to see if they are valid and so have the same
warnings as ->post() itself.
C<$session_object> must be a real L<POE::Session>, not a session ID. Else session
reference count will not be increased and the session may end before receiving all
events.
Returns a schedule handle. The event is removed from when the handle is not referenced
anymore.
=head2 new
new is an alias for add
=head1 SCHEDULE HANDLE METHODS
=head2 delete
Removes a schedule using the handle returned from ->add or ->new.
B<DEPRECATED>: Schedules are now automatically deleted when they are not
referenced anymore. So just setting the container variable to C<undef> will
delete the schedule.
=head1 SEE ALSO
L<POE>, L<DateTime::Set>, L<POE::Component::Cron>.
=head1 BUGS
You can look for information at:
=over 4
=item * RT: CPAN's request tracker
L<http://rt.cpan.org/NoAuth/Bugs.html?Dist=POE-Component-Schedule>
=item * AnnoCPAN: Annotated CPAN documentation
L<http://annocpan.org/dist/POE-Component-Schedule>
=item * CPAN Ratings
L<http://cpanratings.perl.org/p/POE-Component-Schedule>
=item * Search CPAN
L<http://search.cpan.org/dist/POE-Component-Schedule/>
=back
=head1 ACKNOWLEDGMENT
This module is a friendly fork of POE::Component::Cron to extract the generic
parts and isolate the Cron specific code in order to reduce dependencies on
other CPAN modules.
The orignal author of POE::Component::Cron is Chris Fedde.
See L<https://rt.cpan.org/Ticket/Display.html?id=44442>
=head1 AUTHORS
=over 4
=item Olivier MenguE<eacute>, C<<< [email protected] >>>
=item Chris Fedde, C<<< [email protected] >>>
=back
=head1 COPYRIGHT AND LICENSE
=over 4
=item Copyright E<copy> 2007-2008 Chris Fedde
=item Copyright E<copy> 2009 Olivier MenguE<eacute>
=back
This library is free software; you can redistribute it and/or modify
it under the same terms as Perl itself, either Perl version 5.8.3 or,
at your option, any later version of Perl 5 you may have available.
=cut
example_1.pl
(application/octet-stream, 1.2 KB)
#!/usr/bin/perl
use strict;
use warnings;
use lib "/path/to/module"; #--- not needed if Schedule::ATkB is installed
use POE qw(Component::Schedule::ATkB);
use Algorithm::TokenBucket;
use Time::HiRes qw( time );
use Time::Elapse;
my $so_many_task = 3;
my $so_many_seconds = 5;
my $burst_size = 1;
my $rate = $so_many_task / $so_many_seconds;
my $elapsed_time;
print 'start ', time , "\n";
Time::Elapse->lapse($elapsed_time);
POE::Session->create(
inline_states => {
_start => sub {
$_[KERNEL]->yield("add_sched");
},
Tick => sub {
#--- Just to show that we do get Ticks at the said rate
print 'tick ', time, " ( $elapsed_time )\n";
},
add_sched => sub {
$_[HEAP]{sched} = POE::Component::Schedule::ATkB->add(
$_[SESSION], Tick => Algorithm::TokenBucket->new($rate, $burst_size),
);
},
remove_sched => sub {
$_[HEAP]{sched}->delete;
#$_[HEAP]{sched} = undef;
#delete $_[HEAP]{sched};
},
_stop => sub {
print "_stop\n";
},
},
);
POE::Kernel->run();
exit(0);