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 _err_will_block {
 287         return 0 unless $blocking_supported;
 288         return ($_[0] == $eagain || $_[0] == $ewouldblock || $_[0] == $einprogress);
 289 }
 290
 291 sub close_on_empty
 292 {
 293         my $conn = shift;
 294         $conn->{close_on_empty} = 1;
 295 }
 296
 297 #-----------------------------------------------------------------
 298 # Receive side routines
 299
 300 sub new_server {
 301     @_ == 4 || die "Msg->new_server (myhost, myport, login_proc\n";
 302     my ($pkg, $my_host, $my_port, $login_proc) = @_;
 303         my $self = $pkg->new($login_proc);
 304
 305     $self->{sock} = IO::Socket::INET->new (
 306                                           LocalAddr => $my_host,
 307                                           LocalPort => $my_port,
 308                                           Listen    => SOMAXCONN,
 309                                           Proto     => 'tcp',
 310                                           Reuse     => 1);
 311     die "Could not create socket: $! \n" unless $self->{sock};
 312     set_event_handler ($self->{sock}, read => sub { $self->new_client }  );
 313         return $self;
 314 }
 315
 316 sub dequeue
 317 {
 318         my $conn = shift;
 319
 320         if ($conn->{msg} =~ /\n/) {
 321                 my @lines = split /\r?\n/, $conn->{msg};
 322                 if ($conn->{msg} =~ /\n$/) {
 323                         delete $conn->{msg};
 324                 } else {
 325                         $conn->{msg} = pop @lines;
 326                 }
 327                 for (@lines) {
 328                         &{$conn->{rproc}}($conn, defined $_ ? $_ : '');
 329                 }
 330         }
 331 }
 332
 333 sub _rcv {                     # Complement to _send
 334     my $conn = shift; # $rcv_now complement of $flush
 335     # Find out how much has already been received, if at all
 336     my ($msg, $offset, $bytes_to_read, $bytes_read);
 337     my $sock = $conn->{sock};
 338     return unless defined($sock);
 339
 340         my @lines;
 341         blocking($sock, 0);
 342         $bytes_read = sysread ($sock, $msg, 1024, 0);
 343         if (defined ($bytes_read)) {
 344                 if ($bytes_read > 0) {
 345                         $conn->{msg} .= $msg;
 346                 }
 347         } else {
 348                 if (_err_will_block($!)) {
 349                         return ;
 350                 } else {
 351                         $bytes_read = 0;
 352                 }
 353     }
 354
 355 FINISH:
 356     if (defined $bytes_read && $bytes_read == 0) {
 357                 &{$conn->{eproc}}($conn, $!) if exists $conn->{eproc};
 358                 $conn->disconnect;
 359     } else {
 360                 $conn->dequeue if exists $conn->{msg};
 361         }
 362 }
 363
 364 sub new_client {
 365         my $server_conn = shift;
 366     my $sock = $server_conn->{sock}->accept();
 367     my $conn = $server_conn->new($server_conn->{rproc});
 368         $conn->{sock} = $sock;
 369     my ($rproc, $eproc) = &{$server_conn->{rproc}} ($conn, $conn->{peerhost} = $sock->peerhost(), $conn->{peerport} = $sock->peerport());
 370         $conn->{sort} = 'Incoming';
 371         if ($eproc) {
 372                 $conn->{eproc} = $eproc;
 373         set_event_handler ($sock, error => $eproc);
 374         }
 375     if ($rproc) {
 376         $conn->{rproc} = $rproc;
 377         my $callback = sub {$conn->_rcv};
 378         set_event_handler ($sock, read => $callback);
 379     } else {  # Login failed
 380                 &{$conn->{eproc}}($conn, undef) if exists $conn->{eproc};
 381         $conn->disconnect();
 382     }
 383 }
 384
 385 sub close_server
 386 {
 387         my $conn = shift;
 388         set_event_handler ($conn->{sock}, read => undef, write => undef, error => undef );
 389         $conn->{sock}->close;
 390 }
 391
 392 # close all clients (this is for forking really)
 393 sub close_all_clients
 394 {
 395         for (values %conns) {
 396                 $_->disconnect;
 397         }
 398 }
 399
 400 #----------------------------------------------------
 401 # Event loop routines used by both client and server
 402
 403 sub set_event_handler {
 404     shift unless ref($_[0]); # shift if first arg is package name
 405     my ($handle, %args) = @_;
 406     my $callback;
 407     if (exists $args{'write'}) {
 408         $callback = $args{'write'};
 409         if ($callback) {
 410             $wt_callbacks{$handle} = $callback;
 411             $wt_handles->add($handle);
 412         } else {
 413             delete $wt_callbacks{$handle};
 414             $wt_handles->remove($handle);
 415         }
 416     }
 417     if (exists $args{'read'}) {
 418         $callback = $args{'read'};
 419         if ($callback) {
 420             $rd_callbacks{$handle} = $callback;
 421             $rd_handles->add($handle);
 422         } else {
 423             delete $rd_callbacks{$handle};
 424             $rd_handles->remove($handle);
 425        }
 426     }
 427     if (exists $args{'error'}) {
 428         $callback = $args{'error'};
 429         if ($callback) {
 430             $er_callbacks{$handle} = $callback;
 431             $er_handles->add($handle);
 432         } else {
 433             delete $er_callbacks{$handle};
 434             $er_handles->remove($handle);
 435        }
 436     }
 437 }
 438
 439 sub event_loop {
 440     my ($pkg, $loop_count, $timeout) = @_; # event_loop(1) to process events once
 441     my ($conn, $r, $w, $e, $rset, $wset, $eset);
 442     while (1) {
 443
 444        # Quit the loop if no handles left to process
 445         last unless ($rd_handles->count() || $wt_handles->count());
 446
 447                 ($rset, $wset) = IO::Select->select($rd_handles, $wt_handles, $er_handles, $timeout);
 448
 449         foreach $e (@$eset) {
 450             &{$er_callbacks{$e}}($e) if exists $er_callbacks{$e};
 451         }
 452         foreach $r (@$rset) {
 453             &{$rd_callbacks{$r}}($r) if exists $rd_callbacks{$r};
 454         }
 455         foreach $w (@$wset) {
 456             &{$wt_callbacks{$w}}($w) if exists $wt_callbacks{$w};
 457         }
 458
 459                 Timer::handler;
 460
 461         if (defined($loop_count)) {
 462             last unless --$loop_count;
 463         }
 464     }
 465 }
 466
 467 sub DESTROY
 468 {
 469         my $conn = shift;
 470         my $call = $conn->{call} || 'unallocated';
 471         dbg('connll', "Connection $call being destroyed ($noconns)");
 472         $noconns--;
 473 }
 474
 475 1;
 476
 477 __END__
 478