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, 0)};
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 = pack ('N', $len) . $msg;
93 push (@{$conn->{queue}}, $msg);
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($!);
131 return 0; # fail. Message remains in queue ..
134 $offset += $bytes_written;
135 $bytes_to_write -= $bytes_written;
137 delete $conn->{send_offset};
140 last unless $flush; # Go back to select and wait
141 # for it to fire again.
143 # Call me back if queue has not been drained.
145 set_event_handler ($sock, "write" => sub {$conn->_send(0)});
147 set_event_handler ($sock, "write" => undef);
152 sub _err_will_block {
153 if ($blocking_supported) {
154 return ($_[0] == EAGAIN());
158 sub set_non_blocking { # $conn->set_blocking
159 if ($blocking_supported) {
160 # preserve other fcntl flags
161 my $flags = fcntl ($_[0], F_GETFL(), 0);
162 fcntl ($_[0], F_SETFL(), $flags | O_NONBLOCK());
166 if ($blocking_supported) {
167 my $flags = fcntl ($_[0], F_GETFL(), 0);
168 $flags &= ~O_NONBLOCK(); # Clear blocking, but preserve other flags
169 fcntl ($_[0], F_SETFL(), $flags);
173 sub handle_send_err {
174 # For more meaningful handling of send errors, subclass Msg and
176 my ($conn, $err_msg) = @_;
177 warn "Error while sending: $err_msg \n";
178 set_event_handler ($conn->{sock}, "write" => undef);
181 #-----------------------------------------------------------------
182 # Receive side routines
184 my ($g_login_proc,$g_pkg);
187 @_ == 4 || die "Msg->new_server (myhost, myport, login_proc)\n";
188 my ($pkg, $my_host, $my_port, $login_proc) = @_;
190 $main_socket = IO::Socket::INET->new (
191 LocalAddr => $my_host,
192 LocalPort => $my_port,
196 die "Could not create socket: $! \n" unless $main_socket;
197 set_event_handler ($main_socket, "read" => \&_new_client);
198 $g_login_proc = $login_proc; $g_pkg = $pkg;
203 my ($msg, $err) = _rcv ($conn, 1); # 1 ==> rcv now
204 return wantarray ? ($msg, $err) : $msg;
207 sub _rcv { # Complement to _send
208 my ($conn, $rcv_now) = @_; # $rcv_now complement of $flush
209 # Find out how much has already been received, if at all
210 my ($msg, $offset, $bytes_to_read, $bytes_read);
211 my $sock = $conn->{sock};
212 return unless defined($sock);
213 if (exists $conn->{msg}) {
215 $offset = length($msg) - 1; # sysread appends to it.
216 $bytes_to_read = $conn->{bytes_to_read};
217 delete $conn->{'msg'}; # have made a copy
219 # The typical case ...
220 $msg = ""; # Otherwise -w complains
222 $bytes_to_read = 0 ; # Will get set soon
224 # We want to read the message length in blocking mode. Quite
225 # unlikely that we'll get blocked too long reading 4 bytes
226 if (!$bytes_to_read) { # Get new length
228 $conn->set_blocking();
229 $bytes_read = sysread($sock, $buf, 4);
230 if ($! || ($bytes_read != 4)) {
233 $bytes_to_read = unpack ('N', $buf);
235 $conn->set_non_blocking() unless $rcv_now;
236 while ($bytes_to_read) {
237 $bytes_read = sysread ($sock, $msg, $bytes_to_read, $offset);
238 if (defined ($bytes_read)) {
239 if ($bytes_read == 0) {
242 $bytes_to_read -= $bytes_read;
243 $offset += $bytes_read;
245 if (_err_will_block($!)) {
246 # Should come here only in non-blocking mode
248 $conn->{bytes_to_read} = $bytes_to_read;
249 return ; # .. _rcv will be called later
250 # when socket is readable again
258 if (length($msg) == 0) {
264 &{$conn->{rcvd_notification_proc}}($conn, $msg, $!);
269 my $sock = $main_socket->accept();
272 'state' => 'connected'
274 my $rcvd_notification_proc =
275 &$g_login_proc ($conn, $sock->peerhost(), $sock->peerport());
276 if ($rcvd_notification_proc) {
277 $conn->{rcvd_notification_proc} = $rcvd_notification_proc;
278 my $callback = sub {_rcv($conn,0)};
279 set_event_handler ($sock, "read" => $callback);
280 } else { # Login failed
287 set_event_handler ($main_socket, "read" => undef);
292 #----------------------------------------------------
293 # Event loop routines used by both client and server
295 sub set_event_handler {
296 shift unless ref($_[0]); # shift if first arg is package name
297 my ($handle, %args) = @_;
299 if (exists $args{'write'}) {
300 $callback = $args{'write'};
302 $wt_callbacks{$handle} = $callback;
303 $wt_handles->add($handle);
305 delete $wt_callbacks{$handle};
306 $wt_handles->remove($handle);
309 if (exists $args{'read'}) {
310 $callback = $args{'read'};
312 $rd_callbacks{$handle} = $callback;
313 $rd_handles->add($handle);
315 delete $rd_callbacks{$handle};
316 $rd_handles->remove($handle);
322 my ($pkg, $loop_count, $timeout) = @_; # event_loop(1) to process events once
323 my ($conn, $r, $w, $rset, $wset);
325 # Quit the loop if no handles left to process
326 last unless ($rd_handles->count() || $wt_handles->count());
328 IO::Select->select ($rd_handles, $wt_handles, undef, $timeout);
329 foreach $r (@$rset) {
330 &{$rd_callbacks{$r}} ($r) if exists $rd_callbacks{$r};
332 foreach $w (@$wset) {
333 &{$wt_callbacks{$w}}($w) if exists $wt_callbacks{$w};
335 if (defined($loop_count)) {
336 last unless --$loop_count;