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