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.90 by root, Mon Sep 29 02:08:57 2008 UTC vs.
Revision 1.97 by root, Thu Oct 2 11:07:59 2008 UTC

… …
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 = 4.234; 19our $VERSION = 4.3;
20 20
21=head1 SYNOPSIS 21=head1 SYNOPSIS
22 22
23 use AnyEvent; 23 use AnyEvent;
24 use AnyEvent::Handle; 24 use AnyEvent::Handle;
… …
240write data and will install a watcher that will write this data to the 240write data and will install a watcher that will write this data to the
241socket. No errors will be reported (this mostly matches how the operating 241socket. No errors will be reported (this mostly matches how the operating
242system treats outstanding data at socket close time). 242system treats outstanding data at socket close time).
243 243
244This will not work for partial TLS data that could not be encoded 244This will not work for partial TLS data that could not be encoded
245yet. This data will be lost. 245yet. This data will be lost. Calling the C<stoptls> method in time might
246help.
246 247
247=item tls => "accept" | "connect" | Net::SSLeay::SSL object 248=item tls => "accept" | "connect" | Net::SSLeay::SSL object
248 249
249When this parameter is given, it enables TLS (SSL) mode, that means 250When this parameter is given, it enables TLS (SSL) mode, that means
250AnyEvent will start a TLS handshake as soon as the conenction has been 251AnyEvent will start a TLS handshake as soon as the conenction has been
… …
281texts. 282texts.
282 283
283Note that you are responsible to depend on the JSON module if you want to 284Note that you are responsible to depend on the JSON module if you want to
284use this functionality, as AnyEvent does not have a dependency itself. 285use this functionality, as AnyEvent does not have a dependency itself.
285 286
286=item filter_r => $cb
287
288=item filter_w => $cb
289
290These exist, but are undocumented at this time. (They are used internally
291by the TLS code).
292
293=back 287=back
294 288
295=cut 289=cut
296 290
297sub new { 291sub new {
… …
301 295
302 $self->{fh} or Carp::croak "mandatory argument fh is missing"; 296 $self->{fh} or Carp::croak "mandatory argument fh is missing";
303 297
304 AnyEvent::Util::fh_nonblocking $self->{fh}, 1; 298 AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
305 299
306 if ($self->{tls}) {
307 require Net::SSLeay;
308 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 300 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx})
309 } 301 if $self->{tls};
310 302
311 $self->{_activity} = AnyEvent->now; 303 $self->{_activity} = AnyEvent->now;
312 $self->_timeout; 304 $self->_timeout;
313 305
314 $self->on_drain (delete $self->{on_drain}) if exists $self->{on_drain}; 306 $self->on_drain (delete $self->{on_drain}) if exists $self->{on_drain};
… …
326 delete $self->{_tw}; 318 delete $self->{_tw};
327 delete $self->{_rw}; 319 delete $self->{_rw};
328 delete $self->{_ww}; 320 delete $self->{_ww};
329 delete $self->{fh}; 321 delete $self->{fh};
330 322
331 $self->stoptls; 323 &_freetls;
332 324
333 delete $self->{on_read}; 325 delete $self->{on_read};
334 delete $self->{_queue}; 326 delete $self->{_queue};
335} 327}
336 328
… …
495 my ($self, $cb) = @_; 487 my ($self, $cb) = @_;
496 488
497 $self->{on_drain} = $cb; 489 $self->{on_drain} = $cb;
498 490
499 $cb->($self) 491 $cb->($self)
500 if $cb && $self->{low_water_mark} >= length $self->{wbuf}; 492 if $cb && $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf});
501} 493}
502 494
503=item $handle->push_write ($data) 495=item $handle->push_write ($data)
504 496
505Queues the given scalar to be written. You can push as much data as you 497Queues the given scalar to be written. You can push as much data as you
… …
522 substr $self->{wbuf}, 0, $len, ""; 514 substr $self->{wbuf}, 0, $len, "";
523 515
524 $self->{_activity} = AnyEvent->now; 516 $self->{_activity} = AnyEvent->now;
525 517
526 $self->{on_drain}($self) 518 $self->{on_drain}($self)
527 if $self->{low_water_mark} >= length $self->{wbuf} 519 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf})
528 && $self->{on_drain}; 520 && $self->{on_drain};
529 521
530 delete $self->{_ww} unless length $self->{wbuf}; 522 delete $self->{_ww} unless length $self->{wbuf};
531 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 523 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
532 $self->_error ($!, 1); 524 $self->_error ($!, 1);
… …
556 548
557 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 549 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
558 ->($self, @_); 550 ->($self, @_);
559 } 551 }
560 552
561 if ($self->{filter_w}) { 553 if ($self->{tls}) {
562 $self->{filter_w}($self, \$_[0]); 554 $self->{_tls_wbuf} .= $_[0];
555
556 &_dotls ($self);
563 } else { 557 } else {
564 $self->{wbuf} .= $_[0]; 558 $self->{wbuf} .= $_[0];
565 $self->_drain_wbuf; 559 $self->_drain_wbuf;
566 } 560 }
567} 561}
… …
584=cut 578=cut
585 579
586register_write_type netstring => sub { 580register_write_type netstring => sub {
587 my ($self, $string) = @_; 581 my ($self, $string) = @_;
588 582
589 sprintf "%d:%s,", (length $string), $string 583 (length $string) . ":$string,"
590}; 584};
591 585
592=item packstring => $format, $data 586=item packstring => $format, $data
593 587
594An octet string prefixed with an encoded length. The encoding C<$format> 588An octet string prefixed with an encoded length. The encoding C<$format>
… …
803 797
804 last; # more data might arrive 798 last; # more data might arrive
805 } 799 }
806 } else { 800 } else {
807 # read side becomes idle 801 # read side becomes idle
808 delete $self->{_rw}; 802 delete $self->{_rw} unless $self->{tls};
809 last; 803 last;
810 } 804 }
811 } 805 }
812 806
813 if ($self->{_eof}) { 807 if ($self->{_eof}) {
… …
1108An octet string prefixed with an encoded length. The encoding C<$format> 1102An octet string prefixed with an encoded length. The encoding C<$format>
1109uses the same format as a Perl C<pack> format, but must specify a single 1103uses the same format as a Perl C<pack> format, but must specify a single
1110integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an 1104integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
1111optional C<!>, C<< < >> or C<< > >> modifier). 1105optional C<!>, C<< < >> or C<< > >> modifier).
1112 1106
1113DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>. 1107For example, DNS over TCP uses a prefix of C<n> (2 octet network order),
1108EPP uses a prefix of C<N> (4 octtes).
1114 1109
1115Example: read a block of data prefixed by its length in BER-encoded 1110Example: read a block of data prefixed by its length in BER-encoded
1116format (very efficient). 1111format (very efficient).
1117 1112
1118 $handle->push_read (packstring => "w", sub { 1113 $handle->push_read (packstring => "w", sub {
… …
1271Note that AnyEvent::Handle will automatically C<start_read> for you when 1266Note that AnyEvent::Handle will automatically C<start_read> for you when
1272you change the C<on_read> callback or push/unshift a read callback, and it 1267you change the C<on_read> callback or push/unshift a read callback, and it
1273will automatically C<stop_read> for you when neither C<on_read> is set nor 1268will automatically C<stop_read> for you when neither C<on_read> is set nor
1274there are any read requests in the queue. 1269there are any read requests in the queue.
1275 1270
1271These methods will have no effect when in TLS mode (as TLS doesn't support
1272half-duplex connections).
1273
1276=cut 1274=cut
1277 1275
1278sub stop_read { 1276sub stop_read {
1279 my ($self) = @_; 1277 my ($self) = @_;
1280 1278
1281 delete $self->{_rw}; 1279 delete $self->{_rw} unless $self->{tls};
1282} 1280}
1283 1281
1284sub start_read { 1282sub start_read {
1285 my ($self) = @_; 1283 my ($self) = @_;
1286 1284
1287 unless ($self->{_rw} || $self->{_eof}) { 1285 unless ($self->{_rw} || $self->{_eof}) {
1288 Scalar::Util::weaken $self; 1286 Scalar::Util::weaken $self;
1289 1287
1290 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1288 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
1291 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf}; 1289 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
1292 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1290 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
1293 1291
1294 if ($len > 0) { 1292 if ($len > 0) {
1295 $self->{_activity} = AnyEvent->now; 1293 $self->{_activity} = AnyEvent->now;
1296 1294
1297 $self->{filter_r} 1295 if ($self->{tls}) {
1298 ? $self->{filter_r}($self, $rbuf) 1296 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1299 : $self->{_in_drain} || $self->_drain_rbuf; 1297
1298 &_dotls ($self);
1299 } else {
1300 $self->_drain_rbuf unless $self->{_in_drain};
1301 }
1300 1302
1301 } elsif (defined $len) { 1303 } elsif (defined $len) {
1302 delete $self->{_rw}; 1304 delete $self->{_rw};
1303 $self->{_eof} = 1; 1305 $self->{_eof} = 1;
1304 $self->_drain_rbuf unless $self->{_in_drain}; 1306 $self->_drain_rbuf unless $self->{_in_drain};
… …
1308 } 1310 }
1309 }); 1311 });
1310 } 1312 }
1311} 1313}
1312 1314
1315# poll the write BIO and send the data if applicable
1313sub _dotls { 1316sub _dotls {
1314 my ($self) = @_; 1317 my ($self) = @_;
1315 1318
1316 my $buf; 1319 my $tmp;
1317 1320
1318 if (length $self->{_tls_wbuf}) { 1321 if (length $self->{_tls_wbuf}) {
1319 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { 1322 while (($tmp = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
1320 substr $self->{_tls_wbuf}, 0, $len, ""; 1323 substr $self->{_tls_wbuf}, 0, $tmp, "";
1321 } 1324 }
1322 } 1325 }
1323 1326
1324 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1325 $self->{wbuf} .= $buf;
1326 $self->_drain_wbuf;
1327 }
1328
1329 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) { 1327 while (defined ($tmp = Net::SSLeay::read ($self->{tls}))) {
1330 if (length $buf) { 1328 unless (length $tmp) {
1331 $self->{rbuf} .= $buf;
1332 $self->_drain_rbuf unless $self->{_in_drain};
1333 } else {
1334 # let's treat SSL-eof as we treat normal EOF 1329 # let's treat SSL-eof as we treat normal EOF
1330 delete $self->{_rw};
1335 $self->{_eof} = 1; 1331 $self->{_eof} = 1;
1336 $self->_shutdown; 1332 &_freetls;
1337 return;
1338 } 1333 }
1339 }
1340 1334
1335 $self->{rbuf} .= $tmp;
1336 $self->_drain_rbuf unless $self->{_in_drain};
1337 $self->{tls} or return; # tls session might have gone away in callback
1338 }
1339
1341 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1340 $tmp = Net::SSLeay::get_error ($self->{tls}, -1);
1342 1341
1343 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1342 if ($tmp != Net::SSLeay::ERROR_WANT_READ ()) {
1344 if ($err == Net::SSLeay::ERROR_SYSCALL ()) { 1343 if ($tmp == Net::SSLeay::ERROR_SYSCALL ()) {
1345 return $self->_error ($!, 1); 1344 return $self->_error ($!, 1);
1346 } elsif ($err == Net::SSLeay::ERROR_SSL ()) { 1345 } elsif ($tmp == Net::SSLeay::ERROR_SSL ()) {
1347 return $self->_error (&Errno::EIO, 1); 1346 return $self->_error (&Errno::EIO, 1);
1348 } 1347 }
1349 1348
1350 # all others are fine for our purposes 1349 # all other errors are fine for our purposes
1350 }
1351
1352 while (length ($tmp = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1353 $self->{wbuf} .= $tmp;
1354 $self->_drain_wbuf;
1351 } 1355 }
1352} 1356}
1353 1357
1354=item $handle->starttls ($tls[, $tls_ctx]) 1358=item $handle->starttls ($tls[, $tls_ctx])
1355 1359
… …
1365 1369
1366The TLS connection object will end up in C<< $handle->{tls} >> after this 1370The TLS connection object will end up in C<< $handle->{tls} >> after this
1367call and can be used or changed to your liking. Note that the handshake 1371call and can be used or changed to your liking. Note that the handshake
1368might have already started when this function returns. 1372might have already started when this function returns.
1369 1373
1374If it an error to start a TLS handshake more than once per
1375AnyEvent::Handle object (this is due to bugs in OpenSSL).
1376
1370=cut 1377=cut
1371 1378
1372sub starttls { 1379sub starttls {
1373 my ($self, $ssl, $ctx) = @_; 1380 my ($self, $ssl, $ctx) = @_;
1374 1381
1375 $self->stoptls; 1382 require Net::SSLeay;
1376 1383
1384 Carp::croak "it is an error to call starttls more than once on an Anyevent::Handle object"
1385 if $self->{tls};
1386
1377 if ($ssl eq "accept") { 1387 if ($ssl eq "accept") {
1378 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ()); 1388 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ());
1379 Net::SSLeay::set_accept_state ($ssl); 1389 Net::SSLeay::set_accept_state ($ssl);
1380 } elsif ($ssl eq "connect") { 1390 } elsif ($ssl eq "connect") {
1381 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ()); 1391 $ssl = Net::SSLeay::new ($ctx || TLS_CTX ());
… …
1390 # and mismaintained ssleay-module doesn't even offer them). 1400 # and mismaintained ssleay-module doesn't even offer them).
1391 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html 1401 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
1392 # 1402 #
1393 # in short: this is a mess. 1403 # in short: this is a mess.
1394 # 1404 #
1395 # note that we do not try to kepe the length constant between writes as we are required to do. 1405 # note that we do not try to keep the length constant between writes as we are required to do.
1396 # we assume that most (but not all) of this insanity only applies to non-blocking cases, 1406 # we assume that most (but not all) of this insanity only applies to non-blocking cases,
1397 # and we drive openssl fully in blocking mode here. 1407 # and we drive openssl fully in blocking mode here. Or maybe we don't - openssl seems to
1408 # have identity issues in that area.
1398 Net::SSLeay::CTX_set_mode ($self->{tls}, 1409 Net::SSLeay::CTX_set_mode ($self->{tls},
1399 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1) 1410 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
1400 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2)); 1411 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
1401 1412
1402 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1413 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1403 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1414 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1404 1415
1405 Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio}); 1416 Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio});
1406 1417
1407 $self->{filter_w} = sub { 1418 &_dotls; # need to trigger the initial handshake
1408 $_[0]{_tls_wbuf} .= ${$_[1]}; 1419 $self->start_read; # make sure we actually do read
1409 &_dotls;
1410 };
1411 $self->{filter_r} = sub {
1412 Net::SSLeay::BIO_write ($_[0]{_rbio}, ${$_[1]});
1413 &_dotls;
1414 };
1415} 1420}
1416 1421
1417=item $handle->stoptls 1422=item $handle->stoptls
1418 1423
1419Destroys the SSL connection, if any. Partial read or write data will be 1424Shuts down the SSL connection - this makes a proper EOF handshake by
1420lost. 1425sending a close notify to the other side, but since OpenSSL doesn't
1426support non-blocking shut downs, it is not possible to re-use the stream
1427afterwards.
1421 1428
1422=cut 1429=cut
1423 1430
1424sub stoptls { 1431sub stoptls {
1425 my ($self) = @_; 1432 my ($self) = @_;
1426 1433
1434 if ($self->{tls}) {
1435 Net::SSLeay::shutdown ($self->{tls});
1436
1437 &_dotls;
1438
1439 # we don't give a shit. no, we do, but we can't. no...
1440 # we, we... have to use openssl :/
1441 &_freetls;
1442 }
1443}
1444
1445sub _freetls {
1446 my ($self) = @_;
1447
1448 return unless $self->{tls};
1449
1427 Net::SSLeay::free (delete $self->{tls}) if $self->{tls}; 1450 Net::SSLeay::free (delete $self->{tls});
1428 1451
1429 delete $self->{_rbio}; 1452 delete @$self{qw(_rbio _wbio _tls_wbuf)};
1430 delete $self->{_wbio};
1431 delete $self->{_tls_wbuf};
1432 delete $self->{filter_r};
1433 delete $self->{filter_w};
1434} 1453}
1435 1454
1436sub DESTROY { 1455sub DESTROY {
1437 my $self = shift; 1456 my $self = shift;
1438 1457
1439 $self->stoptls; 1458 &_freetls;
1440 1459
1441 my $linger = exists $self->{linger} ? $self->{linger} : 3600; 1460 my $linger = exists $self->{linger} ? $self->{linger} : 3600;
1442 1461
1443 if ($linger && length $self->{wbuf}) { 1462 if ($linger && length $self->{wbuf}) {
1444 my $fh = delete $self->{fh}; 1463 my $fh = delete $self->{fh};
… …
1496 } 1515 }
1497} 1516}
1498 1517
1499=back 1518=back
1500 1519
1520
1521=head1 NONFREQUENTLY ASKED QUESTIONS
1522
1523=over 4
1524
1525=item How do I read data until the other side closes the connection?
1526
1527If you just want to read your data into a perl scalar, the easiest way
1528to achieve this is by setting an C<on_read> callback that does nothing,
1529clearing the C<on_eof> callback and in the C<on_error> callback, the data
1530will be in C<$_[0]{rbuf}>:
1531
1532 $handle->on_read (sub { });
1533 $handle->on_eof (undef);
1534 $handle->on_error (sub {
1535 my $data = delete $_[0]{rbuf};
1536 undef $handle;
1537 });
1538
1539The reason to use C<on_error> is that TCP connections, due to latencies
1540and packets loss, might get closed quite violently with an error, when in
1541fact, all data has been received.
1542
1543It is usually better to use acknowledgements when transfering data,
1544to make sure the other side hasn't just died and you got the data
1545intact. This is also one reason why so many internet protocols have an
1546explicit QUIT command.
1547
1548
1549=item I don't want to destroy the handle too early - how do I wait until
1550all data has been written?
1551
1552After writing your last bits of data, set the C<on_drain> callback
1553and destroy the handle in there - with the default setting of
1554C<low_water_mark> this will be called precisely when all data has been
1555written to the socket:
1556
1557 $handle->push_write (...);
1558 $handle->on_drain (sub {
1559 warn "all data submitted to the kernel\n";
1560 undef $handle;
1561 });
1562
1563=back
1564
1565
1501=head1 SUBCLASSING AnyEvent::Handle 1566=head1 SUBCLASSING AnyEvent::Handle
1502 1567
1503In many cases, you might want to subclass AnyEvent::Handle. 1568In many cases, you might want to subclass AnyEvent::Handle.
1504 1569
1505To make this easier, a given version of AnyEvent::Handle uses these 1570To make this easier, a given version of AnyEvent::Handle uses these

Diff Legend

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