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.44 by root, Thu May 29 00:00:07 2008 UTC vs.
Revision 1.85 by root, Thu Aug 21 19:53:19 2008 UTC

1package AnyEvent::Handle; 1package AnyEvent::Handle;
2 2
3no warnings; 3no warnings;
4use strict; 4use strict qw(subs vars);
5 5
6use AnyEvent (); 6use AnyEvent ();
7use AnyEvent::Util qw(WSAEWOULDBLOCK); 7use AnyEvent::Util qw(WSAEWOULDBLOCK);
8use Scalar::Util (); 8use Scalar::Util ();
9use Carp (); 9use Carp ();
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
17=cut 17=cut
18 18
19our $VERSION = '0.04'; 19our $VERSION = 4.232;
20 20
21=head1 SYNOPSIS 21=head1 SYNOPSIS
22 22
23 use AnyEvent; 23 use AnyEvent;
24 use AnyEvent::Handle; 24 use AnyEvent::Handle;
49 49
50This 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
51filehandles. For utility functions for doing non-blocking connects and accepts 51filehandles. For utility functions for doing non-blocking connects and accepts
52on sockets see L<AnyEvent::Util>. 52on sockets see L<AnyEvent::Util>.
53 53
54The L<AnyEvent::Intro> tutorial contains some well-documented
55AnyEvent::Handle examples.
56
54In the following, when the documentation refers to of "bytes" then this 57In the following, when the documentation refers to of "bytes" then this
55means characters. As sysread and syswrite are used for all I/O, their 58means characters. As sysread and syswrite are used for all I/O, their
56treatment of characters applies to this module as well. 59treatment of characters applies to this module as well.
57 60
58All callbacks will be invoked with the handle object as their first 61All callbacks will be invoked with the handle object as their first
70 73
71=item fh => $filehandle [MANDATORY] 74=item fh => $filehandle [MANDATORY]
72 75
73The filehandle this L<AnyEvent::Handle> object will operate on. 76The filehandle this L<AnyEvent::Handle> object will operate on.
74 77
75NOTE: The filehandle will be set to non-blocking (using 78NOTE: The filehandle will be set to non-blocking mode (using
76AnyEvent::Util::fh_nonblocking). 79C<AnyEvent::Util::fh_nonblocking>) by the constructor and needs to stay in
80that mode.
77 81
78=item on_eof => $cb->($handle) 82=item on_eof => $cb->($handle)
79 83
80Set the callback to be called on EOF. 84Set the callback to be called when an end-of-file condition is detected,
85i.e. in the case of a socket, when the other side has closed the
86connection cleanly.
81 87
88For sockets, this just means that the other side has stopped sending data,
89you can still try to write data, and, in fact, one can return from the eof
90callback and continue writing data, as only the read part has been shut
91down.
92
82While not mandatory, it is highly recommended to set an eof callback, 93While not mandatory, it is I<highly> recommended to set an eof callback,
83otherwise you might end up with a closed socket while you are still 94otherwise you might end up with a closed socket while you are still
84waiting for data. 95waiting for data.
85 96
97If an EOF condition has been detected but no C<on_eof> callback has been
98set, then a fatal error will be raised with C<$!> set to <0>.
99
86=item on_error => $cb->($handle) 100=item on_error => $cb->($handle, $fatal)
87 101
88This is the fatal error callback, that is called when, well, a fatal error 102This is the error callback, which is called when, well, some error
89occurs, such as not being able to resolve the hostname, failure to connect 103occured, such as not being able to resolve the hostname, failure to
90or a read error. 104connect or a read error.
91 105
92The object will not be in a usable state when this callback has been 106Some errors are fatal (which is indicated by C<$fatal> being true). On
93called. 107fatal errors the handle object will be shut down and will not be usable
108(but you are free to look at the current C< ->rbuf >). Examples of fatal
109errors are an EOF condition with active (but unsatisifable) read watchers
110(C<EPIPE>) or I/O errors.
111
112Non-fatal errors can be retried by simply returning, but it is recommended
113to simply ignore this parameter and instead abondon the handle object
114when this callback is invoked. Examples of non-fatal errors are timeouts
115C<ETIMEDOUT>) or badly-formatted data (C<EBADMSG>).
94 116
95On callback entrance, the value of C<$!> contains the operating system 117On callback entrance, the value of C<$!> contains the operating system
96error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>). 118error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>).
97 119
98The callback should throw an exception. If it returns, then
99AnyEvent::Handle will C<croak> for you.
100
101While not mandatory, it is I<highly> recommended to set this callback, as 120While not mandatory, it is I<highly> recommended to set this callback, as
102you will not be notified of errors otherwise. The default simply calls 121you will not be notified of errors otherwise. The default simply calls
103die. 122C<croak>.
104 123
105=item on_read => $cb->($handle) 124=item on_read => $cb->($handle)
106 125
107This sets the default read callback, which is called when data arrives 126This sets the default read callback, which is called when data arrives
108and no read request is in the queue. 127and no read request is in the queue (unlike read queue callbacks, this
128callback will only be called when at least one octet of data is in the
129read buffer).
109 130
110To access (and remove data from) the read buffer, use the C<< ->rbuf >> 131To access (and remove data from) the read buffer, use the C<< ->rbuf >>
111method or access the C<$handle->{rbuf}> member directly. 132method or access the C<$handle->{rbuf}> member directly.
112 133
113When an EOF condition is detected then AnyEvent::Handle will first try to 134When an EOF condition is detected then AnyEvent::Handle will first try to
120This sets the callback that is called when the write buffer becomes empty 141This 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). 142(or when the callback is set and the buffer is empty already).
122 143
123To append to the write buffer, use the C<< ->push_write >> method. 144To append to the write buffer, use the C<< ->push_write >> method.
124 145
146This callback is useful when you don't want to put all of your write data
147into the queue at once, for example, when you want to write the contents
148of some file to the socket you might not want to read the whole file into
149memory and push it into the queue, but instead only read more data from
150the file when the write queue becomes empty.
151
125=item timeout => $fractional_seconds 152=item timeout => $fractional_seconds
126 153
127If non-zero, then this enables an "inactivity" timeout: whenever this many 154If non-zero, then this enables an "inactivity" timeout: whenever this many
128seconds pass without a successful read or write on the underlying file 155seconds 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 156handle, the C<on_timeout> callback will be invoked (and if that one is
130missing, an C<ETIMEDOUT> errror will be raised). 157missing, an C<ETIMEDOUT> error will be raised).
131 158
132Note that timeout processing is also active when you currently do not have 159Note 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 160any outstanding read or write requests: If you plan to keep the connection
134idle then you should disable the timout temporarily or ignore the timeout 161idle then you should disable the timout temporarily or ignore the timeout
135in the C<on_timeout> callback. 162in the C<on_timeout> callback.
152be configured to accept only so-and-so much data that it cannot act on 179be configured to accept only so-and-so much data that it cannot act on
153(for example, when expecting a line, an attacker could send an unlimited 180(for example, when expecting a line, an attacker could send an unlimited
154amount of data without a callback ever being called as long as the line 181amount of data without a callback ever being called as long as the line
155isn't finished). 182isn't finished).
156 183
184=item autocork => <boolean>
185
186When disabled (the default), then C<push_write> will try to immediately
187write the data to the handle if possible. This avoids having to register
188a write watcher and wait for the next event loop iteration, but can be
189inefficient if you write multiple small chunks (this disadvantage is
190usually avoided by your kernel's nagle algorithm, see C<low_delay>).
191
192When enabled, then writes will always be queued till the next event loop
193iteration. This is efficient when you do many small writes per iteration,
194but less efficient when you do a single write only.
195
196=item no_delay => <boolean>
197
198When doing small writes on sockets, your operating system kernel might
199wait a bit for more data before actually sending it out. This is called
200the Nagle algorithm, and usually it is beneficial.
201
202In some situations you want as low a delay as possible, which cna be
203accomplishd by setting this option to true.
204
205The default is your opertaing system's default behaviour, this option
206explicitly enables or disables it, if possible.
207
157=item read_size => <bytes> 208=item read_size => <bytes>
158 209
159The default read block size (the amount of bytes this module will try to read 210The default read block size (the amount of bytes this module will try to read
160on each [loop iteration). Default: C<4096>. 211during each (loop iteration). Default: C<8192>.
161 212
162=item low_water_mark => <bytes> 213=item low_water_mark => <bytes>
163 214
164Sets the amount of bytes (default: C<0>) that make up an "empty" write 215Sets the amount of bytes (default: C<0>) that make up an "empty" write
165buffer: If the write reaches this size or gets even samller it is 216buffer: If the write reaches this size or gets even samller it is
166considered empty. 217considered empty.
167 218
219=item linger => <seconds>
220
221If non-zero (default: C<3600>), then the destructor of the
222AnyEvent::Handle object will check wether there is still outstanding write
223data and will install a watcher that will write out this data. No errors
224will be reported (this mostly matches how the operating system treats
225outstanding data at socket close time).
226
227This will not work for partial TLS data that could not yet been
228encoded. This data will be lost.
229
168=item tls => "accept" | "connect" | Net::SSLeay::SSL object 230=item tls => "accept" | "connect" | Net::SSLeay::SSL object
169 231
170When this parameter is given, it enables TLS (SSL) mode, that means it 232When this parameter is given, it enables TLS (SSL) mode, that means
171will start making tls handshake and will transparently encrypt/decrypt 233AnyEvent will start a TLS handshake and will transparently encrypt/decrypt
172data. 234data.
173 235
174TLS mode requires Net::SSLeay to be installed (it will be loaded 236TLS mode requires Net::SSLeay to be installed (it will be loaded
175automatically when you try to create a TLS handle). 237automatically when you try to create a TLS handle).
176 238
177For the TLS server side, use C<accept>, and for the TLS client side of a 239Unlike TCP, TLS has a server and client side: for the TLS server side, use
178connection, use C<connect> mode. 240C<accept>, and for the TLS client side of a connection, use C<connect>
241mode.
179 242
180You can also provide your own TLS connection object, but you have 243You can also provide your own TLS connection object, but you have
181to make sure that you call either C<Net::SSLeay::set_connect_state> 244to make sure that you call either C<Net::SSLeay::set_connect_state>
182or C<Net::SSLeay::set_accept_state> on it before you pass it to 245or C<Net::SSLeay::set_accept_state> on it before you pass it to
183AnyEvent::Handle. 246AnyEvent::Handle.
184 247
185See the C<starttls> method if you need to start TLs negotiation later. 248See the C<starttls> method for when need to start TLS negotiation later.
186 249
187=item tls_ctx => $ssl_ctx 250=item tls_ctx => $ssl_ctx
188 251
189Use the given Net::SSLeay::CTX object to create the new TLS connection 252Use the given Net::SSLeay::CTX object to create the new TLS connection
190(unless a connection object was specified directly). If this parameter is 253(unless a connection object was specified directly). If this parameter is
222 if ($self->{tls}) { 285 if ($self->{tls}) {
223 require Net::SSLeay; 286 require Net::SSLeay;
224 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 287 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
225 } 288 }
226 289
227# $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; # nop
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
230 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
231
232 $self->{_activity} = AnyEvent->now; 290 $self->{_activity} = AnyEvent->now;
233 $self->_timeout; 291 $self->_timeout;
234 292
293 $self->on_drain (delete $self->{on_drain}) if exists $self->{on_drain};
294 $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay};
295
235 $self->start_read; 296 $self->start_read
297 if $self->{on_read};
236 298
237 $self 299 $self
238} 300}
239 301
240sub _shutdown { 302sub _shutdown {
241 my ($self) = @_; 303 my ($self) = @_;
242 304
305 delete $self->{_tw};
243 delete $self->{_rw}; 306 delete $self->{_rw};
244 delete $self->{_ww}; 307 delete $self->{_ww};
245 delete $self->{fh}; 308 delete $self->{fh};
246}
247 309
310 $self->stoptls;
311
312 delete $self->{on_read};
313 delete $self->{_queue};
314}
315
248sub error { 316sub _error {
249 my ($self) = @_; 317 my ($self, $errno, $fatal) = @_;
250 318
251 {
252 local $!;
253 $self->_shutdown; 319 $self->_shutdown
254 } 320 if $fatal;
255 321
256 $self->{on_error}($self) 322 $! = $errno;
323
257 if $self->{on_error}; 324 if ($self->{on_error}) {
258 325 $self->{on_error}($self, $fatal);
326 } else {
259 Carp::croak "AnyEvent::Handle uncaught fatal error: $!"; 327 Carp::croak "AnyEvent::Handle uncaught error: $!";
328 }
260} 329}
261 330
262=item $fh = $handle->fh 331=item $fh = $handle->fh
263 332
264This method returns the file handle of the L<AnyEvent::Handle> object. 333This method returns the file handle of the L<AnyEvent::Handle> object.
295 364
296=cut 365=cut
297 366
298sub on_timeout { 367sub on_timeout {
299 $_[0]{on_timeout} = $_[1]; 368 $_[0]{on_timeout} = $_[1];
369}
370
371=item $handle->autocork ($boolean)
372
373Enables or disables the current autocork behaviour (see C<autocork>
374constructor argument).
375
376=cut
377
378=item $handle->no_delay ($boolean)
379
380Enables or disables the C<no_delay> setting (see constructor argument of
381the same name for details).
382
383=cut
384
385sub no_delay {
386 $_[0]{no_delay} = $_[1];
387
388 eval {
389 local $SIG{__DIE__};
390 setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1];
391 };
300} 392}
301 393
302############################################################################# 394#############################################################################
303 395
304=item $handle->timeout ($seconds) 396=item $handle->timeout ($seconds)
328 # now or in the past already? 420 # now or in the past already?
329 if ($after <= 0) { 421 if ($after <= 0) {
330 $self->{_activity} = $NOW; 422 $self->{_activity} = $NOW;
331 423
332 if ($self->{on_timeout}) { 424 if ($self->{on_timeout}) {
333 $self->{on_timeout}->($self); 425 $self->{on_timeout}($self);
334 } else { 426 } else {
335 $! = Errno::ETIMEDOUT; 427 $self->_error (&Errno::ETIMEDOUT);
336 $self->error;
337 } 428 }
338 429
339 # callbakx could have changed timeout value, optimise 430 # callback could have changed timeout value, optimise
340 return unless $self->{timeout}; 431 return unless $self->{timeout};
341 432
342 # calculate new after 433 # calculate new after
343 $after = $self->{timeout}; 434 $after = $self->{timeout};
344 } 435 }
345 436
346 Scalar::Util::weaken $self; 437 Scalar::Util::weaken $self;
438 return unless $self; # ->error could have destroyed $self
347 439
348 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub { 440 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
349 delete $self->{_tw}; 441 delete $self->{_tw};
350 $self->_timeout; 442 $self->_timeout;
351 }); 443 });
414 if $self->{low_water_mark} >= length $self->{wbuf} 506 if $self->{low_water_mark} >= length $self->{wbuf}
415 && $self->{on_drain}; 507 && $self->{on_drain};
416 508
417 delete $self->{_ww} unless length $self->{wbuf}; 509 delete $self->{_ww} unless length $self->{wbuf};
418 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 510 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
419 $self->error; 511 $self->_error ($!, 1);
420 } 512 }
421 }; 513 };
422 514
423 # try to write data immediately 515 # try to write data immediately
424 $cb->(); 516 $cb->() unless $self->{autocork};
425 517
426 # if still data left in wbuf, we need to poll 518 # if still data left in wbuf, we need to poll
427 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 519 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
428 if length $self->{wbuf}; 520 if length $self->{wbuf};
429 }; 521 };
444 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 536 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
445 ->($self, @_); 537 ->($self, @_);
446 } 538 }
447 539
448 if ($self->{filter_w}) { 540 if ($self->{filter_w}) {
449 $self->{filter_w}->($self, \$_[0]); 541 $self->{filter_w}($self, \$_[0]);
450 } else { 542 } else {
451 $self->{wbuf} .= $_[0]; 543 $self->{wbuf} .= $_[0];
452 $self->_drain_wbuf; 544 $self->_drain_wbuf;
453 } 545 }
454} 546}
455 547
456=item $handle->push_write (type => @args) 548=item $handle->push_write (type => @args)
457 549
458=item $handle->unshift_write (type => @args)
459
460Instead of formatting your data yourself, you can also let this module do 550Instead of formatting your data yourself, you can also let this module do
461the job by specifying a type and type-specific arguments. 551the job by specifying a type and type-specific arguments.
462 552
463Predefined types are (if you have ideas for additional types, feel free to 553Predefined types are (if you have ideas for additional types, feel free to
464drop by and tell us): 554drop by and tell us):
468=item netstring => $string 558=item netstring => $string
469 559
470Formats the given value as netstring 560Formats the given value as netstring
471(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them). 561(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
472 562
473=back
474
475=cut 563=cut
476 564
477register_write_type netstring => sub { 565register_write_type netstring => sub {
478 my ($self, $string) = @_; 566 my ($self, $string) = @_;
479 567
480 sprintf "%d:%s,", (length $string), $string 568 sprintf "%d:%s,", (length $string), $string
569};
570
571=item packstring => $format, $data
572
573An octet string prefixed with an encoded length. The encoding C<$format>
574uses the same format as a Perl C<pack> format, but must specify a single
575integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
576optional C<!>, C<< < >> or C<< > >> modifier).
577
578=cut
579
580register_write_type packstring => sub {
581 my ($self, $format, $string) = @_;
582
583 pack "$format/a*", $string
481}; 584};
482 585
483=item json => $array_or_hashref 586=item json => $array_or_hashref
484 587
485Encodes the given hash or array reference into a JSON object. Unless you 588Encodes the given hash or array reference into a JSON object. Unless you
519 622
520 $self->{json} ? $self->{json}->encode ($ref) 623 $self->{json} ? $self->{json}->encode ($ref)
521 : JSON::encode_json ($ref) 624 : JSON::encode_json ($ref)
522}; 625};
523 626
627=item storable => $reference
628
629Freezes the given reference using L<Storable> and writes it to the
630handle. Uses the C<nfreeze> format.
631
632=cut
633
634register_write_type storable => sub {
635 my ($self, $ref) = @_;
636
637 require Storable;
638
639 pack "w/a*", Storable::nfreeze ($ref)
640};
641
642=back
643
524=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 644=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
525 645
526This function (not method) lets you add your own types to C<push_write>. 646This function (not method) lets you add your own types to C<push_write>.
527Whenever the given C<type> is used, C<push_write> will invoke the code 647Whenever the given C<type> is used, C<push_write> will invoke the code
528reference with the handle object and the remaining arguments. 648reference with the handle object and the remaining arguments.
548ways, the "simple" way, using only C<on_read> and the "complex" way, using 668ways, the "simple" way, using only C<on_read> and the "complex" way, using
549a queue. 669a queue.
550 670
551In the simple case, you just install an C<on_read> callback and whenever 671In the simple case, you just install an C<on_read> callback and whenever
552new data arrives, it will be called. You can then remove some data (if 672new data arrives, it will be called. You can then remove some data (if
553enough is there) from the read buffer (C<< $handle->rbuf >>) if you want 673enough is there) from the read buffer (C<< $handle->rbuf >>). Or you cna
554or not. 674leave the data there if you want to accumulate more (e.g. when only a
675partial message has been received so far).
555 676
556In the more complex case, you want to queue multiple callbacks. In this 677In the more complex case, you want to queue multiple callbacks. In this
557case, AnyEvent::Handle will call the first queued callback each time new 678case, AnyEvent::Handle will call the first queued callback each time new
558data arrives and removes it when it has done its job (see C<push_read>, 679data arrives (also the first time it is queued) and removes it when it has
559below). 680done its job (see C<push_read>, below).
560 681
561This way you can, for example, push three line-reads, followed by reading 682This way you can, for example, push three line-reads, followed by reading
562a chunk of data, and AnyEvent::Handle will execute them in order. 683a chunk of data, and AnyEvent::Handle will execute them in order.
563 684
564Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by 685Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
565the specified number of bytes which give an XML datagram. 686the specified number of bytes which give an XML datagram.
566 687
567 # in the default state, expect some header bytes 688 # in the default state, expect some header bytes
568 $handle->on_read (sub { 689 $handle->on_read (sub {
569 # some data is here, now queue the length-header-read (4 octets) 690 # some data is here, now queue the length-header-read (4 octets)
570 shift->unshift_read_chunk (4, sub { 691 shift->unshift_read (chunk => 4, sub {
571 # header arrived, decode 692 # header arrived, decode
572 my $len = unpack "N", $_[1]; 693 my $len = unpack "N", $_[1];
573 694
574 # now read the payload 695 # now read the payload
575 shift->unshift_read_chunk ($len, sub { 696 shift->unshift_read (chunk => $len, sub {
576 my $xml = $_[1]; 697 my $xml = $_[1];
577 # handle xml 698 # handle xml
578 }); 699 });
579 }); 700 });
580 }); 701 });
581 702
582Example 2: Implement a client for a protocol that replies either with 703Example 2: Implement a client for a protocol that replies either with "OK"
583"OK" and another line or "ERROR" for one request, and 64 bytes for the 704and another line or "ERROR" for the first request that is sent, and 64
584second request. Due tot he availability of a full queue, we can just 705bytes for the second request. Due to the availability of a queue, we can
585pipeline sending both requests and manipulate the queue as necessary in 706just pipeline sending both requests and manipulate the queue as necessary
586the callbacks: 707in the callbacks.
587 708
588 # request one 709When the first callback is called and sees an "OK" response, it will
710C<unshift> another line-read. This line-read will be queued I<before> the
71164-byte chunk callback.
712
713 # request one, returns either "OK + extra line" or "ERROR"
589 $handle->push_write ("request 1\015\012"); 714 $handle->push_write ("request 1\015\012");
590 715
591 # we expect "ERROR" or "OK" as response, so push a line read 716 # we expect "ERROR" or "OK" as response, so push a line read
592 $handle->push_read_line (sub { 717 $handle->push_read (line => sub {
593 # if we got an "OK", we have to _prepend_ another line, 718 # if we got an "OK", we have to _prepend_ another line,
594 # so it will be read before the second request reads its 64 bytes 719 # so it will be read before the second request reads its 64 bytes
595 # which are already in the queue when this callback is called 720 # which are already in the queue when this callback is called
596 # we don't do this in case we got an error 721 # we don't do this in case we got an error
597 if ($_[1] eq "OK") { 722 if ($_[1] eq "OK") {
598 $_[0]->unshift_read_line (sub { 723 $_[0]->unshift_read (line => sub {
599 my $response = $_[1]; 724 my $response = $_[1];
600 ... 725 ...
601 }); 726 });
602 } 727 }
603 }); 728 });
604 729
605 # request two 730 # request two, simply returns 64 octets
606 $handle->push_write ("request 2\015\012"); 731 $handle->push_write ("request 2\015\012");
607 732
608 # simply read 64 bytes, always 733 # simply read 64 bytes, always
609 $handle->push_read_chunk (64, sub { 734 $handle->push_read (chunk => 64, sub {
610 my $response = $_[1]; 735 my $response = $_[1];
611 ... 736 ...
612 }); 737 });
613 738
614=over 4 739=over 4
615 740
616=cut 741=cut
617 742
618sub _drain_rbuf { 743sub _drain_rbuf {
619 my ($self) = @_; 744 my ($self) = @_;
745
746 local $self->{_in_drain} = 1;
620 747
621 if ( 748 if (
622 defined $self->{rbuf_max} 749 defined $self->{rbuf_max}
623 && $self->{rbuf_max} < length $self->{rbuf} 750 && $self->{rbuf_max} < length $self->{rbuf}
624 ) { 751 ) {
625 $! = &Errno::ENOSPC; 752 $self->_error (&Errno::ENOSPC, 1), return;
626 $self->error;
627 } 753 }
628 754
629 return if $self->{in_drain}; 755 while () {
630 local $self->{in_drain} = 1;
631
632 while (my $len = length $self->{rbuf}) { 756 my $len = length $self->{rbuf};
633 no strict 'refs'; 757
634 if (my $cb = shift @{ $self->{_queue} }) { 758 if (my $cb = shift @{ $self->{_queue} }) {
635 unless ($cb->($self)) { 759 unless ($cb->($self)) {
636 if ($self->{_eof}) { 760 if ($self->{_eof}) {
637 # no progress can be made (not enough data and no data forthcoming) 761 # no progress can be made (not enough data and no data forthcoming)
638 $! = &Errno::EPIPE; 762 $self->_error (&Errno::EPIPE, 1), return;
639 $self->error;
640 } 763 }
641 764
642 unshift @{ $self->{_queue} }, $cb; 765 unshift @{ $self->{_queue} }, $cb;
643 return; 766 last;
644 } 767 }
645 } elsif ($self->{on_read}) { 768 } elsif ($self->{on_read}) {
769 last unless $len;
770
646 $self->{on_read}($self); 771 $self->{on_read}($self);
647 772
648 if ( 773 if (
649 $self->{_eof} # if no further data will arrive
650 && $len == length $self->{rbuf} # and no data has been consumed 774 $len == length $self->{rbuf} # if no data has been consumed
651 && !@{ $self->{_queue} } # and the queue is still empty 775 && !@{ $self->{_queue} } # and the queue is still empty
652 && $self->{on_read} # and we still want to read data 776 && $self->{on_read} # but we still have on_read
653 ) { 777 ) {
778 # no further data will arrive
654 # then no progress can be made 779 # so no progress can be made
655 $! = &Errno::EPIPE; 780 $self->_error (&Errno::EPIPE, 1), return
656 $self->error; 781 if $self->{_eof};
782
783 last; # more data might arrive
657 } 784 }
658 } else { 785 } else {
659 # read side becomes idle 786 # read side becomes idle
660 delete $self->{_rw}; 787 delete $self->{_rw};
661 return; 788 last;
662 } 789 }
663 } 790 }
664 791
665 if ($self->{_eof}) { 792 if ($self->{_eof}) {
666 $self->_shutdown; 793 if ($self->{on_eof}) {
667 $self->{on_eof}($self) 794 $self->{on_eof}($self)
668 if $self->{on_eof}; 795 } else {
796 $self->_error (0, 1);
797 }
798 }
799
800 # may need to restart read watcher
801 unless ($self->{_rw}) {
802 $self->start_read
803 if $self->{on_read} || @{ $self->{_queue} };
669 } 804 }
670} 805}
671 806
672=item $handle->on_read ($cb) 807=item $handle->on_read ($cb)
673 808
679 814
680sub on_read { 815sub on_read {
681 my ($self, $cb) = @_; 816 my ($self, $cb) = @_;
682 817
683 $self->{on_read} = $cb; 818 $self->{on_read} = $cb;
819 $self->_drain_rbuf if $cb && !$self->{_in_drain};
684} 820}
685 821
686=item $handle->rbuf 822=item $handle->rbuf
687 823
688Returns the read buffer (as a modifiable lvalue). 824Returns the read buffer (as a modifiable lvalue).
737 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 873 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
738 ->($self, $cb, @_); 874 ->($self, $cb, @_);
739 } 875 }
740 876
741 push @{ $self->{_queue} }, $cb; 877 push @{ $self->{_queue} }, $cb;
742 $self->_drain_rbuf; 878 $self->_drain_rbuf unless $self->{_in_drain};
743} 879}
744 880
745sub unshift_read { 881sub unshift_read {
746 my $self = shift; 882 my $self = shift;
747 my $cb = pop; 883 my $cb = pop;
753 ->($self, $cb, @_); 889 ->($self, $cb, @_);
754 } 890 }
755 891
756 892
757 unshift @{ $self->{_queue} }, $cb; 893 unshift @{ $self->{_queue} }, $cb;
758 $self->_drain_rbuf; 894 $self->_drain_rbuf unless $self->{_in_drain};
759} 895}
760 896
761=item $handle->push_read (type => @args, $cb) 897=item $handle->push_read (type => @args, $cb)
762 898
763=item $handle->unshift_read (type => @args, $cb) 899=item $handle->unshift_read (type => @args, $cb)
793 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, ""); 929 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
794 1 930 1
795 } 931 }
796}; 932};
797 933
798# compatibility with older API
799sub push_read_chunk {
800 $_[0]->push_read (chunk => $_[1], $_[2]);
801}
802
803sub unshift_read_chunk {
804 $_[0]->unshift_read (chunk => $_[1], $_[2]);
805}
806
807=item line => [$eol, ]$cb->($handle, $line, $eol) 934=item line => [$eol, ]$cb->($handle, $line, $eol)
808 935
809The callback will be called only once a full line (including the end of 936The callback will be called only once a full line (including the end of
810line marker, C<$eol>) has been read. This line (excluding the end of line 937line marker, C<$eol>) has been read. This line (excluding the end of line
811marker) will be passed to the callback as second argument (C<$line>), and 938marker) will be passed to the callback as second argument (C<$line>), and
826=cut 953=cut
827 954
828register_read_type line => sub { 955register_read_type line => sub {
829 my ($self, $cb, $eol) = @_; 956 my ($self, $cb, $eol) = @_;
830 957
831 $eol = qr|(\015?\012)| if @_ < 3; 958 if (@_ < 3) {
832 $eol = quotemeta $eol unless ref $eol; 959 # this is more than twice as fast as the generic code below
833 $eol = qr|^(.*?)($eol)|s;
834
835 sub { 960 sub {
836 $_[0]{rbuf} =~ s/$eol// or return; 961 $_[0]{rbuf} =~ s/^([^\015\012]*)(\015?\012)// or return;
837 962
838 $cb->($_[0], $1, $2); 963 $cb->($_[0], $1, $2);
839 1
840 }
841};
842
843# compatibility with older API
844sub push_read_line {
845 my $self = shift;
846 $self->push_read (line => @_);
847}
848
849sub unshift_read_line {
850 my $self = shift;
851 $self->unshift_read (line => @_);
852}
853
854=item netstring => $cb->($handle, $string)
855
856A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
857
858Throws an error with C<$!> set to EBADMSG on format violations.
859
860=cut
861
862register_read_type netstring => sub {
863 my ($self, $cb) = @_;
864
865 sub {
866 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
867 if ($_[0]{rbuf} =~ /[^0-9]/) {
868 $! = &Errno::EBADMSG;
869 $self->error;
870 } 964 1
871 return;
872 } 965 }
966 } else {
967 $eol = quotemeta $eol unless ref $eol;
968 $eol = qr|^(.*?)($eol)|s;
873 969
874 my $len = $1; 970 sub {
971 $_[0]{rbuf} =~ s/$eol// or return;
875 972
876 $self->unshift_read (chunk => $len, sub { 973 $cb->($_[0], $1, $2);
877 my $string = $_[1];
878 $_[0]->unshift_read (chunk => 1, sub {
879 if ($_[1] eq ",") {
880 $cb->($_[0], $string);
881 } else {
882 $! = &Errno::EBADMSG;
883 $self->error;
884 }
885 }); 974 1
886 }); 975 }
887
888 1
889 } 976 }
890}; 977};
891 978
892=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 979=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
893 980
945 return 1; 1032 return 1;
946 } 1033 }
947 1034
948 # reject 1035 # reject
949 if ($reject && $$rbuf =~ $reject) { 1036 if ($reject && $$rbuf =~ $reject) {
950 $! = &Errno::EBADMSG; 1037 $self->_error (&Errno::EBADMSG);
951 $self->error;
952 } 1038 }
953 1039
954 # skip 1040 # skip
955 if ($skip && $$rbuf =~ $skip) { 1041 if ($skip && $$rbuf =~ $skip) {
956 $data .= substr $$rbuf, 0, $+[0], ""; 1042 $data .= substr $$rbuf, 0, $+[0], "";
958 1044
959 () 1045 ()
960 } 1046 }
961}; 1047};
962 1048
1049=item netstring => $cb->($handle, $string)
1050
1051A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
1052
1053Throws an error with C<$!> set to EBADMSG on format violations.
1054
1055=cut
1056
1057register_read_type netstring => sub {
1058 my ($self, $cb) = @_;
1059
1060 sub {
1061 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
1062 if ($_[0]{rbuf} =~ /[^0-9]/) {
1063 $self->_error (&Errno::EBADMSG);
1064 }
1065 return;
1066 }
1067
1068 my $len = $1;
1069
1070 $self->unshift_read (chunk => $len, sub {
1071 my $string = $_[1];
1072 $_[0]->unshift_read (chunk => 1, sub {
1073 if ($_[1] eq ",") {
1074 $cb->($_[0], $string);
1075 } else {
1076 $self->_error (&Errno::EBADMSG);
1077 }
1078 });
1079 });
1080
1081 1
1082 }
1083};
1084
1085=item packstring => $format, $cb->($handle, $string)
1086
1087An octet string prefixed with an encoded length. The encoding C<$format>
1088uses the same format as a Perl C<pack> format, but must specify a single
1089integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
1090optional C<!>, C<< < >> or C<< > >> modifier).
1091
1092DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>.
1093
1094Example: read a block of data prefixed by its length in BER-encoded
1095format (very efficient).
1096
1097 $handle->push_read (packstring => "w", sub {
1098 my ($handle, $data) = @_;
1099 });
1100
1101=cut
1102
1103register_read_type packstring => sub {
1104 my ($self, $cb, $format) = @_;
1105
1106 sub {
1107 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1108 defined (my $len = eval { unpack $format, $_[0]{rbuf} })
1109 or return;
1110
1111 $format = length pack $format, $len;
1112
1113 # bypass unshift if we already have the remaining chunk
1114 if ($format + $len <= length $_[0]{rbuf}) {
1115 my $data = substr $_[0]{rbuf}, $format, $len;
1116 substr $_[0]{rbuf}, 0, $format + $len, "";
1117 $cb->($_[0], $data);
1118 } else {
1119 # remove prefix
1120 substr $_[0]{rbuf}, 0, $format, "";
1121
1122 # read remaining chunk
1123 $_[0]->unshift_read (chunk => $len, $cb);
1124 }
1125
1126 1
1127 }
1128};
1129
963=item json => $cb->($handle, $hash_or_arrayref) 1130=item json => $cb->($handle, $hash_or_arrayref)
964 1131
965Reads a JSON object or array, decodes it and passes it to the callback. 1132Reads a JSON object or array, decodes it and passes it to the callback.
966 1133
967If a C<json> object was passed to the constructor, then that will be used 1134If a C<json> object was passed to the constructor, then that will be used
977the C<json> write type description, above, for an actual example. 1144the C<json> write type description, above, for an actual example.
978 1145
979=cut 1146=cut
980 1147
981register_read_type json => sub { 1148register_read_type json => sub {
982 my ($self, $cb, $accept, $reject, $skip) = @_; 1149 my ($self, $cb) = @_;
983 1150
984 require JSON; 1151 require JSON;
985 1152
986 my $data; 1153 my $data;
987 my $rbuf = \$self->{rbuf}; 1154 my $rbuf = \$self->{rbuf};
1002 () 1169 ()
1003 } 1170 }
1004 } 1171 }
1005}; 1172};
1006 1173
1174=item storable => $cb->($handle, $ref)
1175
1176Deserialises a L<Storable> frozen representation as written by the
1177C<storable> write type (BER-encoded length prefix followed by nfreeze'd
1178data).
1179
1180Raises C<EBADMSG> error if the data could not be decoded.
1181
1182=cut
1183
1184register_read_type storable => sub {
1185 my ($self, $cb) = @_;
1186
1187 require Storable;
1188
1189 sub {
1190 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1191 defined (my $len = eval { unpack "w", $_[0]{rbuf} })
1192 or return;
1193
1194 my $format = length pack "w", $len;
1195
1196 # bypass unshift if we already have the remaining chunk
1197 if ($format + $len <= length $_[0]{rbuf}) {
1198 my $data = substr $_[0]{rbuf}, $format, $len;
1199 substr $_[0]{rbuf}, 0, $format + $len, "";
1200 $cb->($_[0], Storable::thaw ($data));
1201 } else {
1202 # remove prefix
1203 substr $_[0]{rbuf}, 0, $format, "";
1204
1205 # read remaining chunk
1206 $_[0]->unshift_read (chunk => $len, sub {
1207 if (my $ref = eval { Storable::thaw ($_[1]) }) {
1208 $cb->($_[0], $ref);
1209 } else {
1210 $self->_error (&Errno::EBADMSG);
1211 }
1212 });
1213 }
1214
1215 1
1216 }
1217};
1218
1007=back 1219=back
1008 1220
1009=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args) 1221=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args)
1010 1222
1011This function (not method) lets you add your own types to C<push_read>. 1223This function (not method) lets you add your own types to C<push_read>.
1029=item $handle->stop_read 1241=item $handle->stop_read
1030 1242
1031=item $handle->start_read 1243=item $handle->start_read
1032 1244
1033In rare cases you actually do not want to read anything from the 1245In rare cases you actually do not want to read anything from the
1034socket. In this case you can call C<stop_read>. Neither C<on_read> no 1246socket. In this case you can call C<stop_read>. Neither C<on_read> nor
1035any queued callbacks will be executed then. To start reading again, call 1247any queued callbacks will be executed then. To start reading again, call
1036C<start_read>. 1248C<start_read>.
1249
1250Note that AnyEvent::Handle will automatically C<start_read> for you when
1251you change the C<on_read> callback or push/unshift a read callback, and it
1252will automatically C<stop_read> for you when neither C<on_read> is set nor
1253there are any read requests in the queue.
1037 1254
1038=cut 1255=cut
1039 1256
1040sub stop_read { 1257sub stop_read {
1041 my ($self) = @_; 1258 my ($self) = @_;
1055 1272
1056 if ($len > 0) { 1273 if ($len > 0) {
1057 $self->{_activity} = AnyEvent->now; 1274 $self->{_activity} = AnyEvent->now;
1058 1275
1059 $self->{filter_r} 1276 $self->{filter_r}
1060 ? $self->{filter_r}->($self, $rbuf) 1277 ? $self->{filter_r}($self, $rbuf)
1061 : $self->_drain_rbuf; 1278 : $self->{_in_drain} || $self->_drain_rbuf;
1062 1279
1063 } elsif (defined $len) { 1280 } elsif (defined $len) {
1064 delete $self->{_rw}; 1281 delete $self->{_rw};
1065 delete $self->{_ww};
1066 delete $self->{_tw};
1067 $self->{_eof} = 1; 1282 $self->{_eof} = 1;
1068 $self->_drain_rbuf; 1283 $self->_drain_rbuf unless $self->{_in_drain};
1069 1284
1070 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1285 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1071 return $self->error; 1286 return $self->_error ($!, 1);
1072 } 1287 }
1073 }); 1288 });
1074 } 1289 }
1075} 1290}
1076 1291
1077sub _dotls { 1292sub _dotls {
1078 my ($self) = @_; 1293 my ($self) = @_;
1294
1295 my $buf;
1079 1296
1080 if (length $self->{_tls_wbuf}) { 1297 if (length $self->{_tls_wbuf}) {
1081 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { 1298 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
1082 substr $self->{_tls_wbuf}, 0, $len, ""; 1299 substr $self->{_tls_wbuf}, 0, $len, "";
1083 } 1300 }
1084 } 1301 }
1085 1302
1086 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1303 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1087 $self->{wbuf} .= $buf; 1304 $self->{wbuf} .= $buf;
1088 $self->_drain_wbuf; 1305 $self->_drain_wbuf;
1089 } 1306 }
1090 1307
1091 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 1308 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1309 if (length $buf) {
1092 $self->{rbuf} .= $buf; 1310 $self->{rbuf} .= $buf;
1093 $self->_drain_rbuf; 1311 $self->_drain_rbuf unless $self->{_in_drain};
1312 } else {
1313 # let's treat SSL-eof as we treat normal EOF
1314 $self->{_eof} = 1;
1315 $self->_shutdown;
1316 return;
1317 }
1094 } 1318 }
1095 1319
1096 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1320 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
1097 1321
1098 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1322 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
1099 if ($err == Net::SSLeay::ERROR_SYSCALL ()) { 1323 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
1100 $self->error; 1324 return $self->_error ($!, 1);
1101 } elsif ($err == Net::SSLeay::ERROR_SSL ()) { 1325 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
1102 $! = &Errno::EIO; 1326 return $self->_error (&Errno::EIO, 1);
1103 $self->error;
1104 } 1327 }
1105 1328
1106 # all others are fine for our purposes 1329 # all others are fine for our purposes
1107 } 1330 }
1108} 1331}
1123call and can be used or changed to your liking. Note that the handshake 1346call and can be used or changed to your liking. Note that the handshake
1124might have already started when this function returns. 1347might have already started when this function returns.
1125 1348
1126=cut 1349=cut
1127 1350
1128# TODO: maybe document...
1129sub starttls { 1351sub starttls {
1130 my ($self, $ssl, $ctx) = @_; 1352 my ($self, $ssl, $ctx) = @_;
1131 1353
1132 $self->stoptls; 1354 $self->stoptls;
1133 1355
1186 1408
1187sub DESTROY { 1409sub DESTROY {
1188 my $self = shift; 1410 my $self = shift;
1189 1411
1190 $self->stoptls; 1412 $self->stoptls;
1413
1414 my $linger = exists $self->{linger} ? $self->{linger} : 3600;
1415
1416 if ($linger && length $self->{wbuf}) {
1417 my $fh = delete $self->{fh};
1418 my $wbuf = delete $self->{wbuf};
1419
1420 my @linger;
1421
1422 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub {
1423 my $len = syswrite $fh, $wbuf, length $wbuf;
1424
1425 if ($len > 0) {
1426 substr $wbuf, 0, $len, "";
1427 } else {
1428 @linger = (); # end
1429 }
1430 });
1431 push @linger, AnyEvent->timer (after => $linger, cb => sub {
1432 @linger = ();
1433 });
1434 }
1191} 1435}
1192 1436
1193=item AnyEvent::Handle::TLS_CTX 1437=item AnyEvent::Handle::TLS_CTX
1194 1438
1195This function creates and returns the Net::SSLeay::CTX object used by 1439This function creates and returns the Net::SSLeay::CTX object used by
1237=over 4 1481=over 4
1238 1482
1239=item * all constructor arguments become object members. 1483=item * all constructor arguments become object members.
1240 1484
1241At least initially, when you pass a C<tls>-argument to the constructor it 1485At least initially, when you pass a C<tls>-argument to the constructor it
1242will end up in C<< $handle->{tls} >>. Those members might be changes or 1486will end up in C<< $handle->{tls} >>. Those members might be changed or
1243mutated later on (for example C<tls> will hold the TLS connection object). 1487mutated later on (for example C<tls> will hold the TLS connection object).
1244 1488
1245=item * other object member names are prefixed with an C<_>. 1489=item * other object member names are prefixed with an C<_>.
1246 1490
1247All object members not explicitly documented (internal use) are prefixed 1491All object members not explicitly documented (internal use) are prefixed

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines