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.31 by root, Sun May 25 00:08:49 2008 UTC

… …
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}) {
293 Scalar::Util::weaken $self; 294 Scalar::Util::weaken $self;
294 my $cb = sub { 295 my $cb = sub {
295 my $len = syswrite $self->{fh}, $self->{wbuf}; 296 my $len = syswrite $self->{fh}, $self->{wbuf};
296 297
297 if ($len > 0) { 298 if ($len >= 0) {
298 substr $self->{wbuf}, 0, $len, ""; 299 substr $self->{wbuf}, 0, $len, "";
299 300
300 $self->{on_drain}($self) 301 $self->{on_drain}($self)
301 if $self->{low_water_mark} >= length $self->{wbuf} 302 if $self->{low_water_mark} >= length $self->{wbuf}
302 && $self->{on_drain}; 303 && $self->{on_drain};
… …
311 312
312 $cb->($self); 313 $cb->($self);
313 }; 314 };
314} 315}
315 316
317our %WH;
318
319sub register_write_type($$) {
320 $WH{$_[0]} = $_[1];
321}
322
316sub push_write { 323sub push_write {
317 my $self = shift; 324 my $self = shift;
325
326 if (@_ > 1) {
327 my $type = shift;
328
329 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
330 ->($self, @_);
331 }
318 332
319 if ($self->{filter_w}) { 333 if ($self->{filter_w}) {
320 $self->{filter_w}->($self, \$_[0]); 334 $self->{filter_w}->($self, \$_[0]);
321 } else { 335 } else {
322 $self->{wbuf} .= $_[0]; 336 $self->{wbuf} .= $_[0];
323 $self->_drain_wbuf; 337 $self->_drain_wbuf;
324 } 338 }
325} 339}
340
341=item $handle->push_write (type => @args)
342
343=item $handle->unshift_write (type => @args)
344
345Instead of formatting your data yourself, you can also let this module do
346the job by specifying a type and type-specific arguments.
347
348Predefined types are (if you have ideas for additional types, feel free to
349drop by and tell us):
350
351=over 4
352
353=item netstring => $string
354
355Formats the given value as netstring
356(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
357
358=back
359
360=cut
361
362register_write_type netstring => sub {
363 my ($self, $string) = @_;
364
365 sprintf "%d:%s,", (length $string), $string
366};
367
368=item AnyEvent::Handle::register_write_type type => $coderef->($self, @args)
369
370This function (not method) lets you add your own types to C<push_write>.
371Whenever the given C<type> is used, C<push_write> will invoke the code
372reference with the handle object and the remaining arguments.
373
374The code reference is supposed to return a single octet string that will
375be appended to the write buffer.
376
377Note that this is a function, and all types registered this way will be
378global, so try to use unique names.
379
380=cut
326 381
327############################################################################# 382#############################################################################
328 383
329=back 384=back
330 385
… …
418 local $self->{in_drain} = 1; 473 local $self->{in_drain} = 1;
419 474
420 while (my $len = length $self->{rbuf}) { 475 while (my $len = length $self->{rbuf}) {
421 no strict 'refs'; 476 no strict 'refs';
422 if (my $cb = shift @{ $self->{queue} }) { 477 if (my $cb = shift @{ $self->{queue} }) {
423 if (!$cb->($self)) { 478 unless ($cb->($self)) {
424 if ($self->{eof}) { 479 if ($self->{eof}) {
425 # no progress can be made (not enough data and no data forthcoming) 480 # no progress can be made (not enough data and no data forthcoming)
426 $! = &Errno::EPIPE; return $self->error; 481 $! = &Errno::EPIPE; return $self->error;
427 } 482 }
428 483
… …
505interested in (which can be none at all) and return a true value. After returning 560interested in (which can be none at all) and return a true value. After returning
506true, it will be removed from the queue. 561true, it will be removed from the queue.
507 562
508=cut 563=cut
509 564
565our %RH;
566
567sub register_read_type($$) {
568 $RH{$_[0]} = $_[1];
569}
570
510sub push_read { 571sub push_read {
511 my ($self, $cb) = @_; 572 my $self = shift;
573 my $cb = pop;
574
575 if (@_) {
576 my $type = shift;
577
578 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
579 ->($self, $cb, @_);
580 }
512 581
513 push @{ $self->{queue} }, $cb; 582 push @{ $self->{queue} }, $cb;
514 $self->_drain_rbuf; 583 $self->_drain_rbuf;
515} 584}
516 585
517sub unshift_read { 586sub unshift_read {
518 my ($self, $cb) = @_; 587 my $self = shift;
588 my $cb = pop;
519 589
590 if (@_) {
591 my $type = shift;
592
593 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
594 ->($self, $cb, @_);
595 }
596
597
520 push @{ $self->{queue} }, $cb; 598 unshift @{ $self->{queue} }, $cb;
521 $self->_drain_rbuf; 599 $self->_drain_rbuf;
522} 600}
523 601
524=item $handle->push_read_chunk ($len, $cb->($self, $data)) 602=item $handle->push_read (type => @args, $cb)
525 603
526=item $handle->unshift_read_chunk ($len, $cb->($self, $data)) 604=item $handle->unshift_read (type => @args, $cb)
527 605
528Append the given callback to the end of the queue (C<push_read_chunk>) or 606Instead of providing a callback that parses the data itself you can chose
529prepend it (C<unshift_read_chunk>). 607between a number of predefined parsing formats, for chunks of data, lines
608etc.
530 609
531The callback will be called only once C<$len> bytes have been read, and 610Predefined types are (if you have ideas for additional types, feel free to
532these C<$len> bytes will be passed to the callback. 611drop by and tell us):
533 612
534=cut 613=over 4
535 614
536sub _read_chunk($$) { 615=item chunk => $octets, $cb->($self, $data)
616
617Invoke the callback only once C<$octets> bytes have been read. Pass the
618data read to the callback. The callback will never be called with less
619data.
620
621Example: read 2 bytes.
622
623 $handle->push_read (chunk => 2, sub {
624 warn "yay ", unpack "H*", $_[1];
625 });
626
627=cut
628
629register_read_type chunk => sub {
537 my ($self, $len, $cb) = @_; 630 my ($self, $cb, $len) = @_;
538 631
539 sub { 632 sub {
540 $len <= length $_[0]{rbuf} or return; 633 $len <= length $_[0]{rbuf} or return;
541 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, ""); 634 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
542 1 635 1
543 } 636 }
544} 637};
545 638
639# compatibility with older API
546sub push_read_chunk { 640sub push_read_chunk {
547 $_[0]->push_read (&_read_chunk); 641 $_[0]->push_read (chunk => $_[1], $_[2]);
548} 642}
549
550 643
551sub unshift_read_chunk { 644sub unshift_read_chunk {
552 $_[0]->unshift_read (&_read_chunk); 645 $_[0]->unshift_read (chunk => $_[1], $_[2]);
553} 646}
554 647
555=item $handle->push_read_line ([$eol, ]$cb->($self, $line, $eol)) 648=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 649
562The callback will be called only once a full line (including the end of 650The 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 651line 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 652marker) will be passed to the callback as second argument (C<$line>), and
565the end of line marker as the third argument (C<$eol>). 653the 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 664Partial lines at the end of the stream will never be returned, as they are
577not marked by the end of line marker. 665not marked by the end of line marker.
578 666
579=cut 667=cut
580 668
581sub _read_line($$) { 669register_read_type line => sub {
582 my $self = shift; 670 my ($self, $cb, $eol) = @_;
583 my $cb = pop;
584 my $eol = @_ ? shift : qr|(\015?\012)|;
585 my $pos;
586 671
672 $eol = qr|(\015?\012)| if @_ < 3;
587 $eol = quotemeta $eol unless ref $eol; 673 $eol = quotemeta $eol unless ref $eol;
588 $eol = qr|^(.*?)($eol)|s; 674 $eol = qr|^(.*?)($eol)|s;
589 675
590 sub { 676 sub {
591 $_[0]{rbuf} =~ s/$eol// or return; 677 $_[0]{rbuf} =~ s/$eol// or return;
592 678
593 $cb->($_[0], $1, $2); 679 $cb->($_[0], $1, $2);
594 1 680 1
595 } 681 }
596} 682};
597 683
684# compatibility with older API
598sub push_read_line { 685sub push_read_line {
599 $_[0]->push_read (&_read_line); 686 my $self = shift;
687 $self->push_read (line => @_);
600} 688}
601 689
602sub unshift_read_line { 690sub unshift_read_line {
603 $_[0]->unshift_read (&_read_line); 691 my $self = shift;
692 $self->unshift_read (line => @_);
604} 693}
694
695=item netstring => $cb->($string)
696
697A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
698
699Throws an error with C<$!> set to EBADMSG on format violations.
700
701=cut
702
703register_read_type netstring => sub {
704 my ($self, $cb) = @_;
705
706 sub {
707 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
708 if ($_[0]{rbuf} =~ /[^0-9]/) {
709 $! = &Errno::EBADMSG;
710 $self->error;
711 }
712 return;
713 }
714
715 my $len = $1;
716
717 $self->unshift_read (chunk => $len, sub {
718 my $string = $_[1];
719 $_[0]->unshift_read (chunk => 1, sub {
720 if ($_[1] eq ",") {
721 $cb->($_[0], $string);
722 } else {
723 $! = &Errno::EBADMSG;
724 $self->error;
725 }
726 });
727 });
728
729 1
730 }
731};
732
733=back
734
735=item AnyEvent::Handle::register_read_type type => $coderef->($self, $cb, @args)
736
737This function (not method) lets you add your own types to C<push_read>.
738
739Whenever the given C<type> is used, C<push_read> will invoke the code
740reference with the handle object, the callback and the remaining
741arguments.
742
743The code reference is supposed to return a callback (usually a closure)
744that works as a plain read callback (see C<< ->push_read ($cb) >>).
745
746It should invoke the passed callback when it is done reading (remember to
747pass C<$self> as first argument as all other callbacks do that).
748
749Note that this is a function, and all types registered this way will be
750global, so try to use unique names.
751
752For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>,
753search for C<register_read_type>)).
605 754
606=item $handle->stop_read 755=item $handle->stop_read
607 756
608=item $handle->start_read 757=item $handle->start_read
609 758

Diff Legend

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