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.43 by root, Wed May 28 23:57:38 2008 UTC vs.
Revision 1.79 by root, Sun Jul 27 08:37:56 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 ();
10use Fcntl (); 10use Fcntl ();
11use Errno qw(EAGAIN EINTR); 11use Errno qw(EAGAIN EINTR);
12use Time::HiRes qw(time);
13 12
14=head1 NAME 13=head1 NAME
15 14
16AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent 15AnyEvent::Handle - non-blocking I/O on file handles via AnyEvent
17 16
18=cut 17=cut
19 18
20our $VERSION = '0.04'; 19our $VERSION = 4.22;
21 20
22=head1 SYNOPSIS 21=head1 SYNOPSIS
23 22
24 use AnyEvent; 23 use AnyEvent;
25 use AnyEvent::Handle; 24 use AnyEvent::Handle;
76NOTE: The filehandle will be set to non-blocking (using 75NOTE: The filehandle will be set to non-blocking (using
77AnyEvent::Util::fh_nonblocking). 76AnyEvent::Util::fh_nonblocking).
78 77
79=item on_eof => $cb->($handle) 78=item on_eof => $cb->($handle)
80 79
81Set the callback to be called on EOF. 80Set the callback to be called when an end-of-file condition is detected,
81i.e. in the case of a socket, when the other side has closed the
82connection cleanly.
82 83
83While not mandatory, it is highly recommended to set an eof callback, 84While not mandatory, it is highly recommended to set an eof callback,
84otherwise you might end up with a closed socket while you are still 85otherwise you might end up with a closed socket while you are still
85waiting for data. 86waiting for data.
86 87
87=item on_error => $cb->($handle) 88=item on_error => $cb->($handle, $fatal)
88 89
89This is the fatal error callback, that is called when, well, a fatal error 90This is the error callback, which is called when, well, some error
90occurs, such as not being able to resolve the hostname, failure to connect 91occured, such as not being able to resolve the hostname, failure to
91or a read error. 92connect or a read error.
92 93
93The object will not be in a usable state when this callback has been 94Some errors are fatal (which is indicated by C<$fatal> being true). On
94called. 95fatal errors the handle object will be shut down and will not be
96usable. Non-fatal errors can be retried by simply returning, but it is
97recommended to simply ignore this parameter and instead abondon the handle
98object when this callback is invoked.
95 99
96On callback entrance, the value of C<$!> contains the operating system 100On callback entrance, the value of C<$!> contains the operating system
97error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>). 101error (or C<ENOSPC>, C<EPIPE>, C<ETIMEDOUT> or C<EBADMSG>).
98 102
99The callback should throw an exception. If it returns, then
100AnyEvent::Handle will C<croak> for you.
101
102While not mandatory, it is I<highly> recommended to set this callback, as 103While not mandatory, it is I<highly> recommended to set this callback, as
103you will not be notified of errors otherwise. The default simply calls 104you will not be notified of errors otherwise. The default simply calls
104die. 105C<croak>.
105 106
106=item on_read => $cb->($handle) 107=item on_read => $cb->($handle)
107 108
108This sets the default read callback, which is called when data arrives 109This sets the default read callback, which is called when data arrives
109and no read request is in the queue. 110and no read request is in the queue (unlike read queue callbacks, this
111callback will only be called when at least one octet of data is in the
112read buffer).
110 113
111To access (and remove data from) the read buffer, use the C<< ->rbuf >> 114To access (and remove data from) the read buffer, use the C<< ->rbuf >>
112method or access the C<$handle->{rbuf}> member directly. 115method or access the C<$handle->{rbuf}> member directly.
113 116
114When an EOF condition is detected then AnyEvent::Handle will first try to 117When an EOF condition is detected then AnyEvent::Handle will first try to
121This sets the callback that is called when the write buffer becomes empty 124This sets the callback that is called when the write buffer becomes empty
122(or when the callback is set and the buffer is empty already). 125(or when the callback is set and the buffer is empty already).
123 126
124To append to the write buffer, use the C<< ->push_write >> method. 127To append to the write buffer, use the C<< ->push_write >> method.
125 128
129This callback is useful when you don't want to put all of your write data
130into the queue at once, for example, when you want to write the contents
131of some file to the socket you might not want to read the whole file into
132memory and push it into the queue, but instead only read more data from
133the file when the write queue becomes empty.
134
126=item timeout => $fractional_seconds 135=item timeout => $fractional_seconds
127 136
128If non-zero, then this enables an "inactivity" timeout: whenever this many 137If non-zero, then this enables an "inactivity" timeout: whenever this many
129seconds pass without a successful read or write on the underlying file 138seconds pass without a successful read or write on the underlying file
130handle, the C<on_timeout> callback will be invoked (and if that one is 139handle, the C<on_timeout> callback will be invoked (and if that one is
131missing, an C<ETIMEDOUT> errror will be raised). 140missing, an C<ETIMEDOUT> error will be raised).
132 141
133Note that timeout processing is also active when you currently do not have 142Note that timeout processing is also active when you currently do not have
134any outstanding read or write requests: If you plan to keep the connection 143any outstanding read or write requests: If you plan to keep the connection
135idle then you should disable the timout temporarily or ignore the timeout 144idle then you should disable the timout temporarily or ignore the timeout
136in the C<on_timeout> callback. 145in the C<on_timeout> callback.
153be configured to accept only so-and-so much data that it cannot act on 162be configured to accept only so-and-so much data that it cannot act on
154(for example, when expecting a line, an attacker could send an unlimited 163(for example, when expecting a line, an attacker could send an unlimited
155amount of data without a callback ever being called as long as the line 164amount of data without a callback ever being called as long as the line
156isn't finished). 165isn't finished).
157 166
167=item autocork => <boolean>
168
169When disabled (the default), then C<push_write> will try to immediately
170write the data to the handle if possible. This avoids having to register
171a write watcher and wait for the next event loop iteration, but can be
172inefficient if you write multiple small chunks (this disadvantage is
173usually avoided by your kernel's nagle algorithm, see C<low_delay>).
174
175When enabled, then writes will always be queued till the next event loop
176iteration. This is efficient when you do many small writes per iteration,
177but less efficient when you do a single write only.
178
179=item no_delay => <boolean>
180
181When doing small writes on sockets, your operating system kernel might
182wait a bit for more data before actually sending it out. This is called
183the Nagle algorithm, and usually it is beneficial.
184
185In some situations you want as low a delay as possible, which cna be
186accomplishd by setting this option to true.
187
188The default is your opertaing system's default behaviour, this option
189explicitly enables or disables it, if possible.
190
158=item read_size => <bytes> 191=item read_size => <bytes>
159 192
160The default read block size (the amount of bytes this module will try to read 193The default read block size (the amount of bytes this module will try to read
161on each [loop iteration). Default: C<4096>. 194during each (loop iteration). Default: C<8192>.
162 195
163=item low_water_mark => <bytes> 196=item low_water_mark => <bytes>
164 197
165Sets the amount of bytes (default: C<0>) that make up an "empty" write 198Sets the amount of bytes (default: C<0>) that make up an "empty" write
166buffer: If the write reaches this size or gets even samller it is 199buffer: If the write reaches this size or gets even samller it is
167considered empty. 200considered empty.
201
202=item linger => <seconds>
203
204If non-zero (default: C<3600>), then the destructor of the
205AnyEvent::Handle object will check wether there is still outstanding write
206data and will install a watcher that will write out this data. No errors
207will be reported (this mostly matches how the operating system treats
208outstanding data at socket close time).
209
210This will not work for partial TLS data that could not yet been
211encoded. This data will be lost.
168 212
169=item tls => "accept" | "connect" | Net::SSLeay::SSL object 213=item tls => "accept" | "connect" | Net::SSLeay::SSL object
170 214
171When this parameter is given, it enables TLS (SSL) mode, that means it 215When this parameter is given, it enables TLS (SSL) mode, that means it
172will start making tls handshake and will transparently encrypt/decrypt 216will start making tls handshake and will transparently encrypt/decrypt
181You can also provide your own TLS connection object, but you have 225You can also provide your own TLS connection object, but you have
182to make sure that you call either C<Net::SSLeay::set_connect_state> 226to make sure that you call either C<Net::SSLeay::set_connect_state>
183or C<Net::SSLeay::set_accept_state> on it before you pass it to 227or C<Net::SSLeay::set_accept_state> on it before you pass it to
184AnyEvent::Handle. 228AnyEvent::Handle.
185 229
186See the C<starttls> method if you need to start TLs negotiation later. 230See the C<starttls> method if you need to start TLS negotiation later.
187 231
188=item tls_ctx => $ssl_ctx 232=item tls_ctx => $ssl_ctx
189 233
190Use the given Net::SSLeay::CTX object to create the new TLS connection 234Use the given Net::SSLeay::CTX object to create the new TLS connection
191(unless a connection object was specified directly). If this parameter is 235(unless a connection object was specified directly). If this parameter is
223 if ($self->{tls}) { 267 if ($self->{tls}) {
224 require Net::SSLeay; 268 require Net::SSLeay;
225 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 269 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx});
226 } 270 }
227 271
228# $self->on_eof (delete $self->{on_eof} ) if $self->{on_eof}; # nop
229# $self->on_error (delete $self->{on_error}) if $self->{on_error}; # nop
230# $self->on_read (delete $self->{on_read} ) if $self->{on_read}; # nop
231 $self->on_drain (delete $self->{on_drain}) if $self->{on_drain};
232
233 $self->{_activity} = time; 272 $self->{_activity} = AnyEvent->now;
234 $self->_timeout; 273 $self->_timeout;
235 274
275 $self->on_drain (delete $self->{on_drain}) if exists $self->{on_drain};
276 $self->no_delay (delete $self->{no_delay}) if exists $self->{no_delay};
277
236 $self->start_read; 278 $self->start_read
279 if $self->{on_read};
237 280
238 $self 281 $self
239} 282}
240 283
241sub _shutdown { 284sub _shutdown {
242 my ($self) = @_; 285 my ($self) = @_;
243 286
287 delete $self->{_tw};
244 delete $self->{_rw}; 288 delete $self->{_rw};
245 delete $self->{_ww}; 289 delete $self->{_ww};
246 delete $self->{fh}; 290 delete $self->{fh};
247}
248 291
292 $self->stoptls;
293}
294
249sub error { 295sub _error {
250 my ($self) = @_; 296 my ($self, $errno, $fatal) = @_;
251 297
252 {
253 local $!;
254 $self->_shutdown; 298 $self->_shutdown
255 } 299 if $fatal;
256 300
257 $self->{on_error}($self) 301 $! = $errno;
302
258 if $self->{on_error}; 303 if ($self->{on_error}) {
259 304 $self->{on_error}($self, $fatal);
305 } else {
260 Carp::croak "AnyEvent::Handle uncaught fatal error: $!"; 306 Carp::croak "AnyEvent::Handle uncaught error: $!";
307 }
261} 308}
262 309
263=item $fh = $handle->fh 310=item $fh = $handle->fh
264 311
265This method returns the file handle of the L<AnyEvent::Handle> object. 312This method returns the file handle of the L<AnyEvent::Handle> object.
296 343
297=cut 344=cut
298 345
299sub on_timeout { 346sub on_timeout {
300 $_[0]{on_timeout} = $_[1]; 347 $_[0]{on_timeout} = $_[1];
348}
349
350=item $handle->autocork ($boolean)
351
352Enables or disables the current autocork behaviour (see C<autocork>
353constructor argument).
354
355=cut
356
357=item $handle->no_delay ($boolean)
358
359Enables or disables the C<no_delay> setting (see constructor argument of
360the same name for details).
361
362=cut
363
364sub no_delay {
365 $_[0]{no_delay} = $_[1];
366
367 eval {
368 local $SIG{__DIE__};
369 setsockopt $_[0]{fh}, &Socket::IPPROTO_TCP, &Socket::TCP_NODELAY, int $_[1];
370 };
301} 371}
302 372
303############################################################################# 373#############################################################################
304 374
305=item $handle->timeout ($seconds) 375=item $handle->timeout ($seconds)
319# also check for time-outs 389# also check for time-outs
320sub _timeout { 390sub _timeout {
321 my ($self) = @_; 391 my ($self) = @_;
322 392
323 if ($self->{timeout}) { 393 if ($self->{timeout}) {
324 my $NOW = time; 394 my $NOW = AnyEvent->now;
325 395
326 # when would the timeout trigger? 396 # when would the timeout trigger?
327 my $after = $self->{_activity} + $self->{timeout} - $NOW; 397 my $after = $self->{_activity} + $self->{timeout} - $NOW;
328
329 warn "next to in $after\n";#d#
330 398
331 # now or in the past already? 399 # now or in the past already?
332 if ($after <= 0) { 400 if ($after <= 0) {
333 $self->{_activity} = $NOW; 401 $self->{_activity} = $NOW;
334 402
335 if ($self->{on_timeout}) { 403 if ($self->{on_timeout}) {
336 $self->{on_timeout}->($self); 404 $self->{on_timeout}($self);
337 } else { 405 } else {
338 $! = Errno::ETIMEDOUT; 406 $self->_error (&Errno::ETIMEDOUT);
339 $self->error;
340 } 407 }
341 408
342 # callbakx could have changed timeout value, optimise 409 # callback could have changed timeout value, optimise
343 return unless $self->{timeout}; 410 return unless $self->{timeout};
344 411
345 # calculate new after 412 # calculate new after
346 $after = $self->{timeout}; 413 $after = $self->{timeout};
347 } 414 }
348 415
349 Scalar::Util::weaken $self; 416 Scalar::Util::weaken $self;
417 return unless $self; # ->error could have destroyed $self
350 418
351 warn "after $after\n";#d#
352 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub { 419 $self->{_tw} ||= AnyEvent->timer (after => $after, cb => sub {
353 delete $self->{_tw}; 420 delete $self->{_tw};
354 $self->_timeout; 421 $self->_timeout;
355 }); 422 });
356 } else { 423 } else {
410 my $len = syswrite $self->{fh}, $self->{wbuf}; 477 my $len = syswrite $self->{fh}, $self->{wbuf};
411 478
412 if ($len >= 0) { 479 if ($len >= 0) {
413 substr $self->{wbuf}, 0, $len, ""; 480 substr $self->{wbuf}, 0, $len, "";
414 481
415 $self->{_activity} = time; 482 $self->{_activity} = AnyEvent->now;
416 483
417 $self->{on_drain}($self) 484 $self->{on_drain}($self)
418 if $self->{low_water_mark} >= length $self->{wbuf} 485 if $self->{low_water_mark} >= length $self->{wbuf}
419 && $self->{on_drain}; 486 && $self->{on_drain};
420 487
421 delete $self->{_ww} unless length $self->{wbuf}; 488 delete $self->{_ww} unless length $self->{wbuf};
422 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 489 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
423 $self->error; 490 $self->_error ($!, 1);
424 } 491 }
425 }; 492 };
426 493
427 # try to write data immediately 494 # try to write data immediately
428 $cb->(); 495 $cb->() unless $self->{autocork};
429 496
430 # if still data left in wbuf, we need to poll 497 # if still data left in wbuf, we need to poll
431 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb) 498 $self->{_ww} = AnyEvent->io (fh => $self->{fh}, poll => "w", cb => $cb)
432 if length $self->{wbuf}; 499 if length $self->{wbuf};
433 }; 500 };
448 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 515 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
449 ->($self, @_); 516 ->($self, @_);
450 } 517 }
451 518
452 if ($self->{filter_w}) { 519 if ($self->{filter_w}) {
453 $self->{filter_w}->($self, \$_[0]); 520 $self->{filter_w}($self, \$_[0]);
454 } else { 521 } else {
455 $self->{wbuf} .= $_[0]; 522 $self->{wbuf} .= $_[0];
456 $self->_drain_wbuf; 523 $self->_drain_wbuf;
457 } 524 }
458} 525}
459 526
460=item $handle->push_write (type => @args) 527=item $handle->push_write (type => @args)
461 528
462=item $handle->unshift_write (type => @args)
463
464Instead of formatting your data yourself, you can also let this module do 529Instead of formatting your data yourself, you can also let this module do
465the job by specifying a type and type-specific arguments. 530the job by specifying a type and type-specific arguments.
466 531
467Predefined types are (if you have ideas for additional types, feel free to 532Predefined types are (if you have ideas for additional types, feel free to
468drop by and tell us): 533drop by and tell us):
472=item netstring => $string 537=item netstring => $string
473 538
474Formats the given value as netstring 539Formats the given value as netstring
475(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them). 540(http://cr.yp.to/proto/netstrings.txt, this is not a recommendation to use them).
476 541
477=back
478
479=cut 542=cut
480 543
481register_write_type netstring => sub { 544register_write_type netstring => sub {
482 my ($self, $string) = @_; 545 my ($self, $string) = @_;
483 546
484 sprintf "%d:%s,", (length $string), $string 547 sprintf "%d:%s,", (length $string), $string
548};
549
550=item packstring => $format, $data
551
552An octet string prefixed with an encoded length. The encoding C<$format>
553uses the same format as a Perl C<pack> format, but must specify a single
554integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
555optional C<!>, C<< < >> or C<< > >> modifier).
556
557=cut
558
559register_write_type packstring => sub {
560 my ($self, $format, $string) = @_;
561
562 pack "$format/a*", $string
485}; 563};
486 564
487=item json => $array_or_hashref 565=item json => $array_or_hashref
488 566
489Encodes the given hash or array reference into a JSON object. Unless you 567Encodes the given hash or array reference into a JSON object. Unless you
523 601
524 $self->{json} ? $self->{json}->encode ($ref) 602 $self->{json} ? $self->{json}->encode ($ref)
525 : JSON::encode_json ($ref) 603 : JSON::encode_json ($ref)
526}; 604};
527 605
606=item storable => $reference
607
608Freezes the given reference using L<Storable> and writes it to the
609handle. Uses the C<nfreeze> format.
610
611=cut
612
613register_write_type storable => sub {
614 my ($self, $ref) = @_;
615
616 require Storable;
617
618 pack "w/a*", Storable::nfreeze ($ref)
619};
620
621=back
622
528=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args) 623=item AnyEvent::Handle::register_write_type type => $coderef->($handle, @args)
529 624
530This function (not method) lets you add your own types to C<push_write>. 625This function (not method) lets you add your own types to C<push_write>.
531Whenever the given C<type> is used, C<push_write> will invoke the code 626Whenever the given C<type> is used, C<push_write> will invoke the code
532reference with the handle object and the remaining arguments. 627reference with the handle object and the remaining arguments.
552ways, the "simple" way, using only C<on_read> and the "complex" way, using 647ways, the "simple" way, using only C<on_read> and the "complex" way, using
553a queue. 648a queue.
554 649
555In the simple case, you just install an C<on_read> callback and whenever 650In the simple case, you just install an C<on_read> callback and whenever
556new data arrives, it will be called. You can then remove some data (if 651new data arrives, it will be called. You can then remove some data (if
557enough is there) from the read buffer (C<< $handle->rbuf >>) if you want 652enough is there) from the read buffer (C<< $handle->rbuf >>). Or you cna
558or not. 653leave the data there if you want to accumulate more (e.g. when only a
654partial message has been received so far).
559 655
560In the more complex case, you want to queue multiple callbacks. In this 656In the more complex case, you want to queue multiple callbacks. In this
561case, AnyEvent::Handle will call the first queued callback each time new 657case, AnyEvent::Handle will call the first queued callback each time new
562data arrives and removes it when it has done its job (see C<push_read>, 658data arrives (also the first time it is queued) and removes it when it has
563below). 659done its job (see C<push_read>, below).
564 660
565This way you can, for example, push three line-reads, followed by reading 661This way you can, for example, push three line-reads, followed by reading
566a chunk of data, and AnyEvent::Handle will execute them in order. 662a chunk of data, and AnyEvent::Handle will execute them in order.
567 663
568Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by 664Example 1: EPP protocol parser. EPP sends 4 byte length info, followed by
569the specified number of bytes which give an XML datagram. 665the specified number of bytes which give an XML datagram.
570 666
571 # in the default state, expect some header bytes 667 # in the default state, expect some header bytes
572 $handle->on_read (sub { 668 $handle->on_read (sub {
573 # some data is here, now queue the length-header-read (4 octets) 669 # some data is here, now queue the length-header-read (4 octets)
574 shift->unshift_read_chunk (4, sub { 670 shift->unshift_read (chunk => 4, sub {
575 # header arrived, decode 671 # header arrived, decode
576 my $len = unpack "N", $_[1]; 672 my $len = unpack "N", $_[1];
577 673
578 # now read the payload 674 # now read the payload
579 shift->unshift_read_chunk ($len, sub { 675 shift->unshift_read (chunk => $len, sub {
580 my $xml = $_[1]; 676 my $xml = $_[1];
581 # handle xml 677 # handle xml
582 }); 678 });
583 }); 679 });
584 }); 680 });
585 681
586Example 2: Implement a client for a protocol that replies either with 682Example 2: Implement a client for a protocol that replies either with "OK"
587"OK" and another line or "ERROR" for one request, and 64 bytes for the 683and another line or "ERROR" for the first request that is sent, and 64
588second request. Due tot he availability of a full queue, we can just 684bytes for the second request. Due to the availability of a queue, we can
589pipeline sending both requests and manipulate the queue as necessary in 685just pipeline sending both requests and manipulate the queue as necessary
590the callbacks: 686in the callbacks.
591 687
592 # request one 688When the first callback is called and sees an "OK" response, it will
689C<unshift> another line-read. This line-read will be queued I<before> the
69064-byte chunk callback.
691
692 # request one, returns either "OK + extra line" or "ERROR"
593 $handle->push_write ("request 1\015\012"); 693 $handle->push_write ("request 1\015\012");
594 694
595 # we expect "ERROR" or "OK" as response, so push a line read 695 # we expect "ERROR" or "OK" as response, so push a line read
596 $handle->push_read_line (sub { 696 $handle->push_read (line => sub {
597 # if we got an "OK", we have to _prepend_ another line, 697 # if we got an "OK", we have to _prepend_ another line,
598 # so it will be read before the second request reads its 64 bytes 698 # so it will be read before the second request reads its 64 bytes
599 # which are already in the queue when this callback is called 699 # which are already in the queue when this callback is called
600 # we don't do this in case we got an error 700 # we don't do this in case we got an error
601 if ($_[1] eq "OK") { 701 if ($_[1] eq "OK") {
602 $_[0]->unshift_read_line (sub { 702 $_[0]->unshift_read (line => sub {
603 my $response = $_[1]; 703 my $response = $_[1];
604 ... 704 ...
605 }); 705 });
606 } 706 }
607 }); 707 });
608 708
609 # request two 709 # request two, simply returns 64 octets
610 $handle->push_write ("request 2\015\012"); 710 $handle->push_write ("request 2\015\012");
611 711
612 # simply read 64 bytes, always 712 # simply read 64 bytes, always
613 $handle->push_read_chunk (64, sub { 713 $handle->push_read (chunk => 64, sub {
614 my $response = $_[1]; 714 my $response = $_[1];
615 ... 715 ...
616 }); 716 });
617 717
618=over 4 718=over 4
619 719
620=cut 720=cut
621 721
622sub _drain_rbuf { 722sub _drain_rbuf {
623 my ($self) = @_; 723 my ($self) = @_;
724
725 local $self->{_in_drain} = 1;
624 726
625 if ( 727 if (
626 defined $self->{rbuf_max} 728 defined $self->{rbuf_max}
627 && $self->{rbuf_max} < length $self->{rbuf} 729 && $self->{rbuf_max} < length $self->{rbuf}
628 ) { 730 ) {
629 $! = &Errno::ENOSPC; 731 return $self->_error (&Errno::ENOSPC, 1);
630 $self->error;
631 } 732 }
632 733
633 return if $self->{in_drain}; 734 while () {
634 local $self->{in_drain} = 1;
635
636 while (my $len = length $self->{rbuf}) { 735 my $len = length $self->{rbuf};
637 no strict 'refs'; 736
638 if (my $cb = shift @{ $self->{_queue} }) { 737 if (my $cb = shift @{ $self->{_queue} }) {
639 unless ($cb->($self)) { 738 unless ($cb->($self)) {
640 if ($self->{_eof}) { 739 if ($self->{_eof}) {
641 # no progress can be made (not enough data and no data forthcoming) 740 # no progress can be made (not enough data and no data forthcoming)
642 $! = &Errno::EPIPE; 741 $self->_error (&Errno::EPIPE, 1), last;
643 $self->error;
644 } 742 }
645 743
646 unshift @{ $self->{_queue} }, $cb; 744 unshift @{ $self->{_queue} }, $cb;
647 return; 745 last;
648 } 746 }
649 } elsif ($self->{on_read}) { 747 } elsif ($self->{on_read}) {
748 last unless $len;
749
650 $self->{on_read}($self); 750 $self->{on_read}($self);
651 751
652 if ( 752 if (
653 $self->{_eof} # if no further data will arrive
654 && $len == length $self->{rbuf} # and no data has been consumed 753 $len == length $self->{rbuf} # if no data has been consumed
655 && !@{ $self->{_queue} } # and the queue is still empty 754 && !@{ $self->{_queue} } # and the queue is still empty
656 && $self->{on_read} # and we still want to read data 755 && $self->{on_read} # but we still have on_read
657 ) { 756 ) {
757 # no further data will arrive
658 # then no progress can be made 758 # so no progress can be made
659 $! = &Errno::EPIPE; 759 $self->_error (&Errno::EPIPE, 1), last
660 $self->error; 760 if $self->{_eof};
761
762 last; # more data might arrive
661 } 763 }
662 } else { 764 } else {
663 # read side becomes idle 765 # read side becomes idle
664 delete $self->{_rw}; 766 delete $self->{_rw};
665 return; 767 last;
666 } 768 }
667 } 769 }
668 770
669 if ($self->{_eof}) {
670 $self->_shutdown;
671 $self->{on_eof}($self) 771 $self->{on_eof}($self)
672 if $self->{on_eof}; 772 if $self->{_eof} && $self->{on_eof};
773
774 # may need to restart read watcher
775 unless ($self->{_rw}) {
776 $self->start_read
777 if $self->{on_read} || @{ $self->{_queue} };
673 } 778 }
674} 779}
675 780
676=item $handle->on_read ($cb) 781=item $handle->on_read ($cb)
677 782
683 788
684sub on_read { 789sub on_read {
685 my ($self, $cb) = @_; 790 my ($self, $cb) = @_;
686 791
687 $self->{on_read} = $cb; 792 $self->{on_read} = $cb;
793 $self->_drain_rbuf if $cb && !$self->{_in_drain};
688} 794}
689 795
690=item $handle->rbuf 796=item $handle->rbuf
691 797
692Returns the read buffer (as a modifiable lvalue). 798Returns the read buffer (as a modifiable lvalue).
741 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read") 847 $cb = ($RH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_read")
742 ->($self, $cb, @_); 848 ->($self, $cb, @_);
743 } 849 }
744 850
745 push @{ $self->{_queue} }, $cb; 851 push @{ $self->{_queue} }, $cb;
746 $self->_drain_rbuf; 852 $self->_drain_rbuf unless $self->{_in_drain};
747} 853}
748 854
749sub unshift_read { 855sub unshift_read {
750 my $self = shift; 856 my $self = shift;
751 my $cb = pop; 857 my $cb = pop;
757 ->($self, $cb, @_); 863 ->($self, $cb, @_);
758 } 864 }
759 865
760 866
761 unshift @{ $self->{_queue} }, $cb; 867 unshift @{ $self->{_queue} }, $cb;
762 $self->_drain_rbuf; 868 $self->_drain_rbuf unless $self->{_in_drain};
763} 869}
764 870
765=item $handle->push_read (type => @args, $cb) 871=item $handle->push_read (type => @args, $cb)
766 872
767=item $handle->unshift_read (type => @args, $cb) 873=item $handle->unshift_read (type => @args, $cb)
797 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, ""); 903 $cb->($_[0], substr $_[0]{rbuf}, 0, $len, "");
798 1 904 1
799 } 905 }
800}; 906};
801 907
802# compatibility with older API
803sub push_read_chunk {
804 $_[0]->push_read (chunk => $_[1], $_[2]);
805}
806
807sub unshift_read_chunk {
808 $_[0]->unshift_read (chunk => $_[1], $_[2]);
809}
810
811=item line => [$eol, ]$cb->($handle, $line, $eol) 908=item line => [$eol, ]$cb->($handle, $line, $eol)
812 909
813The callback will be called only once a full line (including the end of 910The callback will be called only once a full line (including the end of
814line marker, C<$eol>) has been read. This line (excluding the end of line 911line marker, C<$eol>) has been read. This line (excluding the end of line
815marker) will be passed to the callback as second argument (C<$line>), and 912marker) will be passed to the callback as second argument (C<$line>), and
830=cut 927=cut
831 928
832register_read_type line => sub { 929register_read_type line => sub {
833 my ($self, $cb, $eol) = @_; 930 my ($self, $cb, $eol) = @_;
834 931
835 $eol = qr|(\015?\012)| if @_ < 3; 932 if (@_ < 3) {
836 $eol = quotemeta $eol unless ref $eol; 933 # this is more than twice as fast as the generic code below
837 $eol = qr|^(.*?)($eol)|s;
838
839 sub { 934 sub {
840 $_[0]{rbuf} =~ s/$eol// or return; 935 $_[0]{rbuf} =~ s/^([^\015\012]*)(\015?\012)// or return;
841 936
842 $cb->($_[0], $1, $2); 937 $cb->($_[0], $1, $2);
843 1
844 }
845};
846
847# compatibility with older API
848sub push_read_line {
849 my $self = shift;
850 $self->push_read (line => @_);
851}
852
853sub unshift_read_line {
854 my $self = shift;
855 $self->unshift_read (line => @_);
856}
857
858=item netstring => $cb->($handle, $string)
859
860A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
861
862Throws an error with C<$!> set to EBADMSG on format violations.
863
864=cut
865
866register_read_type netstring => sub {
867 my ($self, $cb) = @_;
868
869 sub {
870 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
871 if ($_[0]{rbuf} =~ /[^0-9]/) {
872 $! = &Errno::EBADMSG;
873 $self->error;
874 } 938 1
875 return;
876 } 939 }
940 } else {
941 $eol = quotemeta $eol unless ref $eol;
942 $eol = qr|^(.*?)($eol)|s;
877 943
878 my $len = $1; 944 sub {
945 $_[0]{rbuf} =~ s/$eol// or return;
879 946
880 $self->unshift_read (chunk => $len, sub { 947 $cb->($_[0], $1, $2);
881 my $string = $_[1];
882 $_[0]->unshift_read (chunk => 1, sub {
883 if ($_[1] eq ",") {
884 $cb->($_[0], $string);
885 } else {
886 $! = &Errno::EBADMSG;
887 $self->error;
888 }
889 }); 948 1
890 }); 949 }
891
892 1
893 } 950 }
894}; 951};
895 952
896=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 953=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
897 954
949 return 1; 1006 return 1;
950 } 1007 }
951 1008
952 # reject 1009 # reject
953 if ($reject && $$rbuf =~ $reject) { 1010 if ($reject && $$rbuf =~ $reject) {
954 $! = &Errno::EBADMSG; 1011 $self->_error (&Errno::EBADMSG);
955 $self->error;
956 } 1012 }
957 1013
958 # skip 1014 # skip
959 if ($skip && $$rbuf =~ $skip) { 1015 if ($skip && $$rbuf =~ $skip) {
960 $data .= substr $$rbuf, 0, $+[0], ""; 1016 $data .= substr $$rbuf, 0, $+[0], "";
962 1018
963 () 1019 ()
964 } 1020 }
965}; 1021};
966 1022
1023=item netstring => $cb->($handle, $string)
1024
1025A netstring (http://cr.yp.to/proto/netstrings.txt, this is not an endorsement).
1026
1027Throws an error with C<$!> set to EBADMSG on format violations.
1028
1029=cut
1030
1031register_read_type netstring => sub {
1032 my ($self, $cb) = @_;
1033
1034 sub {
1035 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
1036 if ($_[0]{rbuf} =~ /[^0-9]/) {
1037 $self->_error (&Errno::EBADMSG);
1038 }
1039 return;
1040 }
1041
1042 my $len = $1;
1043
1044 $self->unshift_read (chunk => $len, sub {
1045 my $string = $_[1];
1046 $_[0]->unshift_read (chunk => 1, sub {
1047 if ($_[1] eq ",") {
1048 $cb->($_[0], $string);
1049 } else {
1050 $self->_error (&Errno::EBADMSG);
1051 }
1052 });
1053 });
1054
1055 1
1056 }
1057};
1058
1059=item packstring => $format, $cb->($handle, $string)
1060
1061An octet string prefixed with an encoded length. The encoding C<$format>
1062uses the same format as a Perl C<pack> format, but must specify a single
1063integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
1064optional C<!>, C<< < >> or C<< > >> modifier).
1065
1066DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>.
1067
1068Example: read a block of data prefixed by its length in BER-encoded
1069format (very efficient).
1070
1071 $handle->push_read (packstring => "w", sub {
1072 my ($handle, $data) = @_;
1073 });
1074
1075=cut
1076
1077register_read_type packstring => sub {
1078 my ($self, $cb, $format) = @_;
1079
1080 sub {
1081 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1082 defined (my $len = eval { unpack $format, $_[0]{rbuf} })
1083 or return;
1084
1085 $format = length pack $format, $len;
1086
1087 # bypass unshift if we already have the remaining chunk
1088 if ($format + $len <= length $_[0]{rbuf}) {
1089 my $data = substr $_[0]{rbuf}, $format, $len;
1090 substr $_[0]{rbuf}, 0, $format + $len, "";
1091 $cb->($_[0], $data);
1092 } else {
1093 # remove prefix
1094 substr $_[0]{rbuf}, 0, $format, "";
1095
1096 # read remaining chunk
1097 $_[0]->unshift_read (chunk => $len, $cb);
1098 }
1099
1100 1
1101 }
1102};
1103
967=item json => $cb->($handle, $hash_or_arrayref) 1104=item json => $cb->($handle, $hash_or_arrayref)
968 1105
969Reads a JSON object or array, decodes it and passes it to the callback. 1106Reads a JSON object or array, decodes it and passes it to the callback.
970 1107
971If a C<json> object was passed to the constructor, then that will be used 1108If a C<json> object was passed to the constructor, then that will be used
981the C<json> write type description, above, for an actual example. 1118the C<json> write type description, above, for an actual example.
982 1119
983=cut 1120=cut
984 1121
985register_read_type json => sub { 1122register_read_type json => sub {
986 my ($self, $cb, $accept, $reject, $skip) = @_; 1123 my ($self, $cb) = @_;
987 1124
988 require JSON; 1125 require JSON;
989 1126
990 my $data; 1127 my $data;
991 my $rbuf = \$self->{rbuf}; 1128 my $rbuf = \$self->{rbuf};
1006 () 1143 ()
1007 } 1144 }
1008 } 1145 }
1009}; 1146};
1010 1147
1148=item storable => $cb->($handle, $ref)
1149
1150Deserialises a L<Storable> frozen representation as written by the
1151C<storable> write type (BER-encoded length prefix followed by nfreeze'd
1152data).
1153
1154Raises C<EBADMSG> error if the data could not be decoded.
1155
1156=cut
1157
1158register_read_type storable => sub {
1159 my ($self, $cb) = @_;
1160
1161 require Storable;
1162
1163 sub {
1164 # when we can use 5.10 we can use ".", but for 5.8 we use the re-pack method
1165 defined (my $len = eval { unpack "w", $_[0]{rbuf} })
1166 or return;
1167
1168 my $format = length pack "w", $len;
1169
1170 # bypass unshift if we already have the remaining chunk
1171 if ($format + $len <= length $_[0]{rbuf}) {
1172 my $data = substr $_[0]{rbuf}, $format, $len;
1173 substr $_[0]{rbuf}, 0, $format + $len, "";
1174 $cb->($_[0], Storable::thaw ($data));
1175 } else {
1176 # remove prefix
1177 substr $_[0]{rbuf}, 0, $format, "";
1178
1179 # read remaining chunk
1180 $_[0]->unshift_read (chunk => $len, sub {
1181 if (my $ref = eval { Storable::thaw ($_[1]) }) {
1182 $cb->($_[0], $ref);
1183 } else {
1184 $self->_error (&Errno::EBADMSG);
1185 }
1186 });
1187 }
1188
1189 1
1190 }
1191};
1192
1011=back 1193=back
1012 1194
1013=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args) 1195=item AnyEvent::Handle::register_read_type type => $coderef->($handle, $cb, @args)
1014 1196
1015This function (not method) lets you add your own types to C<push_read>. 1197This function (not method) lets you add your own types to C<push_read>.
1033=item $handle->stop_read 1215=item $handle->stop_read
1034 1216
1035=item $handle->start_read 1217=item $handle->start_read
1036 1218
1037In rare cases you actually do not want to read anything from the 1219In rare cases you actually do not want to read anything from the
1038socket. In this case you can call C<stop_read>. Neither C<on_read> no 1220socket. In this case you can call C<stop_read>. Neither C<on_read> nor
1039any queued callbacks will be executed then. To start reading again, call 1221any queued callbacks will be executed then. To start reading again, call
1040C<start_read>. 1222C<start_read>.
1223
1224Note that AnyEvent::Handle will automatically C<start_read> for you when
1225you change the C<on_read> callback or push/unshift a read callback, and it
1226will automatically C<stop_read> for you when neither C<on_read> is set nor
1227there are any read requests in the queue.
1041 1228
1042=cut 1229=cut
1043 1230
1044sub stop_read { 1231sub stop_read {
1045 my ($self) = @_; 1232 my ($self) = @_;
1056 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1243 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
1057 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf}; 1244 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf};
1058 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1245 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
1059 1246
1060 if ($len > 0) { 1247 if ($len > 0) {
1061 $self->{_activity} = time; 1248 $self->{_activity} = AnyEvent->now;
1062 1249
1063 $self->{filter_r} 1250 $self->{filter_r}
1064 ? $self->{filter_r}->($self, $rbuf) 1251 ? $self->{filter_r}($self, $rbuf)
1065 : $self->_drain_rbuf; 1252 : $self->{_in_drain} || $self->_drain_rbuf;
1066 1253
1067 } elsif (defined $len) { 1254 } elsif (defined $len) {
1068 delete $self->{_rw}; 1255 delete $self->{_rw};
1069 delete $self->{_ww};
1070 delete $self->{_tw};
1071 $self->{_eof} = 1; 1256 $self->{_eof} = 1;
1072 $self->_drain_rbuf; 1257 $self->_drain_rbuf unless $self->{_in_drain};
1073 1258
1074 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 1259 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
1075 return $self->error; 1260 return $self->_error ($!, 1);
1076 } 1261 }
1077 }); 1262 });
1078 } 1263 }
1079} 1264}
1080 1265
1081sub _dotls { 1266sub _dotls {
1082 my ($self) = @_; 1267 my ($self) = @_;
1268
1269 my $buf;
1083 1270
1084 if (length $self->{_tls_wbuf}) { 1271 if (length $self->{_tls_wbuf}) {
1085 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { 1272 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
1086 substr $self->{_tls_wbuf}, 0, $len, ""; 1273 substr $self->{_tls_wbuf}, 0, $len, "";
1087 } 1274 }
1088 } 1275 }
1089 1276
1090 if (defined (my $buf = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1277 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1091 $self->{wbuf} .= $buf; 1278 $self->{wbuf} .= $buf;
1092 $self->_drain_wbuf; 1279 $self->_drain_wbuf;
1093 } 1280 }
1094 1281
1095 while (defined (my $buf = Net::SSLeay::read ($self->{tls}))) { 1282 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) {
1283 if (length $buf) {
1096 $self->{rbuf} .= $buf; 1284 $self->{rbuf} .= $buf;
1097 $self->_drain_rbuf; 1285 $self->_drain_rbuf unless $self->{_in_drain};
1286 } else {
1287 # let's treat SSL-eof as we treat normal EOF
1288 $self->{_eof} = 1;
1289 $self->_shutdown;
1290 return;
1291 }
1098 } 1292 }
1099 1293
1100 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1294 my $err = Net::SSLeay::get_error ($self->{tls}, -1);
1101 1295
1102 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1296 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) {
1103 if ($err == Net::SSLeay::ERROR_SYSCALL ()) { 1297 if ($err == Net::SSLeay::ERROR_SYSCALL ()) {
1104 $self->error; 1298 return $self->_error ($!, 1);
1105 } elsif ($err == Net::SSLeay::ERROR_SSL ()) { 1299 } elsif ($err == Net::SSLeay::ERROR_SSL ()) {
1106 $! = &Errno::EIO; 1300 return $self->_error (&Errno::EIO, 1);
1107 $self->error;
1108 } 1301 }
1109 1302
1110 # all others are fine for our purposes 1303 # all others are fine for our purposes
1111 } 1304 }
1112} 1305}
1127call and can be used or changed to your liking. Note that the handshake 1320call and can be used or changed to your liking. Note that the handshake
1128might have already started when this function returns. 1321might have already started when this function returns.
1129 1322
1130=cut 1323=cut
1131 1324
1132# TODO: maybe document...
1133sub starttls { 1325sub starttls {
1134 my ($self, $ssl, $ctx) = @_; 1326 my ($self, $ssl, $ctx) = @_;
1135 1327
1136 $self->stoptls; 1328 $self->stoptls;
1137 1329
1190 1382
1191sub DESTROY { 1383sub DESTROY {
1192 my $self = shift; 1384 my $self = shift;
1193 1385
1194 $self->stoptls; 1386 $self->stoptls;
1387
1388 my $linger = exists $self->{linger} ? $self->{linger} : 3600;
1389
1390 if ($linger && length $self->{wbuf}) {
1391 my $fh = delete $self->{fh};
1392 my $wbuf = delete $self->{wbuf};
1393
1394 my @linger;
1395
1396 push @linger, AnyEvent->io (fh => $fh, poll => "w", cb => sub {
1397 my $len = syswrite $fh, $wbuf, length $wbuf;
1398
1399 if ($len > 0) {
1400 substr $wbuf, 0, $len, "";
1401 } else {
1402 @linger = (); # end
1403 }
1404 });
1405 push @linger, AnyEvent->timer (after => $linger, cb => sub {
1406 @linger = ();
1407 });
1408 }
1195} 1409}
1196 1410
1197=item AnyEvent::Handle::TLS_CTX 1411=item AnyEvent::Handle::TLS_CTX
1198 1412
1199This function creates and returns the Net::SSLeay::CTX object used by 1413This function creates and returns the Net::SSLeay::CTX object used by
1241=over 4 1455=over 4
1242 1456
1243=item * all constructor arguments become object members. 1457=item * all constructor arguments become object members.
1244 1458
1245At least initially, when you pass a C<tls>-argument to the constructor it 1459At least initially, when you pass a C<tls>-argument to the constructor it
1246will end up in C<< $handle->{tls} >>. Those members might be changes or 1460will end up in C<< $handle->{tls} >>. Those members might be changed or
1247mutated later on (for example C<tls> will hold the TLS connection object). 1461mutated later on (for example C<tls> will hold the TLS connection object).
1248 1462
1249=item * other object member names are prefixed with an C<_>. 1463=item * other object member names are prefixed with an C<_>.
1250 1464
1251All object members not explicitly documented (internal use) are prefixed 1465All object members not explicitly documented (internal use) are prefixed

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines