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.92 by root, Wed Oct 1 08:52:06 2008 UTC vs.
Revision 1.99 by root, Thu Oct 23 02:41:00 2008 UTC

… …
59treatment of characters applies to this module as well. 59treatment of characters applies to this module as well.
60 60
61All callbacks will be invoked with the handle object as their first 61All callbacks will be invoked with the handle object as their first
62argument. 62argument.
63 63
64=head2 SIGPIPE is not handled by this module
65
66SIGPIPE is not handled by this module, so one of the practical
67requirements of using it is to ignore SIGPIPE (C<$SIG{PIPE} =
68'IGNORE'>). At least, this is highly recommend in a networked program: If
69you use AnyEvent::Handle in a filter program (like sort), exiting on
70SIGPIPE is probably the right thing to do.
71
72=head1 METHODS 64=head1 METHODS
73 65
74=over 4 66=over 4
75 67
76=item B<new (%args)> 68=item B<new (%args)>
… …
240write data and will install a watcher that will write this data to the 232write 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 233socket. No errors will be reported (this mostly matches how the operating
242system treats outstanding data at socket close time). 234system treats outstanding data at socket close time).
243 235
244This will not work for partial TLS data that could not be encoded 236This will not work for partial TLS data that could not be encoded
245yet. This data will be lost. 237yet. This data will be lost. Calling the C<stoptls> method in time might
238help.
246 239
247=item tls => "accept" | "connect" | Net::SSLeay::SSL object 240=item tls => "accept" | "connect" | Net::SSLeay::SSL object
248 241
249When this parameter is given, it enables TLS (SSL) mode, that means 242When this parameter is given, it enables TLS (SSL) mode, that means
250AnyEvent will start a TLS handshake as soon as the conenction has been 243AnyEvent will start a TLS handshake as soon as the conenction has been
… …
281texts. 274texts.
282 275
283Note that you are responsible to depend on the JSON module if you want to 276Note 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. 277use this functionality, as AnyEvent does not have a dependency itself.
285 278
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 279=back
294 280
295=cut 281=cut
296 282
297sub new { 283sub new {
… …
301 287
302 $self->{fh} or Carp::croak "mandatory argument fh is missing"; 288 $self->{fh} or Carp::croak "mandatory argument fh is missing";
303 289
304 AnyEvent::Util::fh_nonblocking $self->{fh}, 1; 290 AnyEvent::Util::fh_nonblocking $self->{fh}, 1;
305 291
306 if ($self->{tls}) {
307 require Net::SSLeay;
308 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx}); 292 $self->starttls (delete $self->{tls}, delete $self->{tls_ctx})
309 } 293 if $self->{tls};
310 294
311 $self->{_activity} = AnyEvent->now; 295 $self->{_activity} = AnyEvent->now;
312 $self->_timeout; 296 $self->_timeout;
313 297
314 $self->on_drain (delete $self->{on_drain}) if exists $self->{on_drain}; 298 $self->on_drain (delete $self->{on_drain}) if exists $self->{on_drain};
… …
495 my ($self, $cb) = @_; 479 my ($self, $cb) = @_;
496 480
497 $self->{on_drain} = $cb; 481 $self->{on_drain} = $cb;
498 482
499 $cb->($self) 483 $cb->($self)
500 if $cb && $self->{low_water_mark} >= length $self->{wbuf}; 484 if $cb && $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf});
501} 485}
502 486
503=item $handle->push_write ($data) 487=item $handle->push_write ($data)
504 488
505Queues the given scalar to be written. You can push as much data as you 489Queues the given scalar to be written. You can push as much data as you
… …
522 substr $self->{wbuf}, 0, $len, ""; 506 substr $self->{wbuf}, 0, $len, "";
523 507
524 $self->{_activity} = AnyEvent->now; 508 $self->{_activity} = AnyEvent->now;
525 509
526 $self->{on_drain}($self) 510 $self->{on_drain}($self)
527 if $self->{low_water_mark} >= length $self->{wbuf} 511 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf})
528 && $self->{on_drain}; 512 && $self->{on_drain};
529 513
530 delete $self->{_ww} unless length $self->{wbuf}; 514 delete $self->{_ww} unless length $self->{wbuf};
531 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 515 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) {
532 $self->_error ($!, 1); 516 $self->_error ($!, 1);
… …
556 540
557 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write") 541 @_ = ($WH{$type} or Carp::croak "unsupported type passed to AnyEvent::Handle::push_write")
558 ->($self, @_); 542 ->($self, @_);
559 } 543 }
560 544
561 if ($self->{filter_w}) { 545 if ($self->{tls}) {
562 $self->{filter_w}($self, \$_[0]); 546 $self->{_tls_wbuf} .= $_[0];
547
548 &_dotls ($self);
563 } else { 549 } else {
564 $self->{wbuf} .= $_[0]; 550 $self->{wbuf} .= $_[0];
565 $self->_drain_wbuf; 551 $self->_drain_wbuf;
566 } 552 }
567} 553}
… …
584=cut 570=cut
585 571
586register_write_type netstring => sub { 572register_write_type netstring => sub {
587 my ($self, $string) = @_; 573 my ($self, $string) = @_;
588 574
589 sprintf "%d:%s,", (length $string), $string 575 (length $string) . ":$string,"
590}; 576};
591 577
592=item packstring => $format, $data 578=item packstring => $format, $data
593 579
594An octet string prefixed with an encoded length. The encoding C<$format> 580An octet string prefixed with an encoded length. The encoding C<$format>
… …
803 789
804 last; # more data might arrive 790 last; # more data might arrive
805 } 791 }
806 } else { 792 } else {
807 # read side becomes idle 793 # read side becomes idle
808 delete $self->{_rw}; 794 delete $self->{_rw} unless $self->{tls};
809 last; 795 last;
810 } 796 }
811 } 797 }
812 798
813 if ($self->{_eof}) { 799 if ($self->{_eof}) {
… …
1108An octet string prefixed with an encoded length. The encoding C<$format> 1094An 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 1095uses 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 1096integer only (only one of C<cCsSlLqQiInNvVjJw> is allowed, plus an
1111optional C<!>, C<< < >> or C<< > >> modifier). 1097optional C<!>, C<< < >> or C<< > >> modifier).
1112 1098
1113DNS over TCP uses a prefix of C<n>, EPP uses a prefix of C<N>. 1099For example, DNS over TCP uses a prefix of C<n> (2 octet network order),
1100EPP uses a prefix of C<N> (4 octtes).
1114 1101
1115Example: read a block of data prefixed by its length in BER-encoded 1102Example: read a block of data prefixed by its length in BER-encoded
1116format (very efficient). 1103format (very efficient).
1117 1104
1118 $handle->push_read (packstring => "w", sub { 1105 $handle->push_read (packstring => "w", sub {
… …
1271Note that AnyEvent::Handle will automatically C<start_read> for you when 1258Note 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 1259you 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 1260will automatically C<stop_read> for you when neither C<on_read> is set nor
1274there are any read requests in the queue. 1261there are any read requests in the queue.
1275 1262
1263These methods will have no effect when in TLS mode (as TLS doesn't support
1264half-duplex connections).
1265
1276=cut 1266=cut
1277 1267
1278sub stop_read { 1268sub stop_read {
1279 my ($self) = @_; 1269 my ($self) = @_;
1280 1270
1281 delete $self->{_rw}; 1271 delete $self->{_rw} unless $self->{tls};
1282} 1272}
1283 1273
1284sub start_read { 1274sub start_read {
1285 my ($self) = @_; 1275 my ($self) = @_;
1286 1276
1287 unless ($self->{_rw} || $self->{_eof}) { 1277 unless ($self->{_rw} || $self->{_eof}) {
1288 Scalar::Util::weaken $self; 1278 Scalar::Util::weaken $self;
1289 1279
1290 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub { 1280 $self->{_rw} = AnyEvent->io (fh => $self->{fh}, poll => "r", cb => sub {
1291 my $rbuf = $self->{filter_r} ? \my $buf : \$self->{rbuf}; 1281 my $rbuf = \($self->{tls} ? my $buf : $self->{rbuf});
1292 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf; 1282 my $len = sysread $self->{fh}, $$rbuf, $self->{read_size} || 8192, length $$rbuf;
1293 1283
1294 if ($len > 0) { 1284 if ($len > 0) {
1295 $self->{_activity} = AnyEvent->now; 1285 $self->{_activity} = AnyEvent->now;
1296 1286
1297 $self->{filter_r} 1287 if ($self->{tls}) {
1298 ? $self->{filter_r}($self, $rbuf) 1288 Net::SSLeay::BIO_write ($self->{_rbio}, $$rbuf);
1299 : $self->{_in_drain} || $self->_drain_rbuf; 1289
1290 &_dotls ($self);
1291 } else {
1292 $self->_drain_rbuf unless $self->{_in_drain};
1293 }
1300 1294
1301 } elsif (defined $len) { 1295 } elsif (defined $len) {
1302 delete $self->{_rw}; 1296 delete $self->{_rw};
1303 $self->{_eof} = 1; 1297 $self->{_eof} = 1;
1304 $self->_drain_rbuf unless $self->{_in_drain}; 1298 $self->_drain_rbuf unless $self->{_in_drain};
… …
1308 } 1302 }
1309 }); 1303 });
1310 } 1304 }
1311} 1305}
1312 1306
1307# poll the write BIO and send the data if applicable
1313sub _dotls { 1308sub _dotls {
1314 my ($self) = @_; 1309 my ($self) = @_;
1315 1310
1316 my $buf; 1311 my $tmp;
1317 1312
1318 if (length $self->{_tls_wbuf}) { 1313 if (length $self->{_tls_wbuf}) {
1319 while ((my $len = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { 1314 while (($tmp = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) {
1320 substr $self->{_tls_wbuf}, 0, $len, ""; 1315 substr $self->{_tls_wbuf}, 0, $tmp, "";
1321 } 1316 }
1322 } 1317 }
1323 1318
1324 while (defined ($buf = Net::SSLeay::read ($self->{tls}))) { 1319 while (defined ($tmp = Net::SSLeay::read ($self->{tls}))) {
1325 unless (length $buf) { 1320 unless (length $tmp) {
1326 # let's treat SSL-eof as we treat normal EOF 1321 # let's treat SSL-eof as we treat normal EOF
1327 delete $self->{_rw}; 1322 delete $self->{_rw};
1328 $self->{_eof} = 1; 1323 $self->{_eof} = 1;
1329 &_freetls; 1324 &_freetls;
1330 } 1325 }
1331 1326
1332 $self->{rbuf} .= $buf; 1327 $self->{rbuf} .= $tmp;
1333 $self->_drain_rbuf unless $self->{_in_drain}; 1328 $self->_drain_rbuf unless $self->{_in_drain};
1334 $self->{tls} or return; # tls session might have gone away in callback 1329 $self->{tls} or return; # tls session might have gone away in callback
1335 } 1330 }
1336 1331
1337 my $err = Net::SSLeay::get_error ($self->{tls}, -1); 1332 $tmp = Net::SSLeay::get_error ($self->{tls}, -1);
1338 1333
1339 if ($err!= Net::SSLeay::ERROR_WANT_READ ()) { 1334 if ($tmp != Net::SSLeay::ERROR_WANT_READ ()) {
1340 if ($err == Net::SSLeay::ERROR_SYSCALL ()) { 1335 if ($tmp == Net::SSLeay::ERROR_SYSCALL ()) {
1341 return $self->_error ($!, 1); 1336 return $self->_error ($!, 1);
1342 } elsif ($err == Net::SSLeay::ERROR_SSL ()) { 1337 } elsif ($tmp == Net::SSLeay::ERROR_SSL ()) {
1343 return $self->_error (&Errno::EIO, 1); 1338 return $self->_error (&Errno::EIO, 1);
1344 } 1339 }
1345 1340
1346 # all others are fine for our purposes 1341 # all other errors are fine for our purposes
1347 } 1342 }
1348 1343
1349 if (length ($buf = Net::SSLeay::BIO_read ($self->{_wbio}))) { 1344 while (length ($tmp = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1350 $self->{wbuf} .= $buf; 1345 $self->{wbuf} .= $tmp;
1351 $self->_drain_wbuf; 1346 $self->_drain_wbuf;
1352 } 1347 }
1353} 1348}
1354 1349
1355=item $handle->starttls ($tls[, $tls_ctx]) 1350=item $handle->starttls ($tls[, $tls_ctx])
… …
1373 1368
1374=cut 1369=cut
1375 1370
1376sub starttls { 1371sub starttls {
1377 my ($self, $ssl, $ctx) = @_; 1372 my ($self, $ssl, $ctx) = @_;
1373
1374 require Net::SSLeay;
1378 1375
1379 Carp::croak "it is an error to call starttls more than once on an Anyevent::Handle object" 1376 Carp::croak "it is an error to call starttls more than once on an Anyevent::Handle object"
1380 if $self->{tls}; 1377 if $self->{tls};
1381 1378
1382 if ($ssl eq "accept") { 1379 if ($ssl eq "accept") {
… …
1395 # and mismaintained ssleay-module doesn't even offer them). 1392 # and mismaintained ssleay-module doesn't even offer them).
1396 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html 1393 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
1397 # 1394 #
1398 # in short: this is a mess. 1395 # in short: this is a mess.
1399 # 1396 #
1400 # note that we do not try to kepe the length constant between writes as we are required to do. 1397 # note that we do not try to keep the length constant between writes as we are required to do.
1401 # we assume that most (but not all) of this insanity only applies to non-blocking cases, 1398 # we assume that most (but not all) of this insanity only applies to non-blocking cases,
1402 # and we drive openssl fully in blocking mode here. 1399 # and we drive openssl fully in blocking mode here. Or maybe we don't - openssl seems to
1400 # have identity issues in that area.
1403 Net::SSLeay::CTX_set_mode ($self->{tls}, 1401 Net::SSLeay::CTX_set_mode ($self->{tls},
1404 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1) 1402 (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
1405 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2)); 1403 | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
1406 1404
1407 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1405 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1408 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 1406 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
1409 1407
1410 Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio}); 1408 Net::SSLeay::set_bio ($ssl, $self->{_rbio}, $self->{_wbio});
1411 1409
1412 $self->{filter_w} = sub {
1413 $_[0]{_tls_wbuf} .= ${$_[1]};
1414 &_dotls;
1415 };
1416 $self->{filter_r} = sub {
1417 Net::SSLeay::BIO_write ($_[0]{_rbio}, ${$_[1]});
1418 &_dotls;
1419 };
1420
1421 &_dotls; # need to trigger the initial negotiation exchange 1410 &_dotls; # need to trigger the initial handshake
1411 $self->start_read; # make sure we actually do read
1422} 1412}
1423 1413
1424=item $handle->stoptls 1414=item $handle->stoptls
1425 1415
1426Shuts down the SSL connection - this makes a proper EOF handshake by 1416Shuts down the SSL connection - this makes a proper EOF handshake by
… …
1432 1422
1433sub stoptls { 1423sub stoptls {
1434 my ($self) = @_; 1424 my ($self) = @_;
1435 1425
1436 if ($self->{tls}) { 1426 if ($self->{tls}) {
1437 Net::SSLeay::shutdown $self->{tls}; 1427 Net::SSLeay::shutdown ($self->{tls});
1438 1428
1439 &_dotls; 1429 &_dotls;
1440 1430
1441 # we don't give a shit. no, we do, but we can't. no... 1431 # we don't give a shit. no, we do, but we can't. no...
1442 # we, we... have to use openssl :/ 1432 # we, we... have to use openssl :/
… …
1449 1439
1450 return unless $self->{tls}; 1440 return unless $self->{tls};
1451 1441
1452 Net::SSLeay::free (delete $self->{tls}); 1442 Net::SSLeay::free (delete $self->{tls});
1453 1443
1454 delete @$self{qw(_rbio filter_w _wbio filter_r)}; 1444 delete @$self{qw(_rbio _wbio _tls_wbuf)};
1455} 1445}
1456 1446
1457sub DESTROY { 1447sub DESTROY {
1458 my $self = shift; 1448 my $self = shift;
1459 1449
… …
1480 @linger = (); 1470 @linger = ();
1481 }); 1471 });
1482 } 1472 }
1483} 1473}
1484 1474
1475=item $handle->destroy
1476
1477Shut's down the handle object as much as possible - this call ensures that
1478no further callbacks will be invoked and resources will be freed as much
1479as possible. You must not call any methods on the object afterwards.
1480
1481The handle might still linger in the background and write out remaining
1482data, as specified by the C<linger> option, however.
1483
1484=cut
1485
1486sub destroy {
1487 my ($self) = @_;
1488
1489 $self->DESTROY;
1490 %$self = ();
1491}
1492
1485=item AnyEvent::Handle::TLS_CTX 1493=item AnyEvent::Handle::TLS_CTX
1486 1494
1487This function creates and returns the Net::SSLeay::CTX object used by 1495This function creates and returns the Net::SSLeay::CTX object used by
1488default for TLS mode. 1496default for TLS mode.
1489 1497
… …
1517 } 1525 }
1518} 1526}
1519 1527
1520=back 1528=back
1521 1529
1530
1531=head1 NONFREQUENTLY ASKED QUESTIONS
1532
1533=over 4
1534
1535=item How do I read data until the other side closes the connection?
1536
1537If you just want to read your data into a perl scalar, the easiest way
1538to achieve this is by setting an C<on_read> callback that does nothing,
1539clearing the C<on_eof> callback and in the C<on_error> callback, the data
1540will be in C<$_[0]{rbuf}>:
1541
1542 $handle->on_read (sub { });
1543 $handle->on_eof (undef);
1544 $handle->on_error (sub {
1545 my $data = delete $_[0]{rbuf};
1546 undef $handle;
1547 });
1548
1549The reason to use C<on_error> is that TCP connections, due to latencies
1550and packets loss, might get closed quite violently with an error, when in
1551fact, all data has been received.
1552
1553It is usually better to use acknowledgements when transfering data,
1554to make sure the other side hasn't just died and you got the data
1555intact. This is also one reason why so many internet protocols have an
1556explicit QUIT command.
1557
1558
1559=item I don't want to destroy the handle too early - how do I wait until
1560all data has been written?
1561
1562After writing your last bits of data, set the C<on_drain> callback
1563and destroy the handle in there - with the default setting of
1564C<low_water_mark> this will be called precisely when all data has been
1565written to the socket:
1566
1567 $handle->push_write (...);
1568 $handle->on_drain (sub {
1569 warn "all data submitted to the kernel\n";
1570 undef $handle;
1571 });
1572
1573=item I get different callback invocations in TLS mode/Why can't I pause
1574reading?
1575
1576Unlike, say, TCP, TLS conenctions do not consist of two independent
1577communication channels, one for each direction. Or put differently. the
1578read and write directions are not independent of each other: you cannot
1579write data unless you are also prepared to read, and vice versa.
1580
1581This can mean than, in TLS mode, you might get C<on_error> or C<on_eof>
1582callback invocations when you are not expecting any read data - the reason
1583is that AnyEvent::Handle always reads in TLS mode.
1584
1585During the connection, you have to make sure that you always have a
1586non-empty read-queue, or an C<on_read> watcher. At the end of the
1587connection (or when you no longer want to use it) you can call the
1588C<destroy> method.
1589
1590=back
1591
1592
1522=head1 SUBCLASSING AnyEvent::Handle 1593=head1 SUBCLASSING AnyEvent::Handle
1523 1594
1524In many cases, you might want to subclass AnyEvent::Handle. 1595In many cases, you might want to subclass AnyEvent::Handle.
1525 1596
1526To make this easier, a given version of AnyEvent::Handle uses these 1597To make this easier, a given version of AnyEvent::Handle uses these

Diff Legend

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