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