perl/Msg.pm

   1 #
   2 # This has been taken from the 'Advanced Perl Programming' book by Sriram Srinivasan
   3 #
   4 # I am presuming that the code is distributed on the same basis as perl itself.
   5 #
   6 # I have modified it to suit my devious purposes (Dirk Koopman G1TLH)
   7 #
   8 #
   9 #
  10
  11 package Msg;
  12
  13 use strict;
  14
  15 use DXUtil;
  16
  17 use IO::Select;
  18 use IO::Socket;
  19 use DXDebug;
  20 use Timer;
  21
  22 use vars qw(%rd_callbacks %wt_callbacks %er_callbacks $rd_handles $wt_handles $er_handles $now %conns $noconns $blocking_supported $cnum $total_in $total_out);
  23
  24 %rd_callbacks = ();
  25 %wt_callbacks = ();
  26 %er_callbacks = ();
  27 $rd_handles   = IO::Select->new();
  28 $wt_handles   = IO::Select->new();
  29 $er_handles   = IO::Select->new();
  30 $total_in = $total_out = 0;
  31
  32 $now = time;
  33
  34 BEGIN {
  35     # Checks if blocking is supported
  36     eval {
  37                 local $^W;
  38         require POSIX; POSIX->import(qw(O_NONBLOCK F_SETFL F_GETFL))
  39     };
  40         if ($@ || $main::is_win) {
  41                 $blocking_supported = IO::Socket->can('blocking') ? 2 : 0;
  42         } else {
  43                 $blocking_supported = IO::Socket->can('blocking') ? 2 : 1;
  44         }
  45
  46
  47         # import as many of these errno values as are available
  48         eval {
  49                 local $^W;
  50                 require Errno; Errno->import(qw(EAGAIN EINPROGRESS EWOULDBLOCK));
  51         };
  52
  53         unless ($^O eq 'MSWin32') {
  54                 if ($] >= 5.6) {
  55                         eval {
  56                                 local $^W;
  57                                 require Socket; Socket->import(qw(IPPROTO_TCP TCP_NODELAY));
  58                         };
  59                 } else {
  60                         dbg("IPPROTO_TCP and TCP_NODELAY manually defined");
  61                         eval 'sub IPPROTO_TCP {     6 };';
  62                         eval 'sub TCP_NODELAY {     1 };';
  63                 }
  64         }
  65         # http://support.microsoft.com/support/kb/articles/Q150/5/37.asp
  66         # defines EINPROGRESS as 10035.  We provide it here because some
  67         # Win32 users report POSIX::EINPROGRESS is not vendor-supported.
  68         if ($^O eq 'MSWin32') {
  69                 eval '*EINPROGRESS = sub { 10036 };';
  70                 eval '*EWOULDBLOCK = *EAGAIN = sub { 10035 };';
  71                 eval '*F_GETFL     = sub {     0 };';
  72                 eval '*F_SETFL     = sub {     0 };';
  73                 eval '*IPPROTO_TCP     = sub {     6 };';
  74                 eval '*TCP_NODELAY     = sub {     1 };';
  75                 $blocking_supported = 0;   # it appears that this DOESN'T work :-(
  76         }
  77 }
  78
  79 my $w = $^W;
  80 $^W = 0;
  81 my $eagain = eval {EAGAIN()};
  82 my $einprogress = eval {EINPROGRESS()};
  83 my $ewouldblock = eval {EWOULDBLOCK()};
  84 $^W = $w;
  85 $cnum = 0;
  86
  87
  88 #
  89 #-----------------------------------------------------------------
  90 # Generalised initializer
  91
  92 sub new
  93 {
  94     my ($pkg, $rproc) = @_;
  95         my $obj = ref($pkg);
  96         my $class = $obj || $pkg;
  97
  98     my $conn = {
  99         rproc => $rproc,
 100                 inqueue => [],
 101                 outqueue => [],
 102                 state => 0,
 103                 lineend => "\r\n",
 104                 csort => 'telnet',
 105                 timeval => 60,
 106                 blocking => 0,
 107                 cnum => (($cnum < 999) ? (++$cnum) : ($cnum = 1)),
 108     };
 109
 110         $noconns++;
 111
 112         dbg("Connection created ($noconns)") if isdbg('connll');
 113         return bless $conn, $class;
 114 }
 115
 116 sub set_error
 117 {
 118         my $conn = shift;
 119         my $callback = shift;
 120         $conn->{eproc} = $callback;
 121         set_event_handler($conn->{sock}, error => $callback) if exists $conn->{sock};
 122 }
 123
 124 sub set_rproc
 125 {
 126         my $conn = shift;
 127         my $callback = shift;
 128         $conn->{rproc} = $callback;
 129 }
 130
 131 sub blocking
 132 {
 133         return unless $blocking_supported;
 134
 135         # Make the handle stop blocking, the Windows way.
 136         if ($blocking_supported) {
 137                 $_[0]->blocking($_[1]);
 138         } else {
 139                 my $flags = fcntl ($_[0], F_GETFL, 0);
 140                 if ($_[1]) {
 141                         $flags &= ~O_NONBLOCK;
 142                 } else {
 143                         $flags |= O_NONBLOCK;
 144                 }
 145                 fcntl ($_[0], F_SETFL, $flags);
 146         }
 147 }
 148
 149 # save it
 150 sub conns
 151 {
 152         my $pkg = shift;
 153         my $call = shift;
 154         my $ref;
 155
 156         if (ref $pkg) {
 157                 $call = $pkg->{call} unless $call;
 158                 return undef unless $call;
 159                 dbg("changing $pkg->{call} to $call") if isdbg('connll') && exists $pkg->{call} && $call ne $pkg->{call};
 160                 delete $conns{$pkg->{call}} if exists $pkg->{call} && exists $conns{$pkg->{call}} && $pkg->{call} ne $call;
 161                 $pkg->{call} = $call;
 162                 $ref = $conns{$call} = $pkg;
 163                 dbg("Connection $pkg->{cnum} $call stored") if isdbg('connll');
 164         } else {
 165                 $ref = $conns{$call};
 166         }
 167         return $ref;
 168 }
 169
 170 # this is only called by any dependent processes going away unexpectedly
 171 sub pid_gone
 172 {
 173         my ($pkg, $pid) = @_;
 174
 175         my @pid = grep {$_->{pid} == $pid} values %conns;
 176         foreach my $p (@pid) {
 177                 &{$p->{eproc}}($p, "$pid has gorn") if exists $p->{eproc};
 178                 $p->disconnect;
 179         }
 180 }
 181
 182 #-----------------------------------------------------------------
 183 # Send side routines
 184 sub connect {
 185     my ($pkg, $to_host, $to_port, $rproc) = @_;
 186
 187     # Create a connection end-point object
 188     my $conn = $pkg;
 189         unless (ref $pkg) {
 190                 $conn = $pkg->new($rproc);
 191         }
 192         $conn->{peerhost} = $to_host;
 193         $conn->{peerport} = $to_port;
 194         $conn->{sort} = 'Outgoing';
 195
 196     # Create a new internet socket
 197     my $sock = IO::Socket::INET->new();
 198     return undef unless $sock;
 199
 200         my $proto = getprotobyname('tcp');
 201         $sock->socket(AF_INET, SOCK_STREAM, $proto) or return undef;
 202
 203         blocking($sock, 0);
 204         $conn->{blocking} = 0;
 205
 206         # does the host resolve?
 207         my $ip = gethostbyname($to_host);
 208         return undef unless $ip;
 209
 210         my $r = connect($sock, pack_sockaddr_in($to_port, $ip));
 211         return undef unless $r || _err_will_block($!);
 212
 213         $conn->{sock} = $sock;
 214
 215     if ($conn->{rproc}) {
 216         my $callback = sub {$conn->_rcv};
 217         set_event_handler ($sock, read => $callback);
 218     }
 219     return $conn;
 220 }
 221
 222 sub start_program
 223 {
 224         my ($conn, $line, $sort) = @_;
 225         my $pid;
 226
 227         local $^F = 10000;              # make sure it ain't closed on exec
 228         my ($a, $b) = IO::Socket->socketpair(AF_UNIX, SOCK_STREAM, PF_UNSPEC);
 229         if ($a && $b) {
 230                 $a->autoflush(1);
 231                 $b->autoflush(1);
 232                 $pid = fork;
 233                 if (defined $pid) {
 234                         if ($pid) {
 235                                 close $b;
 236                                 $conn->{sock} = $a;
 237                                 $conn->{csort} = $sort;
 238                                 $conn->{lineend} = "\cM" if $sort eq 'ax25';
 239                                 $conn->{pid} = $pid;
 240                                 if ($conn->{rproc}) {
 241                                         my $callback = sub {$conn->_rcv};
 242                                         Msg::set_event_handler ($a, read => $callback);
 243                                 }
 244                                 dbg("connect $conn->{cnum}: started pid: $conn->{pid} as $line") if isdbg('connect');
 245                         } else {
 246                                 $^W = 0;
 247                                 dbgclose();
 248                                 STDIN->close;
 249                                 STDOUT->close;
 250                                 STDOUT->close;
 251                                 *STDIN = IO::File->new_from_fd($b, 'r') or die;
 252                                 *STDOUT = IO::File->new_from_fd($b, 'w') or die;
 253                                 *STDERR = IO::File->new_from_fd($b, 'w') or die;
 254                                 close $a;
 255                                 unless ($main::is_win) {
 256                                         #                                               $SIG{HUP} = 'IGNORE';
 257                                         $SIG{HUP} = $SIG{CHLD} = $SIG{TERM} = $SIG{INT} = 'DEFAULT';
 258                                         alarm(0);
 259                                 }
 260                                 exec "$line" or dbg("exec '$line' failed $!");
 261                         }
 262                 } else {
 263                         dbg("cannot fork for $line");
 264                 }
 265         } else {
 266                 dbg("no socket pair $! for $line");
 267         }
 268         return $pid;
 269 }
 270
 271 sub disconnect
 272 {
 273     my $conn = shift;
 274         return if exists $conn->{disconnecting};
 275
 276         $conn->{disconnecting} = 1;
 277     my $sock = delete $conn->{sock};
 278         $conn->{state} = 'E';
 279         $conn->{timeout}->del if $conn->{timeout};
 280
 281         # be careful to delete the correct one
 282         my $call;
 283         if ($call = $conn->{call}) {
 284                 my $ref = $conns{$call};
 285                 delete $conns{$call} if $ref && $ref == $conn;
 286         }
 287         $call ||= 'unallocated';
 288         dbg("Connection $conn->{cnum} $call disconnected") if isdbg('connll');
 289
 290         # get rid of any references
 291         for (keys %$conn) {
 292                 if (ref($conn->{$_})) {
 293                         delete $conn->{$_};
 294                 }
 295         }
 296
 297         if (defined($sock)) {
 298                 set_event_handler ($sock, read => undef, write => undef, error => undef);
 299                 shutdown($sock, 3);
 300                 close($sock);
 301         }
 302
 303         unless ($main::is_win) {
 304                 kill 'TERM', $conn->{pid} if exists $conn->{pid};
 305         }
 306 }
 307
 308 sub send_now {
 309     my ($conn, $msg) = @_;
 310     $conn->enqueue($msg);
 311     $conn->_send (1); # 1 ==> flush
 312 }
 313
 314 sub send_later {
 315     my ($conn, $msg) = @_;
 316     $conn->enqueue($msg);
 317     my $sock = $conn->{sock};
 318     return unless defined($sock);
 319     set_event_handler ($sock, write => sub {$conn->_send(0)});
 320 }
 321
 322 sub enqueue {
 323     my $conn = shift;
 324     push (@{$conn->{outqueue}}, defined $_[0] ? $_[0] : '');
 325 }
 326
 327 sub _send {
 328     my ($conn, $flush) = @_;
 329     my $sock = $conn->{sock};
 330     return unless defined($sock);
 331     my $rq = $conn->{outqueue};
 332
 333     # If $flush is set, set the socket to blocking, and send all
 334     # messages in the queue - return only if there's an error
 335     # If $flush is 0 (deferred mode) make the socket non-blocking, and
 336     # return to the event loop only after every message, or if it
 337     # is likely to block in the middle of a message.
 338
 339 #       if ($conn->{blocking} != $flush) {
 340 #               blocking($sock, $flush);
 341 #               $conn->{blocking} = $flush;
 342 #       }
 343     my $offset = (exists $conn->{send_offset}) ? $conn->{send_offset} : 0;
 344
 345     while (@$rq) {
 346         my $msg            = $rq->[0];
 347                 my $mlth           = length($msg);
 348         my $bytes_to_write = $mlth - $offset;
 349         my $bytes_written  = 0;
 350                 confess("Negative Length! msg: '$msg' lth: $mlth offset: $offset") if $bytes_to_write < 0;
 351         while ($bytes_to_write > 0) {
 352             $bytes_written = syswrite ($sock, $msg,
 353                                        $bytes_to_write, $offset);
 354             if (!defined($bytes_written)) {
 355                 if (_err_will_block($!)) {
 356                     # Should happen only in deferred mode. Record how
 357                     # much we have already sent.
 358                     $conn->{send_offset} = $offset;
 359                     # Event handler should already be set, so we will
 360                     # be called back eventually, and will resume sending
 361                     return 1;
 362                 } else {    # Uh, oh
 363                                         &{$conn->{eproc}}($conn, $!) if exists $conn->{eproc};
 364                                         $conn->disconnect;
 365                     return 0; # fail. Message remains in queue ..
 366                 }
 367             } elsif (isdbg('raw')) {
 368                                 my $call = $conn->{call} || 'none';
 369                                 dbgdump('raw', "$call send $bytes_written: ", $msg);
 370                         }
 371                         $total_out      += $bytes_written;
 372             $offset         += $bytes_written;
 373             $bytes_to_write -= $bytes_written;
 374         }
 375         delete $conn->{send_offset};
 376         $offset = 0;
 377         shift @$rq;
 378         #last unless $flush; # Go back to select and wait
 379                             # for it to fire again.
 380     }
 381     # Call me back if queue has not been drained.
 382     unless (@$rq) {
 383         set_event_handler ($sock, write => undef);
 384                 if (exists $conn->{close_on_empty}) {
 385                         &{$conn->{eproc}}($conn, undef) if exists $conn->{eproc};
 386                         $conn->disconnect;
 387                 }
 388     }
 389     1;  # Success
 390 }
 391
 392 sub dup_sock
 393 {
 394         my $conn = shift;
 395         my $oldsock = $conn->{sock};
 396         my $rc = $rd_callbacks{$oldsock};
 397         my $wc = $wt_callbacks{$oldsock};
 398         my $ec = $er_callbacks{$oldsock};
 399         my $sock = $oldsock->new_from_fd($oldsock, "w+");
 400         if ($sock) {
 401                 set_event_handler($oldsock, read=>undef, write=>undef, error=>undef);
 402                 $conn->{sock} = $sock;
 403                 set_event_handler($sock, read=>$rc, write=>$wc, error=>$ec);
 404                 $oldsock->close;
 405         }
 406 }
 407
 408 sub _err_will_block {
 409         return 0 unless $blocking_supported;
 410         return ($_[0] == $eagain || $_[0] == $ewouldblock || $_[0] == $einprogress);
 411 }
 412
 413 sub close_on_empty
 414 {
 415         my $conn = shift;
 416         $conn->{close_on_empty} = 1;
 417 }
 418
 419 #-----------------------------------------------------------------
 420 # Receive side routines
 421
 422 sub new_server {
 423     @_ == 4 || die "Msg->new_server (myhost, myport, login_proc\n";
 424     my ($pkg, $my_host, $my_port, $login_proc) = @_;
 425         my $self = $pkg->new($login_proc);
 426
 427     $self->{sock} = IO::Socket::INET->new (
 428                                           LocalAddr => "$my_host:$my_port",
 429 #                                          LocalPort => $my_port,
 430                                           Listen    => SOMAXCONN,
 431                                           Proto     => 'tcp',
 432                                           Reuse => 1);
 433     die "Could not create socket: $! \n" unless $self->{sock};
 434     set_event_handler ($self->{sock}, read => sub { $self->new_client }  );
 435         return $self;
 436 }
 437
 438
 439 sub nolinger
 440 {
 441         my $conn = shift;
 442
 443         unless ($main::is_win) {
 444                 if (isdbg('sock')) {
 445                         my ($l, $t) = unpack "ll", getsockopt($conn->{sock}, SOL_SOCKET, SO_LINGER);
 446                         my $k = unpack 'l', getsockopt($conn->{sock}, SOL_SOCKET, SO_KEEPALIVE);
 447                         my $n = $main::is_win ? 0 : unpack "l", getsockopt($conn->{sock}, IPPROTO_TCP, TCP_NODELAY);
 448                         dbg("Linger is: $l $t, keepalive: $k, nagle: $n");
 449                 }
 450
 451                 eval {setsockopt($conn->{sock}, SOL_SOCKET, SO_KEEPALIVE, 1)} or dbg("setsockopt keepalive: $!");
 452                 eval {setsockopt($conn->{sock}, SOL_SOCKET, SO_LINGER, pack("ll", 0, 0))} or dbg("setsockopt linger: $!");
 453                 eval {setsockopt($conn->{sock}, IPPROTO_TCP, TCP_NODELAY, 1)} or eval {setsockopt($conn->{sock}, SOL_SOCKET, TCP_NODELAY, 1)} or dbg("setsockopt tcp_nodelay: $!");
 454                 $conn->{sock}->autoflush(0);
 455
 456                 if (isdbg('sock')) {
 457                         my ($l, $t) = unpack "ll", getsockopt($conn->{sock}, SOL_SOCKET, SO_LINGER);
 458                         my $k = unpack 'l', getsockopt($conn->{sock}, SOL_SOCKET, SO_KEEPALIVE);
 459                         my $n = $main::is_win ? 0 : unpack "l", getsockopt($conn->{sock}, IPPROTO_TCP, TCP_NODELAY);
 460                         dbg("Linger is: $l $t, keepalive: $k, nagle: $n");
 461                 }
 462         }
 463 }
 464
 465 sub dequeue
 466 {
 467         my $conn = shift;
 468
 469         if ($conn->{msg} =~ /\n/) {
 470                 my @lines = split /\r?\n/, $conn->{msg};
 471                 if ($conn->{msg} =~ /\n$/) {
 472                         delete $conn->{msg};
 473                 } else {
 474                         $conn->{msg} = pop @lines;
 475                 }
 476                 for (@lines) {
 477                         &{$conn->{rproc}}($conn, defined $_ ? $_ : '');
 478                 }
 479         }
 480 }
 481
 482 sub _rcv {                     # Complement to _send
 483     my $conn = shift; # $rcv_now complement of $flush
 484     # Find out how much has already been received, if at all
 485     my ($msg, $offset, $bytes_to_read, $bytes_read);
 486     my $sock = $conn->{sock};
 487     return unless defined($sock);
 488
 489         my @lines;
 490         if ($conn->{blocking}) {
 491                 blocking($sock, 0);
 492                 $conn->{blocking} = 0;
 493         }
 494         $bytes_read = sysread ($sock, $msg, 1024, 0);
 495         if (defined ($bytes_read)) {
 496                 if ($bytes_read > 0) {
 497                         $total_in += $bytes_read;
 498                         if (isdbg('raw')) {
 499                                 my $call = $conn->{call} || 'none';
 500                                 dbgdump('raw', "$call read $bytes_read: ", $msg);
 501                         }
 502                         if ($conn->{echo}) {
 503                                 my @ch = split //, $msg;
 504                                 my $out;
 505                                 for (@ch) {
 506                                         if (/[\cH\x7f]/) {
 507                                                 $out .= "\cH \cH";
 508                                                 $conn->{msg} =~ s/.$//;
 509                                         } else {
 510                                                 $out .= $_;
 511                                                 $conn->{msg} .= $_;
 512                                         }
 513                                 }
 514                                 if (defined $out) {
 515                                         set_event_handler ($sock, write => sub{$conn->_send(0)});
 516                                         push @{$conn->{outqueue}}, $out;
 517                                 }
 518                         } else {
 519                                 $conn->{msg} .= $msg;
 520                         }
 521                 }
 522         } else {
 523                 if (_err_will_block($!)) {
 524                         return ;
 525                 } else {
 526                         $bytes_read = 0;
 527                 }
 528     }
 529
 530 FINISH:
 531     if (defined $bytes_read && $bytes_read == 0) {
 532                 &{$conn->{eproc}}($conn, $!) if exists $conn->{eproc};
 533                 $conn->disconnect;
 534     } else {
 535                 unless ($conn->{disable_read}) {
 536                         $conn->dequeue if exists $conn->{msg};
 537                 }
 538         }
 539 }
 540
 541 sub new_client {
 542         my $server_conn = shift;
 543     my $sock = $server_conn->{sock}->accept();
 544         if ($sock) {
 545                 my $conn = $server_conn->new($server_conn->{rproc});
 546                 $conn->{sock} = $sock;
 547                 blocking($sock, 0);
 548                 $conn->nolinger;
 549                 $conn->{blocking} = 0;
 550                 my ($rproc, $eproc) = &{$server_conn->{rproc}} ($conn, $conn->{peerhost} = $sock->peerhost(), $conn->{peerport} = $sock->peerport());
 551                 $conn->{sort} = 'Incoming';
 552                 if ($eproc) {
 553                         $conn->{eproc} = $eproc;
 554                         set_event_handler ($sock, error => $eproc);
 555                 }
 556                 if ($rproc) {
 557                         $conn->{rproc} = $rproc;
 558                         my $callback = sub {$conn->_rcv};
 559                         set_event_handler ($sock, read => $callback);
 560                 } else {  # Login failed
 561                         &{$conn->{eproc}}($conn, undef) if exists $conn->{eproc};
 562                         $conn->disconnect();
 563                 }
 564         } else {
 565                 dbg("Msg: error on accept ($!)") if isdbg('err');
 566         }
 567 }
 568
 569 sub close_server
 570 {
 571         my $conn = shift;
 572         set_event_handler ($conn->{sock}, read => undef, write => undef, error => undef );
 573         $conn->{sock}->close;
 574 }
 575
 576 # close all clients (this is for forking really)
 577 sub close_all_clients
 578 {
 579         foreach my $conn (values %conns) {
 580                 $conn->disconnect;
 581         }
 582 }
 583
 584 sub disable_read
 585 {
 586         my $conn = shift;
 587         set_event_handler ($conn->{sock}, read => undef);
 588         return $_[0] ? $conn->{disable_read} = $_[0] : $_[0];
 589 }
 590
 591 #
 592 #----------------------------------------------------
 593 # Event loop routines used by both client and server
 594
 595 sub set_event_handler {
 596     shift unless ref($_[0]); # shift if first arg is package name
 597     my ($handle, %args) = @_;
 598     my $callback;
 599     if (exists $args{'write'}) {
 600         $callback = $args{'write'};
 601         if ($callback) {
 602             $wt_callbacks{$handle} = $callback;
 603             $wt_handles->add($handle);
 604         } else {
 605             delete $wt_callbacks{$handle};
 606             $wt_handles->remove($handle);
 607         }
 608     }
 609     if (exists $args{'read'}) {
 610         $callback = $args{'read'};
 611         if ($callback) {
 612             $rd_callbacks{$handle} = $callback;
 613             $rd_handles->add($handle);
 614         } else {
 615             delete $rd_callbacks{$handle};
 616             $rd_handles->remove($handle);
 617        }
 618     }
 619     if (exists $args{'error'}) {
 620         $callback = $args{'error'};
 621         if ($callback) {
 622             $er_callbacks{$handle} = $callback;
 623             $er_handles->add($handle);
 624         } else {
 625             delete $er_callbacks{$handle};
 626             $er_handles->remove($handle);
 627        }
 628     }
 629 }
 630
 631 sub event_loop {
 632     my ($pkg, $loop_count, $timeout, $wronly) = @_; # event_loop(1) to process events once
 633     my ($conn, $r, $w, $e, $rset, $wset, $eset);
 634     while (1) {
 635
 636        # Quit the loop if no handles left to process
 637                 if ($wronly) {
 638                         last unless $wt_handles->count();
 639
 640                         ($rset, $wset, $eset) = IO::Select->select(undef, $wt_handles, undef, $timeout);
 641
 642                         foreach $w (@$wset) {
 643                                 &{$wt_callbacks{$w}}($w) if exists $wt_callbacks{$w};
 644                         }
 645                 } else {
 646
 647                         last unless ($rd_handles->count() || $wt_handles->count());
 648
 649                         ($rset, $wset, $eset) = IO::Select->select($rd_handles, $wt_handles, $er_handles, $timeout);
 650
 651                         foreach $e (@$eset) {
 652                                 &{$er_callbacks{$e}}($e) if exists $er_callbacks{$e};
 653                         }
 654                         foreach $r (@$rset) {
 655                                 &{$rd_callbacks{$r}}($r) if exists $rd_callbacks{$r};
 656                         }
 657                         foreach $w (@$wset) {
 658                                 &{$wt_callbacks{$w}}($w) if exists $wt_callbacks{$w};
 659                         }
 660                 }
 661
 662                 Timer::handler;
 663
 664         if (defined($loop_count)) {
 665             last unless --$loop_count;
 666         }
 667     }
 668 }
 669
 670 sub sleep
 671 {
 672         my ($pkg, $interval) = @_;
 673         my $now = time;
 674         while (time - $now < $interval) {
 675                 $pkg->event_loop(10, 0.01);
 676         }
 677 }
 678
 679 sub DESTROY
 680 {
 681         my $conn = shift;
 682         my $call = $conn->{call} || 'unallocated';
 683         my $host = $conn->{peerhost} || '';
 684         my $port = $conn->{peerport} || '';
 685         dbg("Connection $conn->{cnum} $call [$host $port] being destroyed") if isdbg('connll');
 686         $noconns--;
 687 }
 688
 689 1;
 690
 691 __END__
 692