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
  18 use vars qw(%rd_callbacks %wt_callbacks $rd_handles $wt_handles $now @timerchain);
  19
  20 %rd_callbacks = ();
  21 %wt_callbacks = ();
  22 $rd_handles   = IO::Select->new();
  23 $wt_handles   = IO::Select->new();
  24 $now = time;
  25 @timerchain = ();
  26
  27 my $blocking_supported = 0;
  28
  29 BEGIN {
  30     # Checks if blocking is supported
  31     eval {
  32         require POSIX; POSIX->import(qw (F_SETFL O_NONBLOCK EAGAIN));
  33     };
  34     $blocking_supported = 1 unless $@;
  35 }
  36
  37 #
  38 #-----------------------------------------------------------------
  39 # Generalised initializer
  40
  41 sub new
  42 {
  43     my ($pkg, $rproc) = @_;
  44         my $obj = ref($pkg);
  45         my $class = $obj || $pkg;
  46
  47     my $conn = {
  48         rproc => $rproc,
  49                 inqueue => [],
  50                 outqueue => [],
  51                 state => 0,
  52                 lineend => "\r\n",
  53                 csort => 'telnet',
  54     };
  55
  56         return bless $conn, $class;
  57 }
  58
  59 #-----------------------------------------------------------------
  60 # Send side routines
  61 sub connect {
  62     my ($pkg, $to_host, $to_port, $rproc) = @_;
  63
  64     # Create a new internet socket
  65     my $sock = IO::Socket::INET->new (
  66                                       PeerAddr => $to_host,
  67                                       PeerPort => $to_port,
  68                                       Proto    => 'tcp',
  69                                       Reuse    => 1);
  70
  71     return undef unless $sock;
  72
  73     # Create a connection end-point object
  74     my $conn = $pkg;
  75         unless (ref $pkg) {
  76                 $conn = $pkg->new($rproc);
  77         }
  78         $conn->{sock} = $sock;
  79
  80     if ($conn->{rproc}) {
  81         my $callback = sub {_rcv($conn)};
  82         set_event_handler ($sock, "read" => $callback);
  83     }
  84     return $conn;
  85 }
  86
  87 sub disconnect {
  88     my $conn = shift;
  89     my $sock = delete $conn->{sock};
  90         $conn->{state} = 'E';
  91         delete $conn->{cmd};
  92         $conn->{timeout}->del_timer if $conn->{timeout};
  93         return unless defined($sock);
  94     set_event_handler ($sock, "read" => undef, "write" => undef);
  95     shutdown($sock, 3);
  96         close($sock);
  97 }
  98
  99 sub send_now {
 100     my ($conn, $msg) = @_;
 101     $conn->enqueue($msg);
 102     $conn->_send (1); # 1 ==> flush
 103 }
 104
 105 sub send_later {
 106     my ($conn, $msg) = @_;
 107     $conn->enqueue($msg);
 108     my $sock = $conn->{sock};
 109     return unless defined($sock);
 110     set_event_handler ($sock, "write" => sub {$conn->_send(0)});
 111 }
 112
 113 sub enqueue {
 114     my $conn = shift;
 115     push (@{$conn->{outqueue}}, $_[0]);
 116 }
 117
 118 sub _send {
 119     my ($conn, $flush) = @_;
 120     my $sock = $conn->{sock};
 121     return unless defined($sock);
 122     my ($rq) = $conn->{outqueue};
 123
 124     # If $flush is set, set the socket to blocking, and send all
 125     # messages in the queue - return only if there's an error
 126     # If $flush is 0 (deferred mode) make the socket non-blocking, and
 127     # return to the event loop only after every message, or if it
 128     # is likely to block in the middle of a message.
 129
 130     $flush ? $conn->set_blocking() : $conn->set_non_blocking();
 131     my $offset = (exists $conn->{send_offset}) ? $conn->{send_offset} : 0;
 132
 133     while (@$rq) {
 134         my $msg            = $rq->[0];
 135                 my $mlth           = length($msg);
 136         my $bytes_to_write = $mlth - $offset;
 137         my $bytes_written  = 0;
 138                 confess("Negative Length! msg: '$msg' lth: $mlth offset: $offset") if $bytes_to_write < 0;
 139         while ($bytes_to_write > 0) {
 140             $bytes_written = syswrite ($sock, $msg,
 141                                        $bytes_to_write, $offset);
 142             if (!defined($bytes_written)) {
 143                 if (_err_will_block($!)) {
 144                     # Should happen only in deferred mode. Record how
 145                     # much we have already sent.
 146                     $conn->{send_offset} = $offset;
 147                     # Event handler should already be set, so we will
 148                     # be called back eventually, and will resume sending
 149                     return 1;
 150                 } else {    # Uh, oh
 151                                         delete $conn->{send_offset};
 152                     $conn->handle_send_err($!);
 153                                         $conn->disconnect;
 154                     return 0; # fail. Message remains in queue ..
 155                 }
 156             }
 157             $offset         += $bytes_written;
 158             $bytes_to_write -= $bytes_written;
 159         }
 160         delete $conn->{send_offset};
 161         $offset = 0;
 162         shift @$rq;
 163         last unless $flush; # Go back to select and wait
 164                             # for it to fire again.
 165     }
 166     # Call me back if queue has not been drained.
 167     if (@$rq) {
 168         set_event_handler ($sock, "write" => sub {$conn->_send(0)});
 169     } else {
 170         set_event_handler ($sock, "write" => undef);
 171     }
 172     1;  # Success
 173 }
 174
 175 sub _err_will_block {
 176     if ($blocking_supported) {
 177         return ($_[0] == EAGAIN());
 178     }
 179     return 0;
 180 }
 181 sub set_non_blocking {                        # $conn->set_blocking
 182     if ($blocking_supported) {
 183         # preserve other fcntl flags
 184         my $flags = fcntl ($_[0], F_GETFL(), 0);
 185         fcntl ($_[0], F_SETFL(), $flags | O_NONBLOCK());
 186     }
 187 }
 188 sub set_blocking {
 189     if ($blocking_supported) {
 190         my $flags = fcntl ($_[0], F_GETFL(), 0);
 191         $flags  &= ~O_NONBLOCK(); # Clear blocking, but preserve other flags
 192         fcntl ($_[0], F_SETFL(), $flags);
 193     }
 194 }
 195
 196 sub handle_send_err {
 197    # For more meaningful handling of send errors, subclass Msg and
 198    # rebless $conn.
 199    my ($conn, $err_msg) = @_;
 200    warn "Error while sending: $err_msg \n";
 201    set_event_handler ($conn->{sock}, "write" => undef);
 202 }
 203
 204 #-----------------------------------------------------------------
 205 # Receive side routines
 206
 207 sub new_server {
 208     @_ == 4 || die "Msg->new_server (myhost, myport, login_proc\n";
 209     my ($pkg, $my_host, $my_port, $login_proc) = @_;
 210         my $self = $pkg->new($login_proc);
 211
 212     $self->{sock} = IO::Socket::INET->new (
 213                                           LocalAddr => $my_host,
 214                                           LocalPort => $my_port,
 215                                           Listen    => 5,
 216                                           Proto     => 'tcp',
 217                                           Reuse     => 1);
 218     die "Could not create socket: $! \n" unless $self->{sock};
 219     set_event_handler ($self->{sock}, "read" => sub { $self->new_client }  );
 220         return $self;
 221 }
 222
 223 sub dequeue
 224 {
 225         my $conn = shift;
 226         my $msg;
 227
 228         while ($msg = shift @{$conn->{inqueue}}){
 229                 &{$conn->{rproc}}($conn, $msg, $!);
 230                 $! = 0;
 231         }
 232 }
 233
 234 sub _rcv {                     # Complement to _send
 235     my $conn = shift; # $rcv_now complement of $flush
 236     # Find out how much has already been received, if at all
 237     my ($msg, $offset, $bytes_to_read, $bytes_read);
 238     my $sock = $conn->{sock};
 239     return unless defined($sock);
 240
 241         my @lines;
 242     $conn->set_non_blocking();
 243         $bytes_read = sysread ($sock, $msg, 1024, 0);
 244         if (defined ($bytes_read)) {
 245                 if ($bytes_read > 0) {
 246                         if ($msg =~ /\n/) {
 247                                 @lines = split /\r?\n/, $msg;
 248                                 $lines[0] = '' unless @lines;
 249                                 $lines[0] = $conn->{msg} . $lines[0] if exists $conn->{msg};
 250                                 push @lines, ' ' unless @lines;
 251                                 if ($msg =~ /\n$/) {
 252                                         delete $conn->{msg};
 253                                 } else {
 254                                         $conn->{msg} = pop @lines;
 255                                 }
 256                                 push @{$conn->{inqueue}}, @lines if @lines;
 257                         } else {
 258                                 $conn->{msg} .= $msg;
 259                         }
 260                 }
 261         } else {
 262                 if (_err_will_block($!)) {
 263                         return ;
 264                 } else {
 265                         $bytes_read = 0;
 266                 }
 267     }
 268
 269 FINISH:
 270     if (defined $bytes_read && $bytes_read == 0) {
 271 #               $conn->disconnect();
 272                 &{$conn->{rproc}}($conn, undef, $!);
 273                 delete $conn->{inqueue};
 274     } else {
 275                 $conn->dequeue;
 276         }
 277 }
 278
 279 sub new_client {
 280         my $server_conn = shift;
 281     my $sock = $server_conn->{sock}->accept();
 282     my $conn = $server_conn->new($server_conn->{rproc});
 283         $conn->{sock} = $sock;
 284     my $rproc = &{$server_conn->{rproc}} ($conn, $sock->peerhost(), $sock->peerport());
 285     if ($rproc) {
 286         $conn->{rproc} = $rproc;
 287         my $callback = sub {_rcv($conn)};
 288         set_event_handler ($sock, "read" => $callback);
 289     } else {  # Login failed
 290         $conn->disconnect();
 291     }
 292 }
 293
 294 sub close_server
 295 {
 296         my $conn = shift;
 297         set_event_handler ($conn->{sock}, "read" => undef);
 298         $conn->{sock}->close;
 299 }
 300
 301 #----------------------------------------------------
 302 # Event loop routines used by both client and server
 303
 304 sub set_event_handler {
 305     shift unless ref($_[0]); # shift if first arg is package name
 306     my ($handle, %args) = @_;
 307     my $callback;
 308     if (exists $args{'write'}) {
 309         $callback = $args{'write'};
 310         if ($callback) {
 311             $wt_callbacks{$handle} = $callback;
 312             $wt_handles->add($handle);
 313         } else {
 314             delete $wt_callbacks{$handle};
 315             $wt_handles->remove($handle);
 316         }
 317     }
 318     if (exists $args{'read'}) {
 319         $callback = $args{'read'};
 320         if ($callback) {
 321             $rd_callbacks{$handle} = $callback;
 322             $rd_handles->add($handle);
 323         } else {
 324             delete $rd_callbacks{$handle};
 325             $rd_handles->remove($handle);
 326        }
 327     }
 328 }
 329
 330 sub new_timer
 331 {
 332     my ($pkg, $time, $proc, $recur) = @_;
 333         my $obj = ref($pkg);
 334         my $class = $obj || $pkg;
 335         my $self = bless { t=>$time + time, proc=>$proc }, $class;
 336         $self->{interval} = $time if $recur;
 337         push @timerchain, $self;
 338         return $self;
 339 }
 340
 341 sub del_timer
 342 {
 343         my $self = shift;
 344         @timerchain = grep {$_ != $self} @timerchain;
 345 }
 346
 347 sub event_loop {
 348     my ($pkg, $loop_count, $timeout) = @_; # event_loop(1) to process events once
 349     my ($conn, $r, $w, $rset, $wset);
 350     while (1) {
 351
 352        # Quit the loop if no handles left to process
 353         last unless ($rd_handles->count() || $wt_handles->count());
 354
 355                 ($rset, $wset) =
 356             IO::Select->select ($rd_handles, $wt_handles, undef, $timeout);
 357                 $now = time;
 358
 359         foreach $r (@$rset) {
 360             &{$rd_callbacks{$r}} ($r) if exists $rd_callbacks{$r};
 361         }
 362         foreach $w (@$wset) {
 363             &{$wt_callbacks{$w}}($w) if exists $wt_callbacks{$w};
 364         }
 365
 366                 # handle things on the timer chain
 367                 for (@timerchain) {
 368                         if ($now >= $_->{t}) {
 369                                 &{$_->{proc}}();
 370                                 $_->{t} = $now + $_->{interval} if exists $_->{interval};
 371                         }
 372                 }
 373
 374                 # remove dead timers
 375                 @timerchain = grep { $_->{t} > $now } @timerchain;
 376
 377         if (defined($loop_count)) {
 378             last unless --$loop_count;
 379         }
 380     }
 381 }
 382
 383 1;
 384
 385 __END__
 386