2 # This has been taken from the 'Advanced Perl Programming' book by Sriram Srinivasan
4 # I am presuming that the code is distributed on the same basis as perl itself.
6 # I have modified it to suit my devious purposes (Dirk Koopman G1TLH)
18 use vars qw(%rd_callbacks %wt_callbacks $rd_handles $wt_handles $now @timerchain);
22 $rd_handles = IO::Select->new();
23 $wt_handles = IO::Select->new();
27 my $blocking_supported = 0;
30 # Checks if blocking is supported
32 require POSIX; POSIX->import(qw (F_SETFL O_NONBLOCK EAGAIN));
34 $blocking_supported = 1 unless $@;
38 #-----------------------------------------------------------------
39 # Generalised initializer
43 my ($pkg, $rproc) = @_;
45 my $class = $obj || $pkg;
55 return bless $conn, $class;
58 #-----------------------------------------------------------------
61 my ($pkg, $to_host, $to_port, $rproc) = @_;
63 # Create a new internet socket
64 my $sock = IO::Socket::INET->new (
70 return undef unless $sock;
72 # Create a connection end-point object
75 $conn = $pkg->new($rproc);
77 $conn->{sock} = $sock;
80 my $callback = sub {_rcv($conn)};
81 set_event_handler ($sock, "read" => $callback);
88 my $sock = delete $conn->{sock};
91 $conn->{timeout}->del_timer if $conn->{timeout};
92 return unless defined($sock);
93 set_event_handler ($sock, "read" => undef, "write" => undef);
99 my ($conn, $msg) = @_;
100 $conn->enqueue($msg);
101 $conn->_send (1); # 1 ==> flush
105 my ($conn, $msg) = @_;
106 $conn->enqueue($msg);
107 my $sock = $conn->{sock};
108 return unless defined($sock);
109 set_event_handler ($sock, "write" => sub {$conn->_send(0)});
114 push (@{$conn->{outqueue}}, $_[0]);
118 my ($conn, $flush) = @_;
119 my $sock = $conn->{sock};
120 return unless defined($sock);
121 my ($rq) = $conn->{outqueue};
123 # If $flush is set, set the socket to blocking, and send all
124 # messages in the queue - return only if there's an error
125 # If $flush is 0 (deferred mode) make the socket non-blocking, and
126 # return to the event loop only after every message, or if it
127 # is likely to block in the middle of a message.
129 $flush ? $conn->set_blocking() : $conn->set_non_blocking();
130 my $offset = (exists $conn->{send_offset}) ? $conn->{send_offset} : 0;
134 my $mlth = length($msg);
135 my $bytes_to_write = $mlth - $offset;
136 my $bytes_written = 0;
137 confess("Negative Length! msg: '$msg' lth: $mlth offset: $offset") if $bytes_to_write < 0;
138 while ($bytes_to_write > 0) {
139 $bytes_written = syswrite ($sock, $msg,
140 $bytes_to_write, $offset);
141 if (!defined($bytes_written)) {
142 if (_err_will_block($!)) {
143 # Should happen only in deferred mode. Record how
144 # much we have already sent.
145 $conn->{send_offset} = $offset;
146 # Event handler should already be set, so we will
147 # be called back eventually, and will resume sending
150 delete $conn->{send_offset};
151 $conn->handle_send_err($!);
153 return 0; # fail. Message remains in queue ..
156 $offset += $bytes_written;
157 $bytes_to_write -= $bytes_written;
159 delete $conn->{send_offset};
162 last unless $flush; # Go back to select and wait
163 # for it to fire again.
165 # Call me back if queue has not been drained.
167 set_event_handler ($sock, "write" => sub {$conn->_send(0)});
169 set_event_handler ($sock, "write" => undef);
174 sub _err_will_block {
175 if ($blocking_supported) {
176 return ($_[0] == EAGAIN());
180 sub set_non_blocking { # $conn->set_blocking
181 if ($blocking_supported) {
182 # preserve other fcntl flags
183 my $flags = fcntl ($_[0], F_GETFL(), 0);
184 fcntl ($_[0], F_SETFL(), $flags | O_NONBLOCK());
188 if ($blocking_supported) {
189 my $flags = fcntl ($_[0], F_GETFL(), 0);
190 $flags &= ~O_NONBLOCK(); # Clear blocking, but preserve other flags
191 fcntl ($_[0], F_SETFL(), $flags);
195 sub handle_send_err {
196 # For more meaningful handling of send errors, subclass Msg and
198 my ($conn, $err_msg) = @_;
199 warn "Error while sending: $err_msg \n";
200 set_event_handler ($conn->{sock}, "write" => undef);
203 #-----------------------------------------------------------------
204 # Receive side routines
207 @_ == 4 || die "Msg->new_server (myhost, myport, login_proc\n";
208 my ($pkg, $my_host, $my_port, $login_proc) = @_;
209 my $self = $pkg->new($login_proc);
211 $self->{sock} = IO::Socket::INET->new (
212 LocalAddr => $my_host,
213 LocalPort => $my_port,
217 die "Could not create socket: $! \n" unless $self->{sock};
218 set_event_handler ($self->{sock}, "read" => sub { $self->new_client } );
227 while ($msg = shift @{$conn->{inqueue}}){
228 &{$conn->{rproc}}($conn, $msg, $!);
233 sub _rcv { # Complement to _send
234 my $conn = shift; # $rcv_now complement of $flush
235 # Find out how much has already been received, if at all
236 my ($msg, $offset, $bytes_to_read, $bytes_read);
237 my $sock = $conn->{sock};
238 return unless defined($sock);
241 $conn->set_non_blocking();
242 $bytes_read = sysread ($sock, $msg, 1024, 0);
243 if (defined ($bytes_read)) {
244 if ($bytes_read > 0) {
246 @lines = split /\r?\n/, $msg;
247 $lines[0] = '' unless @lines;
248 $lines[0] = $conn->{msg} . $lines[0] if exists $conn->{msg};
249 push @lines, ' ' unless @lines;
253 $conn->{msg} = pop @lines;
255 push @{$conn->{inqueue}}, @lines if @lines;
257 $conn->{msg} .= $msg;
261 if (_err_will_block($!)) {
269 if (defined $bytes_read && $bytes_read == 0) {
270 # $conn->disconnect();
271 &{$conn->{rproc}}($conn, undef, $!);
272 delete $conn->{inqueue};
279 my $server_conn = shift;
280 my $sock = $server_conn->{sock}->accept();
281 my $conn = $server_conn->new($server_conn->{rproc});
282 $conn->{sock} = $sock;
283 my $rproc = &{$server_conn->{rproc}} ($conn, $sock->peerhost(), $sock->peerport());
285 $conn->{rproc} = $rproc;
286 my $callback = sub {_rcv($conn)};
287 set_event_handler ($sock, "read" => $callback);
288 } else { # Login failed
296 set_event_handler ($conn->{sock}, "read" => undef);
297 $conn->{sock}->close;
300 #----------------------------------------------------
301 # Event loop routines used by both client and server
303 sub set_event_handler {
304 shift unless ref($_[0]); # shift if first arg is package name
305 my ($handle, %args) = @_;
307 if (exists $args{'write'}) {
308 $callback = $args{'write'};
310 $wt_callbacks{$handle} = $callback;
311 $wt_handles->add($handle);
313 delete $wt_callbacks{$handle};
314 $wt_handles->remove($handle);
317 if (exists $args{'read'}) {
318 $callback = $args{'read'};
320 $rd_callbacks{$handle} = $callback;
321 $rd_handles->add($handle);
323 delete $rd_callbacks{$handle};
324 $rd_handles->remove($handle);
331 my ($pkg, $time, $proc, $recur) = @_;
333 my $class = $obj || $pkg;
334 my $self = bless { t=>$time + time, proc=>$proc }, $class;
335 $self->{interval} = $time if $recur;
336 push @timerchain, $self;
343 @timerchain = grep {$_ != $self} @timerchain;
347 my ($pkg, $loop_count, $timeout) = @_; # event_loop(1) to process events once
348 my ($conn, $r, $w, $rset, $wset);
351 # Quit the loop if no handles left to process
352 last unless ($rd_handles->count() || $wt_handles->count());
355 IO::Select->select ($rd_handles, $wt_handles, undef, $timeout);
358 foreach $r (@$rset) {
359 &{$rd_callbacks{$r}} ($r) if exists $rd_callbacks{$r};
361 foreach $w (@$wset) {
362 &{$wt_callbacks{$w}}($w) if exists $wt_callbacks{$w};
365 # handle things on the timer chain
367 if ($now >= $_->{t}) {
369 $_->{t} = $now + $_->{interval} if exists $_->{interval};
374 @timerchain = grep { $_->{t} > $now } @timerchain;
376 if (defined($loop_count)) {
377 last unless --$loop_count;