ViewVC Help
View File | Revision Log | Show Annotations | Download File
/cvs/AnyEvent/lib/AnyEvent/Handle.pm
(Generate patch)

Comparing AnyEvent/lib/AnyEvent/Handle.pm (file contents):
Revision 1.28 by root, Sat May 24 22:27:11 2008 UTC vs.
Revision 1.37 by root, Mon May 26 20:02:22 2008 UTC

… …
2 2
3no warnings; 3no warnings;
4use strict; 4use strict;
5 5
6use AnyEvent (); 6use AnyEvent ();
7use AnyEvent::Util (); 7use AnyEvent::Util qw(WSAWOULDBLOCK);
8use Scalar::Util (); 8use Scalar::Util ();
9use Carp (); 9use Carp ();
10use Fcntl (); 10use Fcntl ();
11use Errno qw/EAGAIN EINTR/; 11use Errno qw/EAGAIN EINTR/;
12 12
13=head1 NAME 13=head1 NAME
14 14
15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent 15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
16 16
17This module is experimental.
18
19=cut 17=cut
20 18
21our $VERSION = '0.04'; 19our $VERSION = '0.04';
22 20
23=head1 SYNOPSIS 21=head1 SYNOPSIS
… …
25 use AnyEvent; 23 use AnyEvent;
26 use AnyEvent::Handle; 24 use AnyEvent::Handle;
27 25
28 my $cv = AnyEvent->condvar; 26 my $cv = AnyEvent->condvar;
29 27
30 my $ae_fh = AnyEvent::Handle->new (fh => \*STDIN); 28 my $handle =
31
32 #TODO
33
34 # or use the constructor to pass the callback:
35
36 my $ae_fh2 =
37 AnyEvent::Handle->new ( 29 AnyEvent::Handle->new (
38 fh => \*STDIN, 30 fh => \*STDIN,
39 on_eof => sub { 31 on_eof => sub {
40 $cv->broadcast; 32 $cv->broadcast;
41 }, 33 },
42 #TODO
43 ); 34 );
44 35
45 $cv->wait; 36 # send some request line
37 $handle->push_write ("getinfo\015\012");
38
39 # read the response line
40 $handle->push_read (line => sub {
41 my ($handle, $line) = @_;
42 warn "read line <$line>\n";
43 $cv->send;
44 });
45
46 $cv->recv;
46 47
47=head1 DESCRIPTION 48=head1 DESCRIPTION
48 49
49This module is a helper module to make it easier to do event-based I/O on 50This module is a helper module to make it easier to do event-based I/O on
50filehandles. For utility functions for doing non-blocking connects and accepts 51filehandles. For utility functions for doing non-blocking connects and accepts
… …
90 91
91The object will not be in a usable state when this callback has been 92The object will not be in a usable state when this callback has been
92called. 93called.
93 94
94On callback entrance, the value of C<$!> contains the operating system 95On callback entrance, the value of C<$!> contains the operating system
95error (or C<ENOSPC> or C<EPIPE>). 96error (or C<ENOSPC>, C<EPIPE> or C<EBADMSG>).
97
98The callbakc should throw an exception. If it returns, then
99AnyEvent::Handle will C<croak> for you.
96 100
97While not mandatory, it is I<highly> recommended to set this callback, as 101While not mandatory, it is I<highly> recommended to set this callback, as
98you will not be notified of errors otherwise. The default simply calls 102you will not be notified of errors otherwise. The default simply calls
99die. 103die.
100 104
… …
168 172
169=back 173=back
170 174
171=cut 175=cut
172 176
173our (%RH, %WH);
174
175sub register_read_type($$) {
176 $RH{$_[0]} = $_[1];
177}
178
179sub register_write_type($$) {
180 $WH{$_[0]} = $_[1];
181}
182
183sub new { 177sub new {
184 my $class = shift; 178 my $class = shift;
185 179
186 my $self = bless { @_ }, $class; 180 my $self = bless { @_ }, $class;
187 181
… …
218 { 212 {
219 local $!; 213 local $!;
220 $self->_shutdown; 214 $self->_shutdown;
221 } 215 }
222 216
223 if ($self->{on_error}) {
224 $self->{on_error}($self); 217 $self->{on_error}($self)
225 } else { 218 if $self->{on_error};
219
226 die "AnyEvent::Handle uncaught fatal error: $!"; 220 Carp::croak "AnyEvent::Handle uncaught fatal error: $!";
227 }
228} 221}
229 222
230=item $fh = $handle->fh 223=item $fh = $handle->fh
231 224
232This method returns the file handle of the L<AnyEvent::Handle> object. 225This method returns the file handle of the L<AnyEvent::Handle> object.
… …
297=cut 290=cut
298 291
299sub _drain_wbuf { 292sub _drain_wbuf {
300 my ($self) = @_; 293 my ($self) = @_;
301 294
302 unless ($self->{ww}) { 295 if (!$self->{ww} && length $self->{wbuf}) {
296
303 Scalar::Util::weaken $self; 297 Scalar::Util::weaken $self;
298
304 my $cb = sub { 299 my $cb = sub {
305 my $len = syswrite $self->{fh}, $self->{wbuf}; 300 my $len = syswrite $self->{fh}, $self->{wbuf};
306 301
307 if ($len > 0) { 302 if ($len >= 0) {
308 substr $self->{wbuf}, 0, $len, ""; 303 substr $self->{wbuf}, 0, $len, "";
309 304
310 $self->{on_drain}($self) 305 $self->{on_drain}($self)
311 if $self->{low_water_mark} >= length $self->{wbuf} 306 if $self->{low_water_mark} >= length $self->{wbuf}
312 && $self->{on_drain}; 307 && $self->{on_drain};
313 308
314 delete $self->{ww} unless length $self->{wbuf}; 309 delete $self->{ww} unless length $self->{wbuf};
315 } elsif ($! != EAGAIN && $! != EINTR) { 310 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAWOULDBLOCK) {
316 $self->error; 311 $self->error;
317 } 312 }
318 }; 313 };
319 314
315 # try to write data immediately
316 $cb->();
317
318 # if still data left in wbuf, we need to poll
320 $self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb); 319 $self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
321 320 if length $self->{wbuf};
322 $cb->($self);
323 }; 321 };
322}
323
324our %WH;
325
326sub register_write_type($$) {
327 $WH{$_[0]} = $_[1];
324} 328}
325 329
326sub push_write { 330sub push_write {
327 my $self = shift; 331 my $self = shift;
332
333 if (@_ > 1) {
334 my $type = shift;
335
336 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
337 ->($self, @_);
338 }
328 339
329 if ($self->{filter_w}) { 340 if ($self->{filter_w}) {
330 $self->{filter_w}->($self, \$_[0]); 341 $self->{filter_w}->($self, \$_[0]);
331 } else { 342 } else {
332 $self->{wbuf} .= $_[0]; 343 $self->{wbuf} .= $_[0];
333 $self->_drain_wbuf; 344 $self->_drain_wbuf;
334 } 345 }
335} 346}
347
348=item $handle->push_write (type => @args)
349
350=item $handle->unshift_write (type => @args)
351
352Instead of formatting your data yourself, you can also let this module do
353the job by specifying a type and type-specific arguments.
354
355Predefined types are (if you have ideas for additional types, feel free to
356drop by and tell us):
357
358=over 4
359
360=item netstring => $string
361
362Formats the given value as netstring
363(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
364
365=back
366
367=cut
368
369register_write_type netstring => sub {
370 my ($self, $string) = @_;
371
372 sprintf "%d:%s,", (length $string), $string
373};
374
375=item AnyEvent::Handle::register_write_type type => $coderef->($self, @args)
376
377This function (not method) lets you add your own types to C<push_write>.
378Whenever the given C<type> is used, C<push_write> will invoke the code
379reference with the handle object and the remaining arguments.
380
381The code reference is supposed to return a single octet string that will
382be appended to the write buffer.
383
384Note that this is a function, and all types registered this way will be
385global, so try to use unique names.
386
387=cut
336 388
337############################################################################# 389#############################################################################
338 390
339=back 391=back
340 392
… …
419 471
420 if ( 472 if (
421 defined $self->{rbuf_max} 473 defined $self->{rbuf_max}
422 && $self->{rbuf_max} < length $self->{rbuf} 474 && $self->{rbuf_max} < length $self->{rbuf}
423 ) { 475 ) {
424 $! = &Errno::ENOSPC; return $self->error; 476 $! = &Errno::ENOSPC;
477 $self->error;
425 } 478 }
426 479
427 return if $self->{in_drain}; 480 return if $self->{in_drain};
428 local $self->{in_drain} = 1; 481 local $self->{in_drain} = 1;
429 482
430 while (my $len = length $self->{rbuf}) { 483 while (my $len = length $self->{rbuf}) {
431 no strict 'refs'; 484 no strict 'refs';
432 if (my $cb = shift @{ $self->{queue} }) { 485 if (my $cb = shift @{ $self->{queue} }) {
433 if (!$cb->($self)) { 486 unless ($cb->($self)) {
434 if ($self->{eof}) { 487 if ($self->{eof}) {
435 # no progress can be made (not enough data and no data forthcoming) 488 # no progress can be made (not enough data and no data forthcoming)
436 $! = &Errno::EPIPE; return $self->error; 489 $! = &Errno::EPIPE;
490 $self->error;
437 } 491 }
438 492
439 unshift @{ $self->{queue} }, $cb; 493 unshift @{ $self->{queue} }, $cb;
440 return; 494 return;
441 } 495 }
… …
447 && $len == length $self->{rbuf} # and no data has been consumed 501 && $len == length $self->{rbuf} # and no data has been consumed
448 && !@{ $self->{queue} } # and the queue is still empty 502 && !@{ $self->{queue} } # and the queue is still empty
449 && $self->{on_read} # and we still want to read data 503 && $self->{on_read} # and we still want to read data
450 ) { 504 ) {
451 # then no progress can be made 505 # then no progress can be made
452 $! = &Errno::EPIPE; return $self->error; 506 $! = &Errno::EPIPE;
507 $self->error;
453 } 508 }
454 } else { 509 } else {
455 # read side becomes idle 510 # read side becomes idle
456 delete $self->{rw}; 511 delete $self->{rw};
457 return; 512 return;
… …
515interested in (which can be none at all) and return a true value. After returning 570interested in (which can be none at all) and return a true value. After returning
516true, it will be removed from the queue. 571true, it will be removed from the queue.
517 572
518=cut 573=cut
519 574
575our %RH;
576
577sub register_read_type($$) {
578 $RH{$_[0]} = $_[1];
579}
580
520sub push_read { 581sub push_read {
521 my $self = shift; 582 my $self = shift;
522 my $cb = pop; 583 my $cb = pop;
523 584
524 if (@_) { 585 if (@_) {
… …
554 615
555Instead of providing a callback that parses the data itself you can chose 616Instead of providing a callback that parses the data itself you can chose
556between a number of predefined parsing formats, for chunks of data, lines 617between a number of predefined parsing formats, for chunks of data, lines
557etc. 618etc.
558 619
559The types currently supported are: 620Predefined types are (if you have ideas for additional types, feel free to
621drop by and tell us):
560 622
561=over 4 623=over 4
562 624
563=item chunk => $octets, $cb->($self, $data) 625=item chunk => $octets, $cb->($self, $data)
564 626
… …
638sub unshift_read_line { 700sub unshift_read_line {
639 my $self = shift; 701 my $self = shift;
640 $self->unshift_read (line => @_); 702 $self->unshift_read (line => @_);
641} 703}
642 704
705=item netstring => $cb->($string)
706
707A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
708
709Throws an error with C<$!> set to EBADMSG on format violations.
710
711=cut
712
713register_read_type netstring => sub {
714 my ($self, $cb) = @_;
715
716 sub {
717 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
718 if ($_[0]{rbuf} =~ /[^0-9]/) {
719 $! = &Errno::EBADMSG;
720 $self->error;
721 }
722 return;
723 }
724
725 my $len = $1;
726
727 $self->unshift_read (chunk => $len, sub {
728 my $string = $_[1];
729 $_[0]->unshift_read (chunk => 1, sub {
730 if ($_[1] eq ",") {
731 $cb->($_[0], $string);
732 } else {
733 $! = &Errno::EBADMSG;
734 $self->error;
735 }
736 });
737 });
738
739 1
740 }
741};
742
743=item regex => $accept[, $reject[, $skip], $cb->($data)
744
745Makes a regex match against the regex object C<$accept> and returns
746everything up to and including the match.
747
748Example: read a single line terminated by '\n'.
749
750 $handle->push_read (regex => qr<\n>, sub { ... });
751
752If C<$reject> is given and not undef, then it determines when the data is
753to be rejected: it is matched against the data when the C<$accept> regex
754does not match and generates an C<EBADMSG> error when it matches. This is
755useful to quickly reject wrong data (to avoid waiting for a timeout or a
756receive buffer overflow).
757
758Example: expect a single decimal number followed by whitespace, reject
759anything else (not the use of an anchor).
760
761 $handle->push_read (regex => qr<^[0-9]+\s>, qr<[^0-9]>, sub { ... });
762
763If C<$skip> is given and not C<undef>, then it will be matched against
764the receive buffer when neither C<$accept> nor C<$reject> match,
765and everything preceding and including the match will be accepted
766unconditionally. This is useful to skip large amounts of data that you
767know cannot be matched, so that the C<$accept> or C<$reject> regex do not
768have to start matching from the beginning. This is purely an optimisation
769and is usually worth only when you expect more than a few kilobytes.
770
771Example: expect a http header, which ends at C<\015\012\015\012>. Since we
772expect the header to be very large (it isn't in practise, but...), we use
773a skip regex to skip initial portions. The skip regex is tricky in that
774it only accepts something not ending in either \015 or \012, as these are
775required for the accept regex.
776
777 $handle->push_read (regex =>
778 qr<\015\012\015\012>,
779 undef, # no reject
780 qr<^.*[^\015\012]>,
781 sub { ... });
782
783=cut
784
785register_read_type regex => sub {
786 my ($self, $cb, $accept, $reject, $skip) = @_;
787
788 my $data;
789 my $rbuf = \$self->{rbuf};
790
791 sub {
792 # accept
793 if ($$rbuf =~ $accept) {
794 $data .= substr $$rbuf, 0, $+[0], "";
795 $cb->($self, $data);
796 return 1;
797 }
798
799 # reject
800 if ($reject && $$rbuf =~ $reject) {
801 $! = &Errno::EBADMSG;
802 $self->error;
803 }
804
805 # skip
806 if ($skip && $$rbuf =~ $skip) {
807 $data .= substr $$rbuf, 0, $+[0], "";
808 }
809
810 ()
811 }
812};
813
643=back 814=back
815
816=item AnyEvent::Handle::register_read_type type => $coderef->($self, $cb, @args)
817
818This function (not method) lets you add your own types to C<push_read>.
819
820Whenever the given C<type> is used, C<push_read> will invoke the code
821reference with the handle object, the callback and the remaining
822arguments.
823
824The code reference is supposed to return a callback (usually a closure)
825that works as a plain read callback (see C<< ->push_read ($cb) >>).
826
827It should invoke the passed callback when it is done reading (remember to
828pass C<$self> as first argument as all other callbacks do that).
829
830Note that this is a function, and all types registered this way will be
831global, so try to use unique names.
832
833For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>,
834search for C<register_read_type>)).
644 835
645=item $handle->stop_read 836=item $handle->stop_read
646 837
647=item $handle->start_read 838=item $handle->start_read
648 839
… …
677 } elsif (defined $len) { 868 } elsif (defined $len) {
678 delete $self->{rw}; 869 delete $self->{rw};
679 $self->{eof} = 1; 870 $self->{eof} = 1;
680 $self->_drain_rbuf; 871 $self->_drain_rbuf;
681 872
682 } elsif ($! != EAGAIN && $! != EINTR) { 873 } elsif ($! != EAGAIN && $! != EINTR && $! != &AnyEvent::Util::WSAWOULDBLOCK) {
683 return $self->error; 874 return $self->error;
684 } 875 }
685 }); 876 });
686 } 877 }
687} 878}
… …
753 # but the openssl maintainers basically said: "trust us, it just works". 944 # but the openssl maintainers basically said: "trust us, it just works".
754 # (unfortunately, we have to hardcode constants because the abysmally misdesigned 945 # (unfortunately, we have to hardcode constants because the abysmally misdesigned
755 # and mismaintained ssleay-module doesn't even offer them). 946 # and mismaintained ssleay-module doesn't even offer them).
756 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html 947 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
757 Net::SSLeay::CTX_set_mode ($self->{tls}, 948 Net::SSLeay::CTX_set_mode ($self->{tls},
758 (eval { Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1) 949 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
759 | (eval { Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2)); 950 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
760 951
761 $self->{tls_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 952 $self->{tls_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
762 $self->{tls_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 953 $self->{tls_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
763 954
764 Net::SSLeay::set_bio ($ssl, $self->{tls_rbio}, $self->{tls_wbio}); 955 Net::SSLeay::set_bio ($ssl, $self->{tls_rbio}, $self->{tls_wbio});

Diff Legend

– Removed lines
+ Added lines
< Changed lines
> Changed lines