Net-XMPP2/samples/EVQ.pm

package EVQ;
use strict;
use AnyEvent;

my $J;

our %reqh;
our $id = 0;
my @req;

sub schedule {
   my $reqcnt = scalar (keys %reqh);
   if ($reqcnt == 0 && !@req) {
      warn "no more jobs, finishing...\n";
      $J->broadcast;
   }
   while ($reqcnt < 200) {
      my $r = pop @req;
      return unless defined $r;
      eval {
         $r->[0]->(addreq ($r->[1]));
      };
      if ($@) { warn "EXCEPTION: $@\n" }
      $reqcnt = scalar (keys %reqh);
   }
}

sub addreq { my $k = $id . "_" . $_[0]; $reqh{$k} = 1; $id++; $k }
sub finreq { delete $reqh{$_[0]}; }

sub push_request {
   my ($s, $cb) = @_;
   push @req, [$cb, $s];
   schedule;
}

our $t;
sub timer {
   $t = AnyEvent->timer (after => 1, cb => sub {
      schedule;
      my $reqcnt = scalar (keys %reqh);
      $reqcnt += @req;
      my $rreqcnt = scalar (keys %reqh);
      warn "$reqcnt outstanding requests [$rreqcnt in progress]\n";
      timer ();
   });
}

sub start {
   $J = AnyEvent->condvar;
   timer;
}
sub wait {
   $J->wait;
}

1
Revision:	1.2
Committed:	Thu Jul 19 11:36:34 2007 UTC (18 years, 10 months ago) by elmex
Branch:	MAIN
CVS Tags:	HEAD
Changes since 1.1:	+9 -5 lines
Log Message:	added initial_presence argument to the IM::Connection and the Client. added and upgraded some examples. further work on the registration forms.
#	User	Rev	Content
1	elmex	1.1	package EVQ;
2			use strict;
3			use AnyEvent;
4
5			my $J;
6
7			our %reqh;
8			our $id = 0;
9			my @req;
10
11			sub schedule {
12			my $reqcnt = scalar (keys %reqh);
13			if ($reqcnt == 0 && !@req) {
14	elmex	1.2	warn "no more jobs, finishing...\n";
15	elmex	1.1	$J->broadcast;
16			}
17	elmex	1.2	while ($reqcnt < 200) {
18	elmex	1.1	my $r = pop @req;
19			return unless defined $r;
20	elmex	1.2	eval {
21			$r->[0]->(addreq ($r->[1]));
22			};
23			if ($@) { warn "EXCEPTION: $@\n" }
24	elmex	1.1	$reqcnt = scalar (keys %reqh);
25			}
26			}
27
28			sub addreq { my $k = $id . "_" . $_[0]; $reqh{$k} = 1; $id++; $k }
29			sub finreq { delete $reqh{$_[0]}; }
30
31			sub push_request {
32			my ($s, $cb) = @_;
33			push @req, [$cb, $s];
34			schedule;
35			}
36
37			our $t;
38			sub timer {
39	elmex	1.2	$t = AnyEvent->timer (after => 1, cb => sub {
40	elmex	1.1	schedule;
41			my $reqcnt = scalar (keys %reqh);
42			$reqcnt += @req;
43	elmex	1.2	my $rreqcnt = scalar (keys %reqh);
44			warn "$reqcnt outstanding requests [$rreqcnt in progress]\n";
45	elmex	1.1	timer ();
46			});
47			}
48
49			sub start {
50			$J = AnyEvent->condvar;
51			timer;
52			}
53			sub wait {
54			$J->wait;
55			}
56
57			1