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.27 by root, Sat May 24 15:26:04 2008 UTC vs.
Revision 1.35 by root, Mon May 26 05:46:35 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
… …
211 } 212 }
212 213
213 if ($self->{on_error}) { 214 if ($self->{on_error}) {
214 $self->{on_error}($self); 215 $self->{on_error}($self);
215 } else { 216 } else {
216 die "AnyEvent::Handle uncaught fatal error: $!"; 217 Carp::croak "AnyEvent::Handle uncaught fatal error: $!";
217 } 218 }
218} 219}
219 220
220=item $fh = $handle->fh 221=item $fh = $handle->fh
221 222
… …
287=cut 288=cut
288 289
289sub _drain_wbuf { 290sub _drain_wbuf {
290 my ($self) = @_; 291 my ($self) = @_;
291 292
292 unless ($self->{ww}) { 293 if (!$self->{ww} && length $self->{wbuf}) {
294
293 Scalar::Util::weaken $self; 295 Scalar::Util::weaken $self;
296
294 my $cb = sub { 297 my $cb = sub {
295 my $len = syswrite $self->{fh}, $self->{wbuf}; 298 my $len = syswrite $self->{fh}, $self->{wbuf};
296 299
297 if ($len > 0) { 300 if ($len >= 0) {
298 substr $self->{wbuf}, 0, $len, ""; 301 substr $self->{wbuf}, 0, $len, "";
299 302
300 $self->{on_drain}($self) 303 $self->{on_drain}($self)
301 if $self->{low_water_mark} >= length $self->{wbuf} 304 if $self->{low_water_mark} >= length $self->{wbuf}
302 && $self->{on_drain}; 305 && $self->{on_drain};
303 306
304 delete $self->{ww} unless length $self->{wbuf}; 307 delete $self->{ww} unless length $self->{wbuf};
305 } elsif ($! != EAGAIN && $! != EINTR) { 308 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAWOULDBLOCK) {
306 $self->error; 309 $self->error;
307 } 310 }
308 }; 311 };
309 312
313 # try to write data immediately
314 $cb->();
315
316 # if still data left in wbuf, we need to poll
310 $self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb); 317 $self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
311 318 if length $self->{wbuf};
312 $cb->($self);
313 }; 319 };
320}
321
322our %WH;
323
324sub register_write_type($$) {
325 $WH{$_[0]} = $_[1];
314} 326}
315 327
316sub push_write { 328sub push_write {
317 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 }
318 337
319 if ($self->{filter_w}) { 338 if ($self->{filter_w}) {
320 $self->{filter_w}->($self, \$_[0]); 339 $self->{filter_w}->($self, \$_[0]);
321 } else { 340 } else {
322 $self->{wbuf} .= $_[0]; 341 $self->{wbuf} .= $_[0];
323 $self->_drain_wbuf; 342 $self->_drain_wbuf;
324 } 343 }
325} 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
326 386
327############################################################################# 387#############################################################################
328 388
329=back 389=back
330 390
… …
418 local $self->{in_drain} = 1; 478 local $self->{in_drain} = 1;
419 479
420 while (my $len = length $self->{rbuf}) { 480 while (my $len = length $self->{rbuf}) {
421 no strict 'refs'; 481 no strict 'refs';
422 if (my $cb = shift @{ $self->{queue} }) { 482 if (my $cb = shift @{ $self->{queue} }) {
423 if (!$cb->($self)) { 483 unless ($cb->($self)) {
424 if ($self->{eof}) { 484 if ($self->{eof}) {
425 # 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)
426 $! = &Errno::EPIPE; return $self->error; 486 $! = &Errno::EPIPE; return $self->error;
427 } 487 }
428 488
… …
505interested 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
506true, it will be removed from the queue. 566true, it will be removed from the queue.
507 567
508=cut 568=cut
509 569
570our %RH;
571
572sub register_read_type($$) {
573 $RH{$_[0]} = $_[1];
574}
575
510sub push_read { 576sub push_read {
511 my ($self, $cb) = @_; 577 my $self = shift;
578 my $cb = pop;
579
580 if (@_) {
581 my $type = shift;
582
583 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
584 ->($self, $cb, @_);
585 }
512 586
513 push @{ $self->{queue} }, $cb; 587 push @{ $self->{queue} }, $cb;
514 $self->_drain_rbuf; 588 $self->_drain_rbuf;
515} 589}
516 590
517sub unshift_read { 591sub unshift_read {
518 my ($self, $cb) = @_; 592 my $self = shift;
593 my $cb = pop;
519 594
595 if (@_) {
596 my $type = shift;
597
598 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
599 ->($self, $cb, @_);
600 }
601
602
520 push @{ $self->{queue} }, $cb; 603 unshift @{ $self->{queue} }, $cb;
521 $self->_drain_rbuf; 604 $self->_drain_rbuf;
522} 605}
523 606
524=item $handle->push_read_chunk ($len, $cb->($self, $data)) 607=item $handle->push_read (type => @args, $cb)
525 608
526=item $handle->unshift_read_chunk ($len, $cb->($self, $data)) 609=item $handle->unshift_read (type => @args, $cb)
527 610
528Append the given callback to the end of the queue (C<push_read_chunk>) or 611Instead of providing a callback that parses the data itself you can chose
529prepend it (C<unshift_read_chunk>). 612between a number of predefined parsing formats, for chunks of data, lines
613etc.
530 614
531The callback will be called only once C<$len> bytes have been read, and 615Predefined types are (if you have ideas for additional types, feel free to
532these C<$len> bytes will be passed to the callback. 616drop by and tell us):
533 617
534=cut 618=over 4
535 619
536sub _read_chunk($$) { 620=item chunk => $octets, $cb->($self, $data)
621
622Invoke the callback only once C<$octets> bytes have been read. Pass the
623data read to the callback. The callback will never be called with less
624data.
625
626Example: read 2 bytes.
627
628 $handle->push_read (chunk => 2, sub {
629 warn "yay ", unpack "H*", $_[1];
630 });
631
632=cut
633
634register_read_type chunk => sub {
537 my ($self, $len, $cb) = @_; 635 my ($self, $cb, $len) = @_;
538 636
539 sub { 637 sub {
540 $len <= length $_[0]{rbuf} or return; 638 $len <= length $_[0]{rbuf} or return;
541 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, ""); 639 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
542 1 640 1
543 } 641 }
544} 642};
545 643
644# compatibility with older API
546sub push_read_chunk { 645sub push_read_chunk {
547 $_[0]->push_read (&_read_chunk); 646 $_[0]->push_read (chunk => $_[1], $_[2]);
548} 647}
549
550 648
551sub unshift_read_chunk { 649sub unshift_read_chunk {
552 $_[0]->unshift_read (&_read_chunk); 650 $_[0]->unshift_read (chunk => $_[1], $_[2]);
553} 651}
554 652
555=item $handle->push_read_line ([$eol, ]$cb->($self, $line, $eol)) 653=item line => [$eol, ]$cb->($self, $line, $eol)
556
557=item $handle->unshift_read_line ([$eol, ]$cb->($self, $line, $eol))
558
559Append the given callback to the end of the queue (C<push_read_line>) or
560prepend it (C<unshift_read_line>).
561 654
562The callback will be called only once a full line (including the end of 655The callback will be called only once a full line (including the end of
563line marker, C<$eol>) has been read. This line (excluding the end of line 656line marker, C<$eol>) has been read. This line (excluding the end of line
564marker) will be passed to the callback as second argument (C<$line>), and 657marker) will be passed to the callback as second argument (C<$line>), and
565the end of line marker as the third argument (C<$eol>). 658the end of line marker as the third argument (C<$eol>).
… …
576Partial lines at the end of the stream will never be returned, as they are 669Partial lines at the end of the stream will never be returned, as they are
577not marked by the end of line marker. 670not marked by the end of line marker.
578 671
579=cut 672=cut
580 673
581sub _read_line($$) { 674register_read_type line => sub {
582 my $self = shift; 675 my ($self, $cb, $eol) = @_;
583 my $cb = pop;
584 my $eol = @_ ? shift : qr|(\015?\012)|;
585 my $pos;
586 676
677 $eol = qr|(\015?\012)| if @_ < 3;
587 $eol = quotemeta $eol unless ref $eol; 678 $eol = quotemeta $eol unless ref $eol;
588 $eol = qr|^(.*?)($eol)|s; 679 $eol = qr|^(.*?)($eol)|s;
589 680
590 sub { 681 sub {
591 $_[0]{rbuf} =~ s/$eol// or return; 682 $_[0]{rbuf} =~ s/$eol// or return;
592 683
593 $cb->($_[0], $1, $2); 684 $cb->($_[0], $1, $2);
594 1 685 1
595 } 686 }
596} 687};
597 688
689# compatibility with older API
598sub push_read_line { 690sub push_read_line {
599 $_[0]->push_read (&_read_line); 691 my $self = shift;
692 $self->push_read (line => @_);
600} 693}
601 694
602sub unshift_read_line { 695sub unshift_read_line {
603 $_[0]->unshift_read (&_read_line); 696 my $self = shift;
697 $self->unshift_read (line => @_);
604} 698}
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=back
739
740=item AnyEvent::Handle::register_read_type type => $coderef->($self, $cb, @args)
741
742This function (not method) lets you add your own types to C<push_read>.
743
744Whenever the given C<type> is used, C<push_read> will invoke the code
745reference with the handle object, the callback and the remaining
746arguments.
747
748The code reference is supposed to return a callback (usually a closure)
749that works as a plain read callback (see C<< ->push_read ($cb) >>).
750
751It should invoke the passed callback when it is done reading (remember to
752pass C<$self> as first argument as all other callbacks do that).
753
754Note that this is a function, and all types registered this way will be
755global, so try to use unique names.
756
757For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>,
758search for C<register_read_type>)).
605 759
606=item $handle->stop_read 760=item $handle->stop_read
607 761
608=item $handle->start_read 762=item $handle->start_read
609 763
… …
638 } elsif (defined $len) { 792 } elsif (defined $len) {
639 delete $self->{rw}; 793 delete $self->{rw};
640 $self->{eof} = 1; 794 $self->{eof} = 1;
641 $self->_drain_rbuf; 795 $self->_drain_rbuf;
642 796
643 } elsif ($! != EAGAIN && $! != EINTR) { 797 } elsif ($! != EAGAIN && $! != EINTR && $! != &AnyEvent::Util::WSAWOULDBLOCK) {
644 return $self->error; 798 return $self->error;
645 } 799 }
646 }); 800 });
647 } 801 }
648} 802}
… …
714 # but the openssl maintainers basically said: "trust us, it just works". 868 # but the openssl maintainers basically said: "trust us, it just works".
715 # (unfortunately, we have to hardcode constants because the abysmally misdesigned 869 # (unfortunately, we have to hardcode constants because the abysmally misdesigned
716 # and mismaintained ssleay-module doesn't even offer them). 870 # and mismaintained ssleay-module doesn't even offer them).
717 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html 871 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
718 Net::SSLeay::CTX_set_mode ($self->{tls}, 872 Net::SSLeay::CTX_set_mode ($self->{tls},
719 (eval { Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1) 873 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
720 | (eval { Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2)); 874 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
721 875
722 $self->{tls_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 876 $self->{tls_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
723 $self->{tls_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 877 $self->{tls_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
724 878
725 Net::SSLeay::set_bio ($ssl, $self->{tls_rbio}, $self->{tls_wbio}); 879 Net::SSLeay::set_bio ($ssl, $self->{tls_rbio}, $self->{tls_wbio});

Diff Legend

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