]> dxcluster.org Git - spider.git/blob - perl/Msg.pm
*** empty log message ***
[spider.git] / 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 require Exporter;
14 @ISA = qw(Exporter);
15
16 use strict;
17 use IO::Select;
18 use IO::Socket;
19 use Carp;
20
21 use vars qw (%rd_callbacks %wt_callbacks $rd_handles $wt_handles);
22
23 %rd_callbacks = ();
24 %wt_callbacks = ();
25 $rd_handles   = IO::Select->new();
26 $wt_handles   = IO::Select->new();
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 # Send side routines
39 sub connect {
40     my ($pkg, $to_host, $to_port,$rcvd_notification_proc) = @_;
41     
42     # Create a new internet socket
43     
44     my $sock = IO::Socket::INET->new (
45                                       PeerAddr => $to_host,
46                                       PeerPort => $to_port,
47                                       Proto    => 'tcp',
48                                       Reuse    => 1);
49
50     return undef unless $sock;
51
52     # Create a connection end-point object
53     my $conn = {
54         sock                   => $sock,
55         rcvd_notification_proc => $rcvd_notification_proc,
56     };
57     
58     if ($rcvd_notification_proc) {
59         my $callback = sub {_rcv($conn, 0)};
60         set_event_handler ($sock, "read" => $callback);
61     }
62     return bless $conn, $pkg;
63 }
64
65 sub disconnect {
66     my $conn = shift;
67     my $sock = delete $conn->{sock};
68     return unless defined($sock);
69     set_event_handler ($sock, "read" => undef, "write" => undef);
70     shutdown($sock, 3);
71         close($sock);
72 }
73
74 sub send_now {
75     my ($conn, $msg) = @_;
76     _enqueue ($conn, $msg);
77     $conn->_send (1); # 1 ==> flush
78 }
79
80 sub send_later {
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)});
86 }
87
88 sub _enqueue {
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);
94 }
95
96 sub _send {
97     my ($conn, $flush) = @_;
98     my $sock = $conn->{sock};
99     return unless defined($sock);
100     my ($rq) = $conn->{queue};
101
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.
107
108     $flush ? $conn->set_blocking() : $conn->set_non_blocking();
109     my $offset = (exists $conn->{send_offset}) ? $conn->{send_offset} : 0;
110
111     while (@$rq) {
112         my $msg            = $rq->[0];
113         my $bytes_to_write = length($msg) - $offset;
114         my $bytes_written  = 0;
115         while ($bytes_to_write) {
116             $bytes_written = syswrite ($sock, $msg,
117                                        $bytes_to_write, $offset);
118             if (!defined($bytes_written)) {
119                 if (_err_will_block($!)) {
120                     # Should happen only in deferred mode. Record how
121                     # much we have already sent.
122                     $conn->{send_offset} = $offset;
123                     # Event handler should already be set, so we will
124                     # be called back eventually, and will resume sending
125                     return 1;
126                 } else {    # Uh, oh
127                     $conn->handle_send_err($!);
128                     return 0; # fail. Message remains in queue ..
129                 }
130             }
131             $offset         += $bytes_written;
132             $bytes_to_write -= $bytes_written;
133         }
134         delete $conn->{send_offset};
135         $offset = 0;
136         shift @$rq;
137         last unless $flush; # Go back to select and wait
138                             # for it to fire again.
139     }
140     # Call me back if queue has not been drained.
141     if (@$rq) {
142         set_event_handler ($sock, "write" => sub {$conn->_send(0)});
143     } else {
144         set_event_handler ($sock, "write" => undef);
145     }
146     1;  # Success
147 }
148
149 sub _err_will_block {
150     if ($blocking_supported) {
151         return ($_[0] == EAGAIN());
152     }
153     return 0;
154 }
155 sub set_non_blocking {                        # $conn->set_blocking
156     if ($blocking_supported) {
157         # preserve other fcntl flags
158         my $flags = fcntl ($_[0], F_GETFL(), 0);
159         fcntl ($_[0], F_SETFL(), $flags | O_NONBLOCK());
160     }
161 }
162 sub set_blocking {
163     if ($blocking_supported) {
164         my $flags = fcntl ($_[0], F_GETFL(), 0);
165         $flags  &= ~O_NONBLOCK(); # Clear blocking, but preserve other flags
166         fcntl ($_[0], F_SETFL(), $flags);
167     }
168 }
169 sub handle_send_err {
170    # For more meaningful handling of send errors, subclass Msg and
171    # rebless $conn.  
172    my ($conn, $err_msg) = @_;
173    warn "Error while sending: $err_msg \n";
174    set_event_handler ($conn->{sock}, "write" => undef);
175 }
176
177 #-----------------------------------------------------------------
178 # Receive side routines
179
180 my ($g_login_proc,$g_pkg);
181 my $main_socket = 0;
182 sub new_server {
183     @_ == 4 || die "Msg->new_server (myhost, myport, login_proc)\n";
184     my ($pkg, $my_host, $my_port, $login_proc) = @_;
185     
186     $main_socket = IO::Socket::INET->new (
187                                           LocalAddr => $my_host,
188                                           LocalPort => $my_port,
189                                           Listen    => 5,
190                                           Proto     => 'tcp',
191                                           Reuse     => 1);
192     die "Could not create socket: $! \n" unless $main_socket;
193     set_event_handler ($main_socket, "read" => \&_new_client);
194     $g_login_proc = $login_proc; $g_pkg = $pkg;
195 }
196
197 sub rcv_now {
198     my ($conn) = @_;
199     my ($msg, $err) = _rcv ($conn, 1); # 1 ==> rcv now
200     return wantarray ? ($msg, $err) : $msg;
201 }
202
203 sub _rcv {                     # Complement to _send
204     my ($conn, $rcv_now) = @_; # $rcv_now complement of $flush
205     # Find out how much has already been received, if at all
206     my ($msg, $offset, $bytes_to_read, $bytes_read);
207     my $sock = $conn->{sock};
208     return unless defined($sock);
209     if (exists $conn->{msg}) {
210         $msg           = $conn->{msg};
211         $offset        = length($msg) - 1;  # sysread appends to it.
212         $bytes_to_read = $conn->{bytes_to_read};
213         delete $conn->{'msg'};              # have made a copy
214     } else {
215         # The typical case ...
216         $msg           = "";                # Otherwise -w complains 
217         $offset        = 0 ;  
218         $bytes_to_read = 0 ;                # Will get set soon
219     }
220     # We want to read the message length in blocking mode. Quite
221     # unlikely that we'll get blocked too long reading 4 bytes
222     if (!$bytes_to_read)  {                 # Get new length 
223         my $buf;
224         $conn->set_blocking();
225         $bytes_read = sysread($sock, $buf, 4);
226         if ($! || ($bytes_read != 4)) {
227             goto FINISH;
228         }
229         $bytes_to_read = unpack ('N', $buf);
230     }
231     $conn->set_non_blocking() unless $rcv_now;
232     while ($bytes_to_read) {
233         $bytes_read = sysread ($sock, $msg, $bytes_to_read, $offset);
234         if (defined ($bytes_read)) {
235             if ($bytes_read == 0) {
236                 last;
237             }
238             $bytes_to_read -= $bytes_read;
239             $offset        += $bytes_read;
240         } else {
241             if (_err_will_block($!)) {
242                 # Should come here only in non-blocking mode
243                 $conn->{msg}           = $msg;
244                 $conn->{bytes_to_read} = $bytes_to_read;
245                 return ;   # .. _rcv will be called later
246                            # when socket is readable again
247             } else {
248                 last;
249             }
250         }
251     }
252
253   FINISH:
254     if (length($msg) == 0) {
255         $conn->disconnect();
256     }
257     if ($rcv_now) {
258         return ($msg, $!);
259     } else {
260         &{$conn->{rcvd_notification_proc}}($conn, $msg, $!);
261     }
262 }
263
264 sub _new_client {
265     my $sock = $main_socket->accept();
266     my $conn = bless {
267         'sock' =>  $sock,
268         'state' => 'connected'
269     }, $g_pkg;
270     my $rcvd_notification_proc =
271         &$g_login_proc ($conn, $sock->peerhost(), $sock->peerport());
272     if ($rcvd_notification_proc) {
273         $conn->{rcvd_notification_proc} = $rcvd_notification_proc;
274         my $callback = sub {_rcv($conn,0)};
275         set_event_handler ($sock, "read" => $callback);
276     } else {  # Login failed
277         $conn->disconnect();
278     }
279 }
280
281 #----------------------------------------------------
282 # Event loop routines used by both client and server
283
284 sub set_event_handler {
285     shift unless ref($_[0]); # shift if first arg is package name
286     my ($handle, %args) = @_;
287     my $callback;
288     if (exists $args{'write'}) {
289         $callback = $args{'write'};
290         if ($callback) {
291             $wt_callbacks{$handle} = $callback;
292             $wt_handles->add($handle);
293         } else {
294             delete $wt_callbacks{$handle};
295             $wt_handles->remove($handle);
296         }
297     }
298     if (exists $args{'read'}) {
299         $callback = $args{'read'};
300         if ($callback) {
301             $rd_callbacks{$handle} = $callback;
302             $rd_handles->add($handle);
303         } else {
304             delete $rd_callbacks{$handle};
305             $rd_handles->remove($handle);
306        }
307     }
308 }
309
310 sub event_loop {
311     my ($pkg, $loop_count, $timeout) = @_; # event_loop(1) to process events once
312     my ($conn, $r, $w, $rset, $wset);
313     while (1) {
314         # Quit the loop if no handles left to process
315         last unless ($rd_handles->count() || $wt_handles->count());
316         ($rset, $wset) =
317             IO::Select->select ($rd_handles, $wt_handles, undef, $timeout);
318         foreach $r (@$rset) {
319             &{$rd_callbacks{$r}} ($r) if exists $rd_callbacks{$r};
320         }
321         foreach $w (@$wset) {
322             &{$wt_callbacks{$w}}($w) if exists $wt_callbacks{$w};
323         }
324         if (defined($loop_count)) {
325             last unless --$loop_count;
326         }
327     }
328 }
329
330 1;
331
332 __END__
333