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)
21 use vars qw (%rd_callbacks %wt_callbacks $rd_handles $wt_handles);
25 $rd_handles = IO::Select->new();
26 $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 $@;
37 #-----------------------------------------------------------------
40 my ($pkg, $to_host, $to_port,$rcvd_notification_proc) = @_;
42 # Create a new internet socket
44 my $sock = IO::Socket::INET->new (
50 return undef unless $sock;
52 # Create a connection end-point object
55 rcvd_notification_proc => $rcvd_notification_proc,
58 if ($rcvd_notification_proc) {
59 my $callback = sub {_rcv($conn)};
60 set_event_handler ($sock, "read" => $callback);
62 return bless $conn, $pkg;
67 my $sock = delete $conn->{sock};
68 return unless defined($sock);
69 set_event_handler ($sock, "read" => undef, "write" => undef);
75 my ($conn, $msg) = @_;
76 _enqueue ($conn, $msg);
77 $conn->_send (1); # 1 ==> flush
81 my ($conn, $msg) = @_;
82 _enqueue($conn, $msg);
83 my $sock = $conn->{sock};
84 return unless defined($sock);
85 set_event_handler ($sock, "write" => sub {$conn->_send(0)});
89 my ($conn, $msg) = @_;
90 # prepend length (encoded as network long)
91 my $len = length($msg);
92 $msg =~ s/([\%\x00-\x1f\x7f-\xff])/sprintf("%%%02X", ord($1))/eg;
93 push (@{$conn->{queue}}, $msg . "\n");
97 my ($conn, $flush) = @_;
98 my $sock = $conn->{sock};
99 return unless defined($sock);
100 my ($rq) = $conn->{queue};
102 # If $flush is set, set the socket to blocking, and send all
103 # messages in the queue - return only if there's an error
104 # If $flush is 0 (deferred mode) make the socket non-blocking, and
105 # return to the event loop only after every message, or if it
106 # is likely to block in the middle of a message.
108 $flush ? $conn->set_blocking() : $conn->set_non_blocking();
109 my $offset = (exists $conn->{send_offset}) ? $conn->{send_offset} : 0;
113 my $mlth = length($msg);
114 my $bytes_to_write = $mlth - $offset;
115 my $bytes_written = 0;
116 confess("Negative Length! msg: '$msg' lth: $mlth offset: $offset") if $bytes_to_write < 0;
117 while ($bytes_to_write > 0) {
118 $bytes_written = syswrite ($sock, $msg,
119 $bytes_to_write, $offset);
120 if (!defined($bytes_written)) {
121 if (_err_will_block($!)) {
122 # Should happen only in deferred mode. Record how
123 # much we have already sent.
124 $conn->{send_offset} = $offset;
125 # Event handler should already be set, so we will
126 # be called back eventually, and will resume sending
129 delete $conn->{send_offset};
130 $conn->handle_send_err($!);
132 return 0; # fail. Message remains in queue ..
135 $offset += $bytes_written;
136 $bytes_to_write -= $bytes_written;
138 delete $conn->{send_offset};
141 last unless $flush; # Go back to select and wait
142 # for it to fire again.
144 # Call me back if queue has not been drained.
146 set_event_handler ($sock, "write" => sub {$conn->_send(0)});
148 set_event_handler ($sock, "write" => undef);
153 sub _err_will_block {
154 if ($blocking_supported) {
155 return ($_[0] == EAGAIN());
159 sub set_non_blocking { # $conn->set_blocking
160 if ($blocking_supported) {
161 # preserve other fcntl flags
162 my $flags = fcntl ($_[0], F_GETFL(), 0);
163 fcntl ($_[0], F_SETFL(), $flags | O_NONBLOCK());
167 if ($blocking_supported) {
168 my $flags = fcntl ($_[0], F_GETFL(), 0);
169 $flags &= ~O_NONBLOCK(); # Clear blocking, but preserve other flags
170 fcntl ($_[0], F_SETFL(), $flags);
174 sub handle_send_err {
175 # For more meaningful handling of send errors, subclass Msg and
177 my ($conn, $err_msg) = @_;
178 warn "Error while sending: $err_msg \n";
179 set_event_handler ($conn->{sock}, "write" => undef);
182 #-----------------------------------------------------------------
183 # Receive side routines
185 my ($g_login_proc,$g_pkg);
188 @_ == 4 || die "Msg->new_server (myhost, myport, login_proc)\n";
189 my ($pkg, $my_host, $my_port, $login_proc) = @_;
191 $main_socket = IO::Socket::INET->new (
192 LocalAddr => $my_host,
193 LocalPort => $my_port,
197 die "Could not create socket: $! \n" unless $main_socket;
198 set_event_handler ($main_socket, "read" => \&_new_client);
199 $g_login_proc = $login_proc; $g_pkg = $pkg;
202 sub _rcv { # Complement to _send
203 my $conn = shift; # $rcv_now complement of $flush
204 # Find out how much has already been received, if at all
205 my ($msg, $offset, $bytes_to_read, $bytes_read);
206 my $sock = $conn->{sock};
207 return unless defined($sock);
210 $conn->set_non_blocking();
211 $bytes_read = sysread ($sock, $msg, 1024, 0);
212 if (defined ($bytes_read)) {
213 if ($bytes_read > 0) {
215 @lines = split /\n/, $msg;
216 $lines[0] = $conn->{msg} . $lines[0] if $conn->{msg};
220 $conn->{msg} = pop @lines;
223 $conn->{msg} .= $msg;
227 if (_err_will_block($!)) {
235 if (defined $bytes_read && $bytes_read == 0) {
236 # $conn->disconnect();
237 &{$conn->{rcvd_notification_proc}}($conn, undef, $!);
243 $msg =~ s/%([0-9A-Fa-f]{2})/chr(hex($1))/eg;
244 &{$conn->{rcvd_notification_proc}}($conn, $msg, $!);
250 my $sock = $main_socket->accept();
253 'state' => 'connected'
255 my $rcvd_notification_proc =
256 &$g_login_proc ($conn, $sock->peerhost(), $sock->peerport());
257 if ($rcvd_notification_proc) {
258 $conn->{rcvd_notification_proc} = $rcvd_notification_proc;
259 my $callback = sub {_rcv($conn)};
260 set_event_handler ($sock, "read" => $callback);
261 } else { # Login failed
268 set_event_handler ($main_socket, "read" => undef);
273 #----------------------------------------------------
274 # Event loop routines used by both client and server
276 sub set_event_handler {
277 shift unless ref($_[0]); # shift if first arg is package name
278 my ($handle, %args) = @_;
280 if (exists $args{'write'}) {
281 $callback = $args{'write'};
283 $wt_callbacks{$handle} = $callback;
284 $wt_handles->add($handle);
286 delete $wt_callbacks{$handle};
287 $wt_handles->remove($handle);
290 if (exists $args{'read'}) {
291 $callback = $args{'read'};
293 $rd_callbacks{$handle} = $callback;
294 $rd_handles->add($handle);
296 delete $rd_callbacks{$handle};
297 $rd_handles->remove($handle);
303 my ($pkg, $loop_count, $timeout) = @_; # event_loop(1) to process events once
304 my ($conn, $r, $w, $rset, $wset);
306 # Quit the loop if no handles left to process
307 last unless ($rd_handles->count() || $wt_handles->count());
309 IO::Select->select ($rd_handles, $wt_handles, undef, $timeout);
310 foreach $r (@$rset) {
311 &{$rd_callbacks{$r}} ($r) if exists $rd_callbacks{$r};
313 foreach $w (@$wset) {
314 &{$wt_callbacks{$w}}($w) if exists $wt_callbacks{$w};
316 if (defined($loop_count)) {
317 last unless --$loop_count;