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;
57 return bless $conn, $class;
60 #-----------------------------------------------------------------
63 my ($pkg, $to_host, $to_port, $rproc) = @_;
65 # Create a connection end-point object
68 $conn = $pkg->new($rproc);
71 # Create a new internet socket
72 my $sock = IO::Socket::INET->new (
77 Timeout => $conn->{timeval} / 2);
79 return undef unless $sock;
81 $conn->{sock} = $sock;
84 my $callback = sub {_rcv($conn)};
85 set_event_handler ($sock, "read" => $callback);
92 my $sock = delete $conn->{sock};
95 $conn->{timeout}->del_timer if $conn->{timeout};
96 return unless defined($sock);
97 set_event_handler ($sock, "read" => undef, "write" => undef);
103 my ($conn, $msg) = @_;
104 $conn->enqueue($msg);
105 $conn->_send (1); # 1 ==> flush
109 my ($conn, $msg) = @_;
110 $conn->enqueue($msg);
111 my $sock = $conn->{sock};
112 return unless defined($sock);
113 set_event_handler ($sock, "write" => sub {$conn->_send(0)});
118 push (@{$conn->{outqueue}}, $_[0]);
122 my ($conn, $flush) = @_;
123 my $sock = $conn->{sock};
124 return unless defined($sock);
125 my $rq = $conn->{outqueue};
127 # If $flush is set, set the socket to blocking, and send all
128 # messages in the queue - return only if there's an error
129 # If $flush is 0 (deferred mode) make the socket non-blocking, and
130 # return to the event loop only after every message, or if it
131 # is likely to block in the middle of a message.
133 $flush ? $conn->set_blocking() : $conn->set_non_blocking();
134 my $offset = (exists $conn->{send_offset}) ? $conn->{send_offset} : 0;
138 my $mlth = length($msg);
139 my $bytes_to_write = $mlth - $offset;
140 my $bytes_written = 0;
141 confess("Negative Length! msg: '$msg' lth: $mlth offset: $offset") if $bytes_to_write < 0;
142 while ($bytes_to_write > 0) {
143 $bytes_written = syswrite ($sock, $msg,
144 $bytes_to_write, $offset);
145 if (!defined($bytes_written)) {
146 if (_err_will_block($!)) {
147 # Should happen only in deferred mode. Record how
148 # much we have already sent.
149 $conn->{send_offset} = $offset;
150 # Event handler should already be set, so we will
151 # be called back eventually, and will resume sending
154 delete $conn->{send_offset};
155 $conn->handle_send_err($!);
157 return 0; # fail. Message remains in queue ..
160 $offset += $bytes_written;
161 $bytes_to_write -= $bytes_written;
163 delete $conn->{send_offset};
166 last unless $flush; # Go back to select and wait
167 # for it to fire again.
169 # Call me back if queue has not been drained.
171 set_event_handler ($sock, "write" => sub {$conn->_send(0)});
173 set_event_handler ($sock, "write" => undef);
178 sub _err_will_block {
179 if ($blocking_supported) {
180 return ($_[0] == EAGAIN());
184 sub set_non_blocking { # $conn->set_blocking
185 if ($blocking_supported) {
186 # preserve other fcntl flags
187 my $flags = fcntl ($_[0], F_GETFL(), 0);
188 fcntl ($_[0], F_SETFL(), $flags | O_NONBLOCK());
192 if ($blocking_supported) {
193 my $flags = fcntl ($_[0], F_GETFL(), 0);
194 $flags &= ~O_NONBLOCK(); # Clear blocking, but preserve other flags
195 fcntl ($_[0], F_SETFL(), $flags);
199 sub handle_send_err {
200 # For more meaningful handling of send errors, subclass Msg and
202 my ($conn, $err_msg) = @_;
203 warn "Error while sending: $err_msg \n";
204 set_event_handler ($conn->{sock}, "write" => undef);
207 #-----------------------------------------------------------------
208 # Receive side routines
211 @_ == 4 || die "Msg->new_server (myhost, myport, login_proc\n";
212 my ($pkg, $my_host, $my_port, $login_proc) = @_;
213 my $self = $pkg->new($login_proc);
215 $self->{sock} = IO::Socket::INET->new (
216 LocalAddr => $my_host,
217 LocalPort => $my_port,
221 die "Could not create socket: $! \n" unless $self->{sock};
222 set_event_handler ($self->{sock}, "read" => sub { $self->new_client } );
231 while ($msg = shift @{$conn->{inqueue}}){
232 &{$conn->{rproc}}($conn, $msg, $!);
237 sub _rcv { # Complement to _send
238 my $conn = shift; # $rcv_now complement of $flush
239 # Find out how much has already been received, if at all
240 my ($msg, $offset, $bytes_to_read, $bytes_read);
241 my $sock = $conn->{sock};
242 return unless defined($sock);
245 $conn->set_non_blocking();
246 $bytes_read = sysread ($sock, $msg, 1024, 0);
247 if (defined ($bytes_read)) {
248 if ($bytes_read > 0) {
250 @lines = split /\r?\n/, $msg;
252 $lines[0] = $conn->{msg} . $lines[0] if exists $conn->{msg};
254 $lines[0] = $conn->{msg} if exists $conn->{msg};
255 push @lines, '' unless @lines;
260 $conn->{msg} = pop @lines;
262 push @{$conn->{inqueue}}, @lines if @lines;
264 $conn->{msg} .= $msg;
268 if (_err_will_block($!)) {
276 if (defined $bytes_read && $bytes_read == 0) {
277 # $conn->disconnect();
278 &{$conn->{rproc}}($conn, undef, $!);
279 delete $conn->{inqueue};
286 my $server_conn = shift;
287 my $sock = $server_conn->{sock}->accept();
288 my $conn = $server_conn->new($server_conn->{rproc});
289 $conn->{sock} = $sock;
290 my $rproc = &{$server_conn->{rproc}} ($conn, $sock->peerhost(), $sock->peerport());
292 $conn->{rproc} = $rproc;
293 my $callback = sub {_rcv($conn)};
294 set_event_handler ($sock, "read" => $callback);
295 } else { # Login failed
303 set_event_handler ($conn->{sock}, "read" => undef);
304 $conn->{sock}->close;
307 #----------------------------------------------------
308 # Event loop routines used by both client and server
310 sub set_event_handler {
311 shift unless ref($_[0]); # shift if first arg is package name
312 my ($handle, %args) = @_;
314 if (exists $args{'write'}) {
315 $callback = $args{'write'};
317 $wt_callbacks{$handle} = $callback;
318 $wt_handles->add($handle);
320 delete $wt_callbacks{$handle};
321 $wt_handles->remove($handle);
324 if (exists $args{'read'}) {
325 $callback = $args{'read'};
327 $rd_callbacks{$handle} = $callback;
328 $rd_handles->add($handle);
330 delete $rd_callbacks{$handle};
331 $rd_handles->remove($handle);
338 my ($pkg, $time, $proc, $recur) = @_;
340 my $class = $obj || $pkg;
341 my $self = bless { t=>$time + time, proc=>$proc }, $class;
342 $self->{interval} = $time if $recur;
343 push @timerchain, $self;
350 @timerchain = grep {$_ != $self} @timerchain;
354 my ($pkg, $loop_count, $timeout) = @_; # event_loop(1) to process events once
355 my ($conn, $r, $w, $rset, $wset);
358 # Quit the loop if no handles left to process
359 last unless ($rd_handles->count() || $wt_handles->count());
362 IO::Select->select ($rd_handles, $wt_handles, undef, $timeout);
365 foreach $r (@$rset) {
366 &{$rd_callbacks{$r}} ($r) if exists $rd_callbacks{$r};
368 foreach $w (@$wset) {
369 &{$wt_callbacks{$w}}($w) if exists $wt_callbacks{$w};
372 # handle things on the timer chain
374 if ($now >= $_->{t}) {
376 $_->{t} = $now + $_->{interval} if exists $_->{interval};
381 @timerchain = grep { $_->{t} > $now } @timerchain;
383 if (defined($loop_count)) {
384 last unless --$loop_count;