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 use IO::Select;
  15 use IO::Socket;
  16 use DXDebug;
  17 use Timer;
  18
  19 use vars qw(%rd_callbacks %wt_callbacks %er_callbacks $rd_handles $wt_handles $er_handles $now %conns $noconns);
  20
  21 %rd_callbacks = ();
  22 %wt_callbacks = ();
  23 %er_callbacks = ();
  24 $rd_handles   = IO::Select->new();
  25 $wt_handles   = IO::Select->new();
  26 $er_handles   = IO::Select->new();
  27
  28 $now = time;
  29 my $blocking_supported = 0;
  30
  31 BEGIN {
  32     # Checks if blocking is supported
  33     eval {
  34         require POSIX; POSIX->import(qw (F_SETFL F_GETFL O_NONBLOCK));
  35     };
  36     $blocking_supported = 1 unless $@;
  37
  38         # import as many of these errno values as are available
  39         eval {
  40                 require Errno; Errno->import(qw(EAGAIN EINPROGRESS EWOULDBLOCK));
  41         };
  42 }
  43
  44 my $w = $^W;
  45 $^W = 0;
  46 my $eagain = eval {EAGAIN()};
  47 my $einprogress = eval {EINPROGRESS()};
  48 my $ewouldblock = eval {EWOULDBLOCK()};
  49 $^W = $w;
  50
  51 #
  52 #-----------------------------------------------------------------
  53 # Generalised initializer
  54
  55 sub new
  56 {
  57     my ($pkg, $rproc) = @_;
  58         my $obj = ref($pkg);
  59         my $class = $obj || $pkg;
  60
  61     my $conn = {
  62         rproc => $rproc,
  63                 inqueue => [],
  64                 outqueue => [],
  65                 state => 0,
  66                 lineend => "\r\n",
  67                 csort => 'telnet',
  68                 timeval => 60,
  69     };
  70
  71         $noconns++;
  72         dbg('connll', "Connection created ($noconns)");
  73         return bless $conn, $class;
  74 }
  75
  76 sub set_error
  77 {
  78         my $conn = shift;
  79         my $callback = shift;
  80         $conn->{eproc} = $callback;
  81         set_event_handler($conn->{sock}, error => $callback) if exists $conn->{sock};
  82 }
  83
  84 sub set_rproc
  85 {
  86         my $conn = shift;
  87         my $callback = shift;
  88         $conn->{rproc} = $callback;
  89 }
  90
  91 sub blocking
  92 {
  93         return unless $blocking_supported;
  94
  95         my $flags = fcntl ($_[0], F_GETFL, 0);
  96         if ($_[1]) {
  97                 $flags &= ~O_NONBLOCK;
  98         } else {
  99                 $flags |= O_NONBLOCK;
 100         }
 101         fcntl ($_[0], F_SETFL, $flags);
 102 }
 103
 104 # save it
 105 sub conns
 106 {
 107         my $pkg = shift;
 108         my $call = shift;
 109         my $ref;
 110
 111         if (ref $pkg) {
 112                 $call = $pkg->{call} unless $call;
 113                 return undef unless $call;
 114                 confess "changing $pkg->{call} to $call" if exists $pkg->{call} && $call ne $pkg->{call};
 115                 $pkg->{call} = $call;
 116                 $ref = $conns{$call} = $pkg;
 117                 dbg('connll', "Connection $call stored");
 118         } else {
 119                 $ref = $conns{$call};
 120         }
 121         return $ref;
 122 }
 123
 124 # this is only called by any dependent processes going away unexpectedly
 125 sub pid_gone
 126 {
 127         my ($pkg, $pid) = @_;
 128
 129         my @pid = grep {$_->{pid} == $pid} values %conns;
 130         for (@pid) {
 131                 &{$_->{eproc}}($_, "$pid has gorn") if exists $_->{eproc};
 132                 $_->disconnect;
 133         }
 134 }
 135
 136 #-----------------------------------------------------------------
 137 # Send side routines
 138 sub connect {
 139     my ($pkg, $to_host, $to_port, $rproc) = @_;
 140
 141     # Create a connection end-point object
 142     my $conn = $pkg;
 143         unless (ref $pkg) {
 144                 $conn = $pkg->new($rproc);
 145         }
 146         $conn->{peerhost} = $to_host;
 147         $conn->{peerport} = $to_port;
 148         $conn->{sort} = 'Outgoing';
 149
 150     # Create a new internet socket
 151     my $sock = IO::Socket::INET->new();
 152     return undef unless $sock;
 153
 154         my $proto = getprotobyname('tcp');
 155         $sock->socket(AF_INET, SOCK_STREAM, $proto) or return undef;
 156
 157         blocking($sock, 0);
 158         my $ip = gethostbyname($to_host);
 159 #       my $r = $sock->connect($to_port, $ip);
 160         my $r = connect($sock, pack_sockaddr_in($to_port, $ip));
 161         return undef unless $r || _err_will_block($!);
 162
 163         $conn->{sock} = $sock;
 164
 165     if ($conn->{rproc}) {
 166         my $callback = sub {$conn->_rcv};
 167         set_event_handler ($sock, read => $callback);
 168     }
 169     return $conn;
 170 }
 171
 172 sub disconnect {
 173     my $conn = shift;
 174         return if exists $conn->{disconnecting};
 175
 176         $conn->{disconnecting} = 1;
 177     my $sock = delete $conn->{sock};
 178         $conn->{state} = 'E';
 179         $conn->{timeout}->del if $conn->{timeout};
 180
 181         # be careful to delete the correct one
 182         my $call;
 183         if ($call = $conn->{call}) {
 184                 my $ref = $conns{$call};
 185                 delete $conns{$call} if $ref && $ref == $conn;
 186         }
 187         $call ||= 'unallocated';
 188         dbg('connll', "Connection $call disconnected");
 189
 190         unless ($^O =~ /^MS/i) {
 191                 kill 'TERM', $conn->{pid} if exists $conn->{pid};
 192         }
 193
 194         # get rid of any references
 195         for (keys %$conn) {
 196                 if (ref($conn->{$_})) {
 197                         delete $conn->{$_};
 198                 }
 199         }
 200
 201         return unless defined($sock);
 202     set_event_handler ($sock, read => undef, write => undef, error => undef);
 203     shutdown($sock, 3);
 204         close($sock);
 205 }
 206
 207 sub send_now {
 208     my ($conn, $msg) = @_;
 209     $conn->enqueue($msg);
 210     $conn->_send (1); # 1 ==> flush
 211 }
 212
 213 sub send_later {
 214     my ($conn, $msg) = @_;
 215     $conn->enqueue($msg);
 216     my $sock = $conn->{sock};
 217     return unless defined($sock);
 218     set_event_handler ($sock, write => sub {$conn->_send(0)});
 219 }
 220
 221 sub enqueue {
 222     my $conn = shift;
 223     push (@{$conn->{outqueue}}, defined $_[0] ? $_[0] : '');
 224 }
 225
 226 sub _send {
 227     my ($conn, $flush) = @_;
 228     my $sock = $conn->{sock};
 229     return unless defined($sock);
 230     my $rq = $conn->{outqueue};
 231
 232     # If $flush is set, set the socket to blocking, and send all
 233     # messages in the queue - return only if there's an error
 234     # If $flush is 0 (deferred mode) make the socket non-blocking, and
 235     # return to the event loop only after every message, or if it
 236     # is likely to block in the middle of a message.
 237
 238         blocking($sock, $flush);
 239     my $offset = (exists $conn->{send_offset}) ? $conn->{send_offset} : 0;
 240
 241     while (@$rq) {
 242         my $msg            = $rq->[0];
 243                 my $mlth           = length($msg);
 244         my $bytes_to_write = $mlth - $offset;
 245         my $bytes_written  = 0;
 246                 confess("Negative Length! msg: '$msg' lth: $mlth offset: $offset") if $bytes_to_write < 0;
 247         while ($bytes_to_write > 0) {
 248             $bytes_written = syswrite ($sock, $msg,
 249                                        $bytes_to_write, $offset);
 250             if (!defined($bytes_written)) {
 251                 if (_err_will_block($!)) {
 252                     # Should happen only in deferred mode. Record how
 253                     # much we have already sent.
 254                     $conn->{send_offset} = $offset;
 255                     # Event handler should already be set, so we will
 256                     # be called back eventually, and will resume sending
 257                     return 1;
 258                 } else {    # Uh, oh
 259                                         &{$conn->{eproc}}($conn, $!) if exists $conn->{eproc};
 260                                         $conn->disconnect;
 261                     return 0; # fail. Message remains in queue ..
 262                 }
 263             }
 264             $offset         += $bytes_written;
 265             $bytes_to_write -= $bytes_written;
 266         }
 267         delete $conn->{send_offset};
 268         $offset = 0;
 269         shift @$rq;
 270         last unless $flush; # Go back to select and wait
 271                             # for it to fire again.
 272     }
 273     # Call me back if queue has not been drained.
 274     if (@$rq) {
 275         set_event_handler ($sock, write => sub {$conn->_send(0)});
 276     } else {
 277         set_event_handler ($sock, write => undef);
 278                 if (exists $conn->{close_on_empty}) {
 279                         &{$conn->{eproc}}($conn, undef) if exists $conn->{eproc};
 280                         $conn->disconnect;
 281                 }
 282     }
 283     1;  # Success
 284 }
 285
 286 sub dup_sock
 287 {
 288         my $conn = shift;
 289         my $oldsock = $conn->{sock};
 290         my $rc = $rd_callbacks{$oldsock};
 291         my $wc = $wt_callbacks{$oldsock};
 292         my $ec = $er_callbacks{$oldsock};
 293         my $sock = $oldsock->new_from_fd($oldsock, "w+");
 294         if ($sock) {
 295                 set_event_handler($oldsock, read=>undef, write=>undef, error=>undef);
 296                 $conn->{sock} = $sock;
 297                 set_event_handler($sock, read=>$rc, write=>$wc, error=>$ec);
 298                 $oldsock->close;
 299         }
 300 }
 301
 302 sub _err_will_block {
 303         return 0 unless $blocking_supported;
 304         return ($_[0] == $eagain || $_[0] == $ewouldblock || $_[0] == $einprogress);
 305 }
 306
 307 sub close_on_empty
 308 {
 309         my $conn = shift;
 310         $conn->{close_on_empty} = 1;
 311 }
 312
 313 #-----------------------------------------------------------------
 314 # Receive side routines
 315
 316 sub new_server {
 317     @_ == 4 || die "Msg->new_server (myhost, myport, login_proc\n";
 318     my ($pkg, $my_host, $my_port, $login_proc) = @_;
 319         my $self = $pkg->new($login_proc);
 320
 321     $self->{sock} = IO::Socket::INET->new (
 322                                           LocalAddr => $my_host,
 323                                           LocalPort => $my_port,
 324                                           Listen    => SOMAXCONN,
 325                                           Proto     => 'tcp',
 326                                           Reuse     => 1);
 327     die "Could not create socket: $! \n" unless $self->{sock};
 328     set_event_handler ($self->{sock}, read => sub { $self->new_client }  );
 329         return $self;
 330 }
 331
 332 sub dequeue
 333 {
 334         my $conn = shift;
 335
 336         if ($conn->{msg} =~ /\n/) {
 337                 my @lines = split /\r?\n/, $conn->{msg};
 338                 if ($conn->{msg} =~ /\n$/) {
 339                         delete $conn->{msg};
 340                 } else {
 341                         $conn->{msg} = pop @lines;
 342                 }
 343                 for (@lines) {
 344                         &{$conn->{rproc}}($conn, defined $_ ? $_ : '');
 345                 }
 346         }
 347 }
 348
 349 sub _rcv {                     # Complement to _send
 350     my $conn = shift; # $rcv_now complement of $flush
 351     # Find out how much has already been received, if at all
 352     my ($msg, $offset, $bytes_to_read, $bytes_read);
 353     my $sock = $conn->{sock};
 354     return unless defined($sock);
 355
 356         my @lines;
 357         blocking($sock, 0);
 358         $bytes_read = sysread ($sock, $msg, 1024, 0);
 359         if (defined ($bytes_read)) {
 360                 if ($bytes_read > 0) {
 361                         $conn->{msg} .= $msg;
 362                 }
 363         } else {
 364                 if (_err_will_block($!)) {
 365                         return ;
 366                 } else {
 367                         $bytes_read = 0;
 368                 }
 369     }
 370
 371 FINISH:
 372     if (defined $bytes_read && $bytes_read == 0) {
 373                 &{$conn->{eproc}}($conn, $!) if exists $conn->{eproc};
 374                 $conn->disconnect;
 375     } else {
 376                 $conn->dequeue if exists $conn->{msg};
 377         }
 378 }
 379
 380 sub new_client {
 381         my $server_conn = shift;
 382     my $sock = $server_conn->{sock}->accept();
 383     my $conn = $server_conn->new($server_conn->{rproc});
 384         $conn->{sock} = $sock;
 385     my ($rproc, $eproc) = &{$server_conn->{rproc}} ($conn, $conn->{peerhost} = $sock->peerhost(), $conn->{peerport} = $sock->peerport());
 386         $conn->{sort} = 'Incoming';
 387         if ($eproc) {
 388                 $conn->{eproc} = $eproc;
 389         set_event_handler ($sock, error => $eproc);
 390         }
 391     if ($rproc) {
 392         $conn->{rproc} = $rproc;
 393         my $callback = sub {$conn->_rcv};
 394         set_event_handler ($sock, read => $callback);
 395     } else {  # Login failed
 396                 &{$conn->{eproc}}($conn, undef) if exists $conn->{eproc};
 397         $conn->disconnect();
 398     }
 399 }
 400
 401 sub close_server
 402 {
 403         my $conn = shift;
 404         set_event_handler ($conn->{sock}, read => undef, write => undef, error => undef );
 405         $conn->{sock}->close;
 406 }
 407
 408 # close all clients (this is for forking really)
 409 sub close_all_clients
 410 {
 411         for (values %conns) {
 412                 $_->disconnect;
 413         }
 414 }
 415
 416 #----------------------------------------------------
 417 # Event loop routines used by both client and server
 418
 419 sub set_event_handler {
 420     shift unless ref($_[0]); # shift if first arg is package name
 421     my ($handle, %args) = @_;
 422     my $callback;
 423     if (exists $args{'write'}) {
 424         $callback = $args{'write'};
 425         if ($callback) {
 426             $wt_callbacks{$handle} = $callback;
 427             $wt_handles->add($handle);
 428         } else {
 429             delete $wt_callbacks{$handle};
 430             $wt_handles->remove($handle);
 431         }
 432     }
 433     if (exists $args{'read'}) {
 434         $callback = $args{'read'};
 435         if ($callback) {
 436             $rd_callbacks{$handle} = $callback;
 437             $rd_handles->add($handle);
 438         } else {
 439             delete $rd_callbacks{$handle};
 440             $rd_handles->remove($handle);
 441        }
 442     }
 443     if (exists $args{'error'}) {
 444         $callback = $args{'error'};
 445         if ($callback) {
 446             $er_callbacks{$handle} = $callback;
 447             $er_handles->add($handle);
 448         } else {
 449             delete $er_callbacks{$handle};
 450             $er_handles->remove($handle);
 451        }
 452     }
 453 }
 454
 455 sub event_loop {
 456     my ($pkg, $loop_count, $timeout) = @_; # event_loop(1) to process events once
 457     my ($conn, $r, $w, $e, $rset, $wset, $eset);
 458     while (1) {
 459
 460        # Quit the loop if no handles left to process
 461         last unless ($rd_handles->count() || $wt_handles->count());
 462
 463                 ($rset, $wset) = IO::Select->select($rd_handles, $wt_handles, $er_handles, $timeout);
 464
 465         foreach $e (@$eset) {
 466             &{$er_callbacks{$e}}($e) if exists $er_callbacks{$e};
 467         }
 468         foreach $r (@$rset) {
 469             &{$rd_callbacks{$r}}($r) if exists $rd_callbacks{$r};
 470         }
 471         foreach $w (@$wset) {
 472             &{$wt_callbacks{$w}}($w) if exists $wt_callbacks{$w};
 473         }
 474
 475                 Timer::handler;
 476
 477         if (defined($loop_count)) {
 478             last unless --$loop_count;
 479         }
 480     }
 481 }
 482
 483 sub DESTROY
 484 {
 485         my $conn = shift;
 486         my $call = $conn->{call} || 'unallocated';
 487         dbg('connll', "Connection $call being destroyed ($noconns)");
 488         $noconns--;
 489 }
 490
 491 1;
 492
 493 __END__
 494