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.36 by root, Mon May 26 18:26:52 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>).
96 97
97While not mandatory, it is I<highly> recommended to set this callback, as 98While not mandatory, it is I<highly> recommended to set this callback, as
98you will not be notified of errors otherwise. The default simply calls 99you will not be notified of errors otherwise. The default simply calls
99die. 100die.
100 101
… …
168 169
169=back 170=back
170 171
171=cut 172=cut
172 173
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 { 174sub new {
184 my $class = shift; 175 my $class = shift;
185 176
186 my $self = bless { @_ }, $class; 177 my $self = bless { @_ }, $class;
187 178
… …
221 } 212 }
222 213
223 if ($self->{on_error}) { 214 if ($self->{on_error}) {
224 $self->{on_error}($self); 215 $self->{on_error}($self);
225 } else { 216 } else {
226 die "AnyEvent::Handle uncaught fatal error: $!"; 217 Carp::croak "AnyEvent::Handle uncaught fatal error: $!";
227 } 218 }
228} 219}
229 220
230=item $fh = $handle->fh 221=item $fh = $handle->fh
231 222
… …
297=cut 288=cut
298 289
299sub _drain_wbuf { 290sub _drain_wbuf {
300 my ($self) = @_; 291 my ($self) = @_;
301 292
302 unless ($self->{ww}) { 293 if (!$self->{ww} && length $self->{wbuf}) {
294
303 Scalar::Util::weaken $self; 295 Scalar::Util::weaken $self;
296
304 my $cb = sub { 297 my $cb = sub {
305 my $len = syswrite $self->{fh}, $self->{wbuf}; 298 my $len = syswrite $self->{fh}, $self->{wbuf};
306 299
307 if ($len > 0) { 300 if ($len >= 0) {
308 substr $self->{wbuf}, 0, $len, ""; 301 substr $self->{wbuf}, 0, $len, "";
309 302
310 $self->{on_drain}($self) 303 $self->{on_drain}($self)
311 if $self->{low_water_mark} >= length $self->{wbuf} 304 if $self->{low_water_mark} >= length $self->{wbuf}
312 && $self->{on_drain}; 305 && $self->{on_drain};
313 306
314 delete $self->{ww} unless length $self->{wbuf}; 307 delete $self->{ww} unless length $self->{wbuf};
315 } elsif ($! != EAGAIN && $! != EINTR) { 308 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAWOULDBLOCK) {
316 $self->error; 309 $self->error;
317 } 310 }
318 }; 311 };
319 312
313 # try to write data immediately
314 $cb->();
315
316 # if still data left in wbuf, we need to poll
320 $self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb); 317 $self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
321 318 if length $self->{wbuf};
322 $cb->($self);
323 }; 319 };
320}
321
322our %WH;
323
324sub register_write_type($$) {
325 $WH{$_[0]} = $_[1];
324} 326}
325 327
326sub push_write { 328sub push_write {
327 my $self = shift; 329 my $self = shift;
330
331 if (@_ > 1) {
332 my $type = shift;
333
334 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
335 ->($self, @_);
336 }
328 337
329 if ($self->{filter_w}) { 338 if ($self->{filter_w}) {
330 $self->{filter_w}->($self, \$_[0]); 339 $self->{filter_w}->($self, \$_[0]);
331 } else { 340 } else {
332 $self->{wbuf} .= $_[0]; 341 $self->{wbuf} .= $_[0];
333 $self->_drain_wbuf; 342 $self->_drain_wbuf;
334 } 343 }
335} 344}
345
346=item $handle->push_write (type => @args)
347
348=item $handle->unshift_write (type => @args)
349
350Instead of formatting your data yourself, you can also let this module do
351the job by specifying a type and type-specific arguments.
352
353Predefined types are (if you have ideas for additional types, feel free to
354drop by and tell us):
355
356=over 4
357
358=item netstring => $string
359
360Formats the given value as netstring
361(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
362
363=back
364
365=cut
366
367register_write_type netstring => sub {
368 my ($self, $string) = @_;
369
370 sprintf "%d:%s,", (length $string), $string
371};
372
373=item AnyEvent::Handle::register_write_type type => $coderef->($self, @args)
374
375This function (not method) lets you add your own types to C<push_write>.
376Whenever the given C<type> is used, C<push_write> will invoke the code
377reference with the handle object and the remaining arguments.
378
379The code reference is supposed to return a single octet string that will
380be appended to the write buffer.
381
382Note that this is a function, and all types registered this way will be
383global, so try to use unique names.
384
385=cut
336 386
337############################################################################# 387#############################################################################
338 388
339=back 389=back
340 390
… …
428 local $self->{in_drain} = 1; 478 local $self->{in_drain} = 1;
429 479
430 while (my $len = length $self->{rbuf}) { 480 while (my $len = length $self->{rbuf}) {
431 no strict 'refs'; 481 no strict 'refs';
432 if (my $cb = shift @{ $self->{queue} }) { 482 if (my $cb = shift @{ $self->{queue} }) {
433 if (!$cb->($self)) { 483 unless ($cb->($self)) {
434 if ($self->{eof}) { 484 if ($self->{eof}) {
435 # no progress can be made (not enough data and no data forthcoming) 485 # no progress can be made (not enough data and no data forthcoming)
436 $! = &Errno::EPIPE; return $self->error; 486 $! = &Errno::EPIPE; return $self->error;
437 } 487 }
438 488
… …
515interested in (which can be none at all) and return a true value. After returning 565interested in (which can be none at all) and return a true value. After returning
516true, it will be removed from the queue. 566true, it will be removed from the queue.
517 567
518=cut 568=cut
519 569
570our %RH;
571
572sub register_read_type($$) {
573 $RH{$_[0]} = $_[1];
574}
575
520sub push_read { 576sub push_read {
521 my $self = shift; 577 my $self = shift;
522 my $cb = pop; 578 my $cb = pop;
523 579
524 if (@_) { 580 if (@_) {
… …
554 610
555Instead of providing a callback that parses the data itself you can chose 611Instead of providing a callback that parses the data itself you can chose
556between a number of predefined parsing formats, for chunks of data, lines 612between a number of predefined parsing formats, for chunks of data, lines
557etc. 613etc.
558 614
559The types currently supported are: 615Predefined types are (if you have ideas for additional types, feel free to
616drop by and tell us):
560 617
561=over 4 618=over 4
562 619
563=item chunk => $octets, $cb->($self, $data) 620=item chunk => $octets, $cb->($self, $data)
564 621
… …
638sub unshift_read_line { 695sub unshift_read_line {
639 my $self = shift; 696 my $self = shift;
640 $self->unshift_read (line => @_); 697 $self->unshift_read (line => @_);
641} 698}
642 699
700=item netstring => $cb->($string)
701
702A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
703
704Throws an error with C<$!> set to EBADMSG on format violations.
705
706=cut
707
708register_read_type netstring => sub {
709 my ($self, $cb) = @_;
710
711 sub {
712 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
713 if ($_[0]{rbuf} =~ /[^0-9]/) {
714 $! = &Errno::EBADMSG;
715 $self->error;
716 }
717 return;
718 }
719
720 my $len = $1;
721
722 $self->unshift_read (chunk => $len, sub {
723 my $string = $_[1];
724 $_[0]->unshift_read (chunk => 1, sub {
725 if ($_[1] eq ",") {
726 $cb->($_[0], $string);
727 } else {
728 $! = &Errno::EBADMSG;
729 $self->error;
730 }
731 });
732 });
733
734 1
735 }
736};
737
738=item regex => $accept[, $reject[, $skip], $cb->($data)
739
740Makes a regex match against the regex object C<$accept> and returns
741everything up to and including the match.
742
743Example: read a single line terminated by '\n'.
744
745 $handle->push_read (regex => qr<\n>, sub { ... });
746
747If C<$reject> is given and not undef, then it determines when the data is
748to be rejected: it is matched against the data when the C<$accept> regex
749does not match and generates an C<EBADMSG> error when it matches. This is
750useful to quickly reject wrong data (to avoid waiting for a timeout or a
751receive buffer overflow).
752
753Example: expect a single decimal number followed by whitespace, reject
754anything else (not the use of an anchor).
755
756 $handle->push_read (regex => qr<^[0-9]+\s>, qr<[^0-9]>, sub { ... });
757
758If C<$skip> is given and not C<undef>, then it will be matched against
759the receive buffer when neither C<$accept> nor C<$reject> match,
760and everything preceding and including the match will be accepted
761unconditionally. This is useful to skip large amounts of data that you
762know cannot be matched, so that the C<$accept> or C<$reject> regex do not
763have to start matching from the beginning. This is purely an optimisation
764and is usually worth only when you expect more than a few kilobytes.
765
766Example: expect a http header, which ends at C<\015\012\015\012>. Since we
767expect the header to be very large (it isn't in practise, but...), we use
768a skip regex to skip initial portions. The skip regex is tricky in that
769it only accepts something not ending in either \015 or \012, as these are
770required for the accept regex.
771
772 $handle->push_read (regex =>
773 qr<\015\012\015\012>,
774 undef, # no reject
775 qr<^.*[^\015\012]>,
776 sub { ... });
777
778=cut
779
780register_read_type regex => sub {
781 my ($self, $cb, $accept, $reject, $skip) = @_;
782
783 my $data;
784 my $rbuf = \$self->{rbuf};
785
786 sub {
787 # accept
788 if ($$rbuf =~ $accept) {
789 $data .= substr $$rbuf, 0, $+[0], "";
790 $cb->($self, $data);
791 return 1;
792 }
793
794 # reject
795 if ($reject && $$rbuf =~ $reject) {
796 $! = &Errno::EBADMSG;
797 $self->error;
798 }
799
800 # skip
801 if ($skip && $$rbuf =~ $skip) {
802 $data .= substr $$rbuf, 0, $+[0], "";
803 }
804
805 ()
806 }
807};
808
643=back 809=back
810
811=item AnyEvent::Handle::register_read_type type => $coderef->($self, $cb, @args)
812
813This function (not method) lets you add your own types to C<push_read>.
814
815Whenever the given C<type> is used, C<push_read> will invoke the code
816reference with the handle object, the callback and the remaining
817arguments.
818
819The code reference is supposed to return a callback (usually a closure)
820that works as a plain read callback (see C<< ->push_read ($cb) >>).
821
822It should invoke the passed callback when it is done reading (remember to
823pass C<$self> as first argument as all other callbacks do that).
824
825Note that this is a function, and all types registered this way will be
826global, so try to use unique names.
827
828For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>,
829search for C<register_read_type>)).
644 830
645=item $handle->stop_read 831=item $handle->stop_read
646 832
647=item $handle->start_read 833=item $handle->start_read
648 834
… …
677 } elsif (defined $len) { 863 } elsif (defined $len) {
678 delete $self->{rw}; 864 delete $self->{rw};
679 $self->{eof} = 1; 865 $self->{eof} = 1;
680 $self->_drain_rbuf; 866 $self->_drain_rbuf;
681 867
682 } elsif ($! != EAGAIN && $! != EINTR) { 868 } elsif ($! != EAGAIN && $! != EINTR && $! != &AnyEvent::Util::WSAWOULDBLOCK) {
683 return $self->error; 869 return $self->error;
684 } 870 }
685 }); 871 });
686 } 872 }
687} 873}
… …
753 # but the openssl maintainers basically said: "trust us, it just works". 939 # but the openssl maintainers basically said: "trust us, it just works".
754 # (unfortunately, we have to hardcode constants because the abysmally misdesigned 940 # (unfortunately, we have to hardcode constants because the abysmally misdesigned
755 # and mismaintained ssleay-module doesn't even offer them). 941 # and mismaintained ssleay-module doesn't even offer them).
756 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html 942 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
757 Net::SSLeay::CTX_set_mode ($self->{tls}, 943 Net::SSLeay::CTX_set_mode ($self->{tls},
758 (eval { Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1) 944 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
759 | (eval { Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2)); 945 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
760 946
761 $self->{tls_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 947 $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 ()); 948 $self->{tls_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
763 949
764 Net::SSLeay::set_bio ($ssl, $self->{tls_rbio}, $self->{tls_wbio}); 950 Net::SSLeay::set_bio ($ssl, $self->{tls_rbio}, $self->{tls_wbio});

Diff Legend

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