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.37 by root, Mon May 26 20:02:22 2008 UTC vs.
Revision 1.46 by root, Thu May 29 00:22:36 2008 UTC

2 2
3no warnings; 3no warnings;
4use strict; 4use strict;
5 5
6use AnyEvent (); 6use AnyEvent ();
7use AnyEvent::Util qw(WSAWOULDBLOCK); 7use AnyEvent::Util qw(WSAEWOULDBLOCK);
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
73The filehandle this L<AnyEvent::Handle> object will operate on. 73The filehandle this L<AnyEvent::Handle> object will operate on.
74 74
75NOTE: The filehandle will be set to non-blocking (using 75NOTE: The filehandle will be set to non-blocking (using
76AnyEvent::Util::fh_nonblocking). 76AnyEvent::Util::fh_nonblocking).
77 77
78=item on_eof => $cb->($self) 78=item on_eof => $cb->($handle)
79 79
80Set the callback to be called on EOF. 80Set the callback to be called on EOF.
81 81
82While not mandatory, it is highly recommended to set an eof callback, 82While not mandatory, it is highly recommended to set an eof callback,
83otherwise you might end up with a closed socket while you are still 83otherwise you might end up with a closed socket while you are still
84waiting for data. 84waiting for data.
85 85
86=item on_error => $cb->($self) 86=item on_error => $cb->($handle)
87 87
88This is the fatal error callback, that is called when, well, a fatal error 88This is the fatal error callback, that is called when, well, a fatal error
89occurs, such as not being able to resolve the hostname, failure to connect 89occurs, such as not being able to resolve the hostname, failure to connect
90or a read error. 90or a read error.
91 91
92The 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
93called. 93called.
94 94
95On callback entrance, the value of C<$!> contains the operating system 95On callback entrance, the value of C<$!> contains the operating system
96error (or C<ENOSPC>, C<EPIPE> or C<EBADMSG>). 96error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>).
97 97
98The callbakc should throw an exception. If it returns, then 98The callback should throw an exception. If it returns, then
99AnyEvent::Handle will C<croak> for you. 99AnyEvent::Handle will C<croak> for you.
100 100
101While 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
102you will not be notified of errors otherwise. The default simply calls 102you will not be notified of errors otherwise. The default simply calls
103die. 103die.
104 104
105=item on_read => $cb->($self) 105=item on_read => $cb->($handle)
106 106
107This sets the default read callback, which is called when data arrives 107This sets the default read callback, which is called when data arrives
108and no read request is in the queue. 108and no read request is in the queue.
109 109
110To access (and remove data from) the read buffer, use the C<< ->rbuf >> 110To access (and remove data from) the read buffer, use the C<< ->rbuf >>
111method or access the C<$self->{rbuf}> member directly. 111method or access the C<$handle->{rbuf}> member directly.
112 112
113When an EOF condition is detected then AnyEvent::Handle will first try to 113When an EOF condition is detected then AnyEvent::Handle will first try to
114feed all the remaining data to the queued callbacks and C<on_read> before 114feed all the remaining data to the queued callbacks and C<on_read> before
115calling the C<on_eof> callback. If no progress can be made, then a fatal 115calling the C<on_eof> callback. If no progress can be made, then a fatal
116error will be raised (with C<$!> set to C<EPIPE>). 116error will be raised (with C<$!> set to C<EPIPE>).
117 117
118=item on_drain => $cb->() 118=item on_drain => $cb->($handle)
119 119
120This sets the callback that is called when the write buffer becomes empty 120This sets the callback that is called when the write buffer becomes empty
121(or when the callback is set and the buffer is empty already). 121(or when the callback is set and the buffer is empty already).
122 122
123To append to the write buffer, use the C<< ->push_write >> method. 123To append to the write buffer, use the C<< ->push_write >> method.
124
125=item timeout => $fractional_seconds
126
127If non-zero, then this enables an "inactivity" timeout: whenever this many
128seconds pass without a successful read or write on the underlying file
129handle, the C<on_timeout> callback will be invoked (and if that one is
130missing, an C<ETIMEDOUT> error will be raised).
131
132Note that timeout processing is also active when you currently do not have
133any outstanding read or write requests: If you plan to keep the connection
134idle then you should disable the timout temporarily or ignore the timeout
135in the C<on_timeout> callback.
136
137Zero (the default) disables this timeout.
138
139=item on_timeout => $cb->($handle)
140
141Called whenever the inactivity timeout passes. If you return from this
142callback, then the timeout will be reset as if some activity had happened,
143so this condition is not fatal in any way.
124 144
125=item rbuf_max => <bytes> 145=item rbuf_max => <bytes>
126 146
127If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>) 147If defined, then a fatal error will be raised (with C<$!> set to C<ENOSPC>)
128when the read buffer ever (strictly) exceeds this size. This is useful to 148when the read buffer ever (strictly) exceeds this size. This is useful to
135isn't finished). 155isn't finished).
136 156
137=item read_size => <bytes> 157=item read_size => <bytes>
138 158
139The default read block size (the amount of bytes this module will try to read 159The default read block size (the amount of bytes this module will try to read
140on each [loop iteration). Default: C<4096>. 160during each (loop iteration). Default: C<8192>.
141 161
142=item low_water_mark => <bytes> 162=item low_water_mark => <bytes>
143 163
144Sets the amount of bytes (default: C<0>) that make up an "empty" write 164Sets the amount of bytes (default: C<0>) that make up an "empty" write
145buffer: If the write reaches this size or gets even samller it is 165buffer: If the write reaches this size or gets even samller it is
168 188
169Use the given Net::SSLeay::CTX object to create the new TLS connection 189Use the given Net::SSLeay::CTX object to create the new TLS connection
170(unless a connection object was specified directly). If this parameter is 190(unless a connection object was specified directly). If this parameter is
171missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>. 191missing, then AnyEvent::Handle will use C<AnyEvent::Handle::TLS_CTX>.
172 192
193=item json => JSON or JSON::XS object
194
195This is the json coder object used by the C<json> read and write types.
196
197If you don't supply it, then AnyEvent::Handle will create and use a
198suitable one, which will write and expect UTF-8 encoded JSON texts.
199
200Note that you are responsible to depend on the JSON module if you want to
201use this functionality, as AnyEvent does not have a dependency itself.
202
203=item filter_r => $cb
204
205=item filter_w => $cb
206
207These exist, but are undocumented at this time.
208
173=back 209=back
174 210
175=cut 211=cut
176 212
177sub new { 213sub new {
186 if ($self->{tls}) { 222 if ($self->{tls}) {
187 require Net::SSLeay; 223 require Net::SSLeay;
188 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 224 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
189 } 225 }
190 226
191 $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; 227# $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; # nop
192 $self->on_error (delete $self->{on_error}) if $self->{on_error}; 228# $self->on_error (delete $self->{on_error}) if $self->{on_error}; # nop
229# $self->on_read (delete $self->{on_read} ) if $self->{on_read}; # nop
193 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain}; 230 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
194 $self->on_read (delete $self->{on_read} ) if $self->{on_read}; 231
232 $self->{_activity} = AnyEvent->now;
233 $self->_timeout;
195 234
196 $self->start_read; 235 $self->start_read;
197 236
198 $self 237 $self
199} 238}
200 239
201sub _shutdown { 240sub _shutdown {
202 my ($self) = @_; 241 my ($self) = @_;
203 242
243 delete $self->{_tw};
204 delete $self->{rw}; 244 delete $self->{_rw};
205 delete $self->{ww}; 245 delete $self->{_ww};
206 delete $self->{fh}; 246 delete $self->{fh};
207} 247}
208 248
209sub error { 249sub error {
210 my ($self) = @_; 250 my ($self) = @_;
224 264
225This method returns the file handle of the L<AnyEvent::Handle> object. 265This method returns the file handle of the L<AnyEvent::Handle> object.
226 266
227=cut 267=cut
228 268
229sub fh { $_[0]->{fh} } 269sub fh { $_[0]{fh} }
230 270
231=item $handle->on_error ($cb) 271=item $handle->on_error ($cb)
232 272
233Replace the current C<on_error> callback (see the C<on_error> constructor argument). 273Replace the current C<on_error> callback (see the C<on_error> constructor argument).
234 274
244 284
245=cut 285=cut
246 286
247sub on_eof { 287sub on_eof {
248 $_[0]{on_eof} = $_[1]; 288 $_[0]{on_eof} = $_[1];
289}
290
291=item $handle->on_timeout ($cb)
292
293Replace the current C<on_timeout> callback, or disables the callback
294(but not the timeout) if C<$cb> = C<undef>. See C<timeout> constructor
295argument.
296
297=cut
298
299sub on_timeout {
300 $_[0]{on_timeout} = $_[1];
301}
302
303#############################################################################
304
305=item $handle->timeout ($seconds)
306
307Configures (or disables) the inactivity timeout.
308
309=cut
310
311sub timeout {
312 my ($self, $timeout) = @_;
313
314 $self->{timeout} = $timeout;
315 $self->_timeout;
316}
317
318# reset the timeout watcher, as neccessary
319# also check for time-outs
320sub _timeout {
321 my ($self) = @_;
322
323 if ($self->{timeout}) {
324 my $NOW = AnyEvent->now;
325
326 # when would the timeout trigger?
327 my $after = $self->{_activity} + $self->{timeout} - $NOW;
328
329 # now or in the past already?
330 if ($after <= 0) {
331 $self->{_activity} = $NOW;
332
333 if ($self->{on_timeout}) {
334 $self->{on_timeout}->($self);
335 } else {
336 $! = Errno::ETIMEDOUT;
337 $self->error;
338 }
339
340 # callbakx could have changed timeout value, optimise
341 return unless $self->{timeout};
342
343 # calculate new after
344 $after = $self->{timeout};
345 }
346
347 Scalar::Util::weaken $self;
348
349 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
350 delete $self->{_tw};
351 $self->_timeout;
352 });
353 } else {
354 delete $self->{_tw};
355 }
249} 356}
250 357
251############################################################################# 358#############################################################################
252 359
253=back 360=back
290=cut 397=cut
291 398
292sub _drain_wbuf { 399sub _drain_wbuf {
293 my ($self) = @_; 400 my ($self) = @_;
294 401
295 if (!$self->{ww} && length $self->{wbuf}) { 402 if (!$self->{_ww} && length $self->{wbuf}) {
296 403
297 Scalar::Util::weaken $self; 404 Scalar::Util::weaken $self;
298 405
299 my $cb = sub { 406 my $cb = sub {
300 my $len = syswrite $self->{fh}, $self->{wbuf}; 407 my $len = syswrite $self->{fh}, $self->{wbuf};
301 408
302 if ($len >= 0) { 409 if ($len >= 0) {
303 substr $self->{wbuf}, 0, $len, ""; 410 substr $self->{wbuf}, 0, $len, "";
411
412 $self->{_activity} = AnyEvent->now;
304 413
305 $self->{on_drain}($self) 414 $self->{on_drain}($self)
306 if $self->{low_water_mark} >= length $self->{wbuf} 415 if $self->{low_water_mark} >= length $self->{wbuf}
307 && $self->{on_drain}; 416 && $self->{on_drain};
308 417
309 delete $self->{ww} unless length $self->{wbuf}; 418 delete $self->{_ww} unless length $self->{wbuf};
310 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAWOULDBLOCK) { 419 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
311 $self->error; 420 $self->error;
312 } 421 }
313 }; 422 };
314 423
315 # try to write data immediately 424 # try to write data immediately
316 $cb->(); 425 $cb->();
317 426
318 # if still data left in wbuf, we need to poll 427 # if still data left in wbuf, we need to poll
319 $self->{ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 428 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
320 if length $self->{wbuf}; 429 if length $self->{wbuf};
321 }; 430 };
322} 431}
323 432
324our %WH; 433our %WH;
370 my ($self, $string) = @_; 479 my ($self, $string) = @_;
371 480
372 sprintf "%d:%s,", (length $string), $string 481 sprintf "%d:%s,", (length $string), $string
373}; 482};
374 483
484=item json => $array_or_hashref
485
486Encodes the given hash or array reference into a JSON object. Unless you
487provide your own JSON object, this means it will be encoded to JSON text
488in UTF-8.
489
490JSON objects (and arrays) are self-delimiting, so you can write JSON at
491one end of a handle and read them at the other end without using any
492additional framing.
493
494The generated JSON text is guaranteed not to contain any newlines: While
495this module doesn't need delimiters after or between JSON texts to be
496able to read them, many other languages depend on that.
497
498A simple RPC protocol that interoperates easily with others is to send
499JSON arrays (or objects, although arrays are usually the better choice as
500they mimic how function argument passing works) and a newline after each
501JSON text:
502
503 $handle->push_write (json => ["method", "arg1", "arg2"]); # whatever
504 $handle->push_write ("\012");
505
506An AnyEvent::Handle receiver would simply use the C<json> read type and
507rely on the fact that the newline will be skipped as leading whitespace:
508
509 $handle->push_read (json => sub { my $array = $_[1]; ... });
510
511Other languages could read single lines terminated by a newline and pass
512this line into their JSON decoder of choice.
513
514=cut
515
516register_write_type json => sub {
517 my ($self, $ref) = @_;
518
519 require JSON;
520
521 $self->{json} ? $self->{json}->encode ($ref)
522 : JSON::encode_json ($ref)
523};
524
375=item AnyEvent::Handle::register_write_type type => $coderef->($self, @args) 525=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
376 526
377This function (not method) lets you add your own types to C<push_write>. 527This 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 528Whenever the given C<type> is used, C<push_write> will invoke the code
379reference with the handle object and the remaining arguments. 529reference with the handle object and the remaining arguments.
380 530
480 return if $self->{in_drain}; 630 return if $self->{in_drain};
481 local $self->{in_drain} = 1; 631 local $self->{in_drain} = 1;
482 632
483 while (my $len = length $self->{rbuf}) { 633 while (my $len = length $self->{rbuf}) {
484 no strict 'refs'; 634 no strict 'refs';
485 if (my $cb = shift @{ $self->{queue} }) { 635 if (my $cb = shift @{ $self->{_queue} }) {
486 unless ($cb->($self)) { 636 unless ($cb->($self)) {
487 if ($self->{eof}) { 637 if ($self->{_eof}) {
488 # no progress can be made (not enough data and no data forthcoming) 638 # no progress can be made (not enough data and no data forthcoming)
489 $! = &Errno::EPIPE; 639 $! = &Errno::EPIPE;
490 $self->error; 640 $self->error;
491 } 641 }
492 642
493 unshift @{ $self->{queue} }, $cb; 643 unshift @{ $self->{_queue} }, $cb;
494 return; 644 return;
495 } 645 }
496 } elsif ($self->{on_read}) { 646 } elsif ($self->{on_read}) {
497 $self->{on_read}($self); 647 $self->{on_read}($self);
498 648
499 if ( 649 if (
500 $self->{eof} # if no further data will arrive 650 $self->{_eof} # if no further data will arrive
501 && $len == length $self->{rbuf} # and no data has been consumed 651 && $len == length $self->{rbuf} # and no data has been consumed
502 && !@{ $self->{queue} } # and the queue is still empty 652 && !@{ $self->{_queue} } # and the queue is still empty
503 && $self->{on_read} # and we still want to read data 653 && $self->{on_read} # and we still want to read data
504 ) { 654 ) {
505 # then no progress can be made 655 # then no progress can be made
506 $! = &Errno::EPIPE; 656 $! = &Errno::EPIPE;
507 $self->error; 657 $self->error;
508 } 658 }
509 } else { 659 } else {
510 # read side becomes idle 660 # read side becomes idle
511 delete $self->{rw}; 661 delete $self->{_rw};
512 return; 662 return;
513 } 663 }
514 } 664 }
515 665
516 if ($self->{eof}) { 666 if ($self->{_eof}) {
517 $self->_shutdown; 667 $self->_shutdown;
518 $self->{on_eof}($self) 668 $self->{on_eof}($self)
519 if $self->{on_eof}; 669 if $self->{on_eof};
520 } 670 }
521} 671}
587 737
588 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 738 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
589 ->($self, $cb, @_); 739 ->($self, $cb, @_);
590 } 740 }
591 741
592 push @{ $self->{queue} }, $cb; 742 push @{ $self->{_queue} }, $cb;
593 $self->_drain_rbuf; 743 $self->_drain_rbuf;
594} 744}
595 745
596sub unshift_read { 746sub unshift_read {
597 my $self = shift; 747 my $self = shift;
603 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read") 753 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::unshift_read")
604 ->($self, $cb, @_); 754 ->($self, $cb, @_);
605 } 755 }
606 756
607 757
608 unshift @{ $self->{queue} }, $cb; 758 unshift @{ $self->{_queue} }, $cb;
609 $self->_drain_rbuf; 759 $self->_drain_rbuf;
610} 760}
611 761
612=item $handle->push_read (type => @args, $cb) 762=item $handle->push_read (type => @args, $cb)
613 763
620Predefined types are (if you have ideas for additional types, feel free to 770Predefined types are (if you have ideas for additional types, feel free to
621drop by and tell us): 771drop by and tell us):
622 772
623=over 4 773=over 4
624 774
625=item chunk => $octets, $cb->($self, $data) 775=item chunk => $octets, $cb->($handle, $data)
626 776
627Invoke the callback only once C<$octets> bytes have been read. Pass the 777Invoke the callback only once C<$octets> bytes have been read. Pass the
628data read to the callback. The callback will never be called with less 778data read to the callback. The callback will never be called with less
629data. 779data.
630 780
653 803
654sub unshift_read_chunk { 804sub unshift_read_chunk {
655 $_[0]->unshift_read (chunk => $_[1], $_[2]); 805 $_[0]->unshift_read (chunk => $_[1], $_[2]);
656} 806}
657 807
658=item line => [$eol, ]$cb->($self, $line, $eol) 808=item line => [$eol, ]$cb->($handle, $line, $eol)
659 809
660The callback will be called only once a full line (including the end of 810The callback will be called only once a full line (including the end of
661line marker, C<$eol>) has been read. This line (excluding the end of line 811line marker, C<$eol>) has been read. This line (excluding the end of line
662marker) will be passed to the callback as second argument (C<$line>), and 812marker) will be passed to the callback as second argument (C<$line>), and
663the end of line marker as the third argument (C<$eol>). 813the end of line marker as the third argument (C<$eol>).
700sub unshift_read_line { 850sub unshift_read_line {
701 my $self = shift; 851 my $self = shift;
702 $self->unshift_read (line => @_); 852 $self->unshift_read (line => @_);
703} 853}
704 854
705=item netstring => $cb->($string) 855=item netstring => $cb->($handle, $string)
706 856
707A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement). 857A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
708 858
709Throws an error with C<$!> set to EBADMSG on format violations. 859Throws an error with C<$!> set to EBADMSG on format violations.
710 860
738 888
739 1 889 1
740 } 890 }
741}; 891};
742 892
743=item regex => $accept[, $reject[, $skip], $cb->($data) 893=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
744 894
745Makes a regex match against the regex object C<$accept> and returns 895Makes a regex match against the regex object C<$accept> and returns
746everything up to and including the match. 896everything up to and including the match.
747 897
748Example: read a single line terminated by '\n'. 898Example: read a single line terminated by '\n'.
809 959
810 () 960 ()
811 } 961 }
812}; 962};
813 963
964=item json => $cb->($handle, $hash_or_arrayref)
965
966Reads a JSON object or array, decodes it and passes it to the callback.
967
968If a C<json> object was passed to the constructor, then that will be used
969for the final decode, otherwise it will create a JSON coder expecting UTF-8.
970
971This read type uses the incremental parser available with JSON version
9722.09 (and JSON::XS version 2.2) and above. You have to provide a
973dependency on your own: this module will load the JSON module, but
974AnyEvent does not depend on it itself.
975
976Since JSON texts are fully self-delimiting, the C<json> read and write
977types are an ideal simple RPC protocol: just exchange JSON datagrams. See
978the C<json> write type description, above, for an actual example.
979
980=cut
981
982register_read_type json => sub {
983 my ($self, $cb, $accept, $reject, $skip) = @_;
984
985 require JSON;
986
987 my $data;
988 my $rbuf = \$self->{rbuf};
989
990 my $json = $self->{json} ||= JSON->new->utf8;
991
992 sub {
993 my $ref = $json->incr_parse ($self->{rbuf});
994
995 if ($ref) {
996 $self->{rbuf} = $json->incr_text;
997 $json->incr_text = "";
998 $cb->($self, $ref);
999
1000 1
1001 } else {
1002 $self->{rbuf} = "";
1003 ()
1004 }
1005 }
1006};
1007
814=back 1008=back
815 1009
816=item AnyEvent::Handle::register_read_type type => $coderef->($self, $cb, @args) 1010=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args)
817 1011
818This function (not method) lets you add your own types to C<push_read>. 1012This function (not method) lets you add your own types to C<push_read>.
819 1013
820Whenever the given C<type> is used, C<push_read> will invoke the code 1014Whenever the given C<type> is used, C<push_read> will invoke the code
821reference with the handle object, the callback and the remaining 1015reference with the handle object, the callback and the remaining
823 1017
824The code reference is supposed to return a callback (usually a closure) 1018The code reference is supposed to return a callback (usually a closure)
825that works as a plain read callback (see C<< ->push_read ($cb) >>). 1019that works as a plain read callback (see C<< ->push_read ($cb) >>).
826 1020
827It should invoke the passed callback when it is done reading (remember to 1021It should invoke the passed callback when it is done reading (remember to
828pass C<$self> as first argument as all other callbacks do that). 1022pass C<$handle> as first argument as all other callbacks do that).
829 1023
830Note that this is a function, and all types registered this way will be 1024Note that this is a function, and all types registered this way will be
831global, so try to use unique names. 1025global, so try to use unique names.
832 1026
833For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>, 1027For examples, see the source of this module (F<perldoc -m AnyEvent::Handle>,
845=cut 1039=cut
846 1040
847sub stop_read { 1041sub stop_read {
848 my ($self) = @_; 1042 my ($self) = @_;
849 1043
850 delete $self->{rw}; 1044 delete $self->{_rw};
851} 1045}
852 1046
853sub start_read { 1047sub start_read {
854 my ($self) = @_; 1048 my ($self) = @_;
855 1049
856 unless ($self->{rw} || $self->{eof}) { 1050 unless ($self->{_rw} || $self->{_eof}) {
857 Scalar::Util::weaken $self; 1051 Scalar::Util::weaken $self;
858 1052
859 $self->{rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1053 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
860 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf}; 1054 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
861 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1055 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
862 1056
863 if ($len > 0) { 1057 if ($len > 0) {
1058 $self->{_activity} = AnyEvent->now;
1059
864 $self->{filter_r} 1060 $self->{filter_r}
865 ? $self->{filter_r}->($self, $rbuf) 1061 ? $self->{filter_r}->($self, $rbuf)
866 : $self->_drain_rbuf; 1062 : $self->_drain_rbuf;
867 1063
868 } elsif (defined $len) { 1064 } elsif (defined $len) {
869 delete $self->{rw}; 1065 delete $self->{_rw};
1066 delete $self->{_ww};
1067 delete $self->{_tw};
870 $self->{eof} = 1; 1068 $self->{_eof} = 1;
871 $self->_drain_rbuf; 1069 $self->_drain_rbuf;
872 1070
873 } elsif ($! != EAGAIN && $! != EINTR && $! != &AnyEvent::Util::WSAWOULDBLOCK) { 1071 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
874 return $self->error; 1072 return $self->error;
875 } 1073 }
876 }); 1074 });
877 } 1075 }
878} 1076}
879 1077
880sub _dotls { 1078sub _dotls {
881 my ($self) = @_; 1079 my ($self) = @_;
882 1080
883 if (length $self->{tls_wbuf}) { 1081 if (length $self->{_tls_wbuf}) {
884 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{tls_wbuf})) > 0) { 1082 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
885 substr $self->{tls_wbuf}, 0, $len, ""; 1083 substr $self->{_tls_wbuf}, 0, $len, "";
886 } 1084 }
887 } 1085 }
888 1086
889 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{tls_wbio}))) { 1087 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
890 $self->{wbuf} .= $buf; 1088 $self->{wbuf} .= $buf;
891 $self->_drain_wbuf; 1089 $self->_drain_wbuf;
892 } 1090 }
893 1091
894 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 1092 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) {
919The first argument is the same as the C<tls> constructor argument (either 1117The first argument is the same as the C<tls> constructor argument (either
920C<"connect">, C<"accept"> or an existing Net::SSLeay object). 1118C<"connect">, C<"accept"> or an existing Net::SSLeay object).
921 1119
922The second argument is the optional C<Net::SSLeay::CTX> object that is 1120The second argument is the optional C<Net::SSLeay::CTX> object that is
923used when AnyEvent::Handle has to create its own TLS connection object. 1121used when AnyEvent::Handle has to create its own TLS connection object.
1122
1123The TLS connection object will end up in C<< $handle->{tls} >> after this
1124call and can be used or changed to your liking. Note that the handshake
1125might have already started when this function returns.
924 1126
925=cut 1127=cut
926 1128
927# TODO: maybe document... 1129# TODO: maybe document...
928sub starttls { 1130sub starttls {
947 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html 1149 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
948 Net::SSLeay::CTX_set_mode ($self->{tls}, 1150 Net::SSLeay::CTX_set_mode ($self->{tls},
949 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1) 1151 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
950 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2)); 1152 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
951 1153
952 $self->{tls_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1154 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
953 $self->{tls_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1155 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
954 1156
955 Net::SSLeay::set_bio ($ssl, $self->{tls_rbio}, $self->{tls_wbio}); 1157 Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio});
956 1158
957 $self->{filter_w} = sub { 1159 $self->{filter_w} = sub {
958 $_[0]{tls_wbuf} .= ${$_[1]}; 1160 $_[0]{_tls_wbuf} .= ${$_[1]};
959 &_dotls; 1161 &_dotls;
960 }; 1162 };
961 $self->{filter_r} = sub { 1163 $self->{filter_r} = sub {
962 Net::SSLeay::BIO_write ($_[0]{tls_rbio}, ${$_[1]}); 1164 Net::SSLeay::BIO_write ($_[0]{_rbio}, ${$_[1]});
963 &_dotls; 1165 &_dotls;
964 }; 1166 };
965} 1167}
966 1168
967=item $handle->stoptls 1169=item $handle->stoptls
973 1175
974sub stoptls { 1176sub stoptls {
975 my ($self) = @_; 1177 my ($self) = @_;
976 1178
977 Net::SSLeay::free (delete $self->{tls}) if $self->{tls}; 1179 Net::SSLeay::free (delete $self->{tls}) if $self->{tls};
1180
978 delete $self->{tls_rbio}; 1181 delete $self->{_rbio};
979 delete $self->{tls_wbio}; 1182 delete $self->{_wbio};
980 delete $self->{tls_wbuf}; 1183 delete $self->{_tls_wbuf};
981 delete $self->{filter_r}; 1184 delete $self->{filter_r};
982 delete $self->{filter_w}; 1185 delete $self->{filter_w};
983} 1186}
984 1187
985sub DESTROY { 1188sub DESTROY {
1023 } 1226 }
1024} 1227}
1025 1228
1026=back 1229=back
1027 1230
1231=head1 SUBCLASSING AnyEvent::Handle
1232
1233In many cases, you might want to subclass AnyEvent::Handle.
1234
1235To make this easier, a given version of AnyEvent::Handle uses these
1236conventions:
1237
1238=over 4
1239
1240=item * all constructor arguments become object members.
1241
1242At least initially, when you pass a C<tls>-argument to the constructor it
1243will end up in C<< $handle->{tls} >>. Those members might be changes or
1244mutated later on (for example C<tls> will hold the TLS connection object).
1245
1246=item * other object member names are prefixed with an C<_>.
1247
1248All object members not explicitly documented (internal use) are prefixed
1249with an underscore character, so the remaining non-C<_>-namespace is free
1250for use for subclasses.
1251
1252=item * all members not documented here and not prefixed with an underscore
1253are free to use in subclasses.
1254
1255Of course, new versions of AnyEvent::Handle may introduce more "public"
1256member variables, but thats just life, at least it is documented.
1257
1258=back
1259
1028=head1 AUTHOR 1260=head1 AUTHOR
1029 1261
1030Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>. 1262Robin Redeker C<< <elmex at ta-sa.org> >>, Marc Lehmann <schmorp@schmorp.de>.
1031 1263
1032=cut 1264=cut

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines