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.235 by root, Tue May 8 19:41:22 2012 UTC vs.
Revision 1.255 by root, Wed Jul 29 13:02:48 2020 UTC

30 30
31 $cv->recv; 31 $cv->recv;
32 32
33=head1 DESCRIPTION 33=head1 DESCRIPTION
34 34
35This is a helper module to make it easier to do event-based I/O on 35This is a helper module to make it easier to do event-based I/O
36stream-based filehandles (sockets, pipes, and other stream things). 36on stream-based filehandles (sockets, pipes, and other stream
37things). Specifically, it doesn't work as expected on files, packet-based
38sockets or similar things.
37 39
38The L<AnyEvent::Intro> tutorial contains some well-documented 40The L<AnyEvent::Intro> tutorial contains some well-documented
39AnyEvent::Handle examples. 41AnyEvent::Handle examples.
40 42
41In the following, where the documentation refers to "bytes", it means 43In the following, where the documentation refers to "bytes", it means
53package AnyEvent::Handle; 55package AnyEvent::Handle;
54 56
55use Scalar::Util (); 57use Scalar::Util ();
56use List::Util (); 58use List::Util ();
57use Carp (); 59use Carp ();
58use Errno qw(EAGAIN EINTR); 60use Errno qw(EAGAIN EWOULDBLOCK EINTR);
59 61
60use AnyEvent (); BEGIN { AnyEvent::common_sense } 62use AnyEvent (); BEGIN { AnyEvent::common_sense }
61use AnyEvent::Util qw(WSAEWOULDBLOCK); 63use AnyEvent::Util qw(WSAEWOULDBLOCK);
62 64
63our $VERSION = $AnyEvent::VERSION; 65our $VERSION = $AnyEvent::VERSION;
91 93
92=item fh => $filehandle [C<fh> or C<connect> MANDATORY] 94=item fh => $filehandle [C<fh> or C<connect> MANDATORY]
93 95
94The filehandle this L<AnyEvent::Handle> object will operate on. 96The filehandle this L<AnyEvent::Handle> object will operate on.
95NOTE: The filehandle will be set to non-blocking mode (using 97NOTE: The filehandle will be set to non-blocking mode (using
96C<AnyEvent::Util::fh_nonblocking>) by the constructor and needs to stay in 98C<AnyEvent::fh_unblock>) by the constructor and needs to stay in
97that mode. 99that mode.
98 100
99=item connect => [$host, $service] [C<fh> or C<connect> MANDATORY] 101=item connect => [$host, $service] [C<fh> or C<connect> MANDATORY]
100 102
101Try to connect to the specified host and service (port), using 103Try to connect to the specified host and service (port), using
131 133
132The peer's numeric host and port (the socket peername) are passed as 134The peer's numeric host and port (the socket peername) are passed as
133parameters, together with a retry callback. At the time it is called the 135parameters, together with a retry callback. At the time it is called the
134read and write queues, EOF status, TLS status and similar properties of 136read and write queues, EOF status, TLS status and similar properties of
135the handle will have been reset. 137the handle will have been reset.
136
137It is not allowed to use the read or write queues while the handle object
138is connecting.
139 138
140If, for some reason, the handle is not acceptable, calling C<$retry> will 139If, for some reason, the handle is not acceptable, calling C<$retry> will
141continue with the next connection target (in case of multi-homed hosts or 140continue with the next connection target (in case of multi-homed hosts or
142SRV records there can be multiple connection endpoints). The C<$retry> 141SRV records there can be multiple connection endpoints). The C<$retry>
143callback can be invoked after the connect callback returns, i.e. one can 142callback can be invoked after the connect callback returns, i.e. one can
496callback. 495callback.
497 496
498This callback will only be called on TLS shutdowns, not when the 497This callback will only be called on TLS shutdowns, not when the
499underlying handle signals EOF. 498underlying handle signals EOF.
500 499
501=item json => JSON or JSON::XS object 500=item json => L<JSON>, L<JSON::PP> or L<JSON::XS> object
502 501
503This is the json coder object used by the C<json> read and write types. 502This is the json coder object used by the C<json> read and write types.
504 503
505If you don't supply it, then AnyEvent::Handle will create and use a 504If you don't supply it, then AnyEvent::Handle will create and use a
506suitable one (on demand), which will write and expect UTF-8 encoded JSON 505suitable one (on demand), which will write and expect UTF-8 encoded
506JSON texts (either using L<JSON::XS> or L<JSON>). The written texts are
507guaranteed not to contain any newline character.
508
509For security reasons, this encoder will likely I<not> handle numbers and
510strings, only arrays and objects/hashes. The reason is that originally
511JSON was self-delimited, but Dougles Crockford thought it was a splendid
512idea to redefine JSON incompatibly, so this is no longer true.
513
514For protocols that used back-to-back JSON texts, this might lead to
515run-ins, where two or more JSON texts will be interpreted as one JSON
507texts. 516text.
508 517
518For this reason, if the default encoder uses L<JSON::XS>, it will default
519to not allowing anything but arrays and objects/hashes, at least for the
520forseeable future (it will change at some point). This might or might not
521be true for the L<JSON> module, so this might cause a security issue.
522
523If you depend on either behaviour, you should create your own json object
524and pass it in explicitly.
525
526=item cbor => L<CBOR::XS> object
527
528This is the cbor coder object used by the C<cbor> read and write types.
529
530If you don't supply it, then AnyEvent::Handle will create and use a
531suitable one (on demand), which will write CBOR without using extensions,
532if possible.
533
509Note that you are responsible to depend on the JSON module if you want to 534Note that you are responsible to depend on the L<CBOR::XS> module if you
510use this functionality, as AnyEvent does not have a dependency itself. 535want to use this functionality, as AnyEvent does not have a dependency on
536it itself.
511 537
512=back 538=back
513 539
514=cut 540=cut
515 541
557 } else { 583 } else {
558 if ($self->{on_connect_error}) { 584 if ($self->{on_connect_error}) {
559 $self->{on_connect_error}($self, "$!"); 585 $self->{on_connect_error}($self, "$!");
560 $self->destroy if $self; 586 $self->destroy if $self;
561 } else { 587 } else {
562 $self->_error ($!, 1); 588 $self->error ($!, 1);
563 } 589 }
564 } 590 }
565 }, 591 },
566 sub { 592 sub {
567 local $self->{fh} = $_[0]; 593 local $self->{fh} = $_[0];
587 # with AnyEvent::Handle, do them a favour. 613 # with AnyEvent::Handle, do them a favour.
588 my $type = getsockopt $self->{fh}, Socket::SOL_SOCKET (), Socket::SO_TYPE (); 614 my $type = getsockopt $self->{fh}, Socket::SOL_SOCKET (), Socket::SO_TYPE ();
589 Carp::croak "AnyEvent::Handle: only stream sockets supported, anything else will NOT work!" 615 Carp::croak "AnyEvent::Handle: only stream sockets supported, anything else will NOT work!"
590 if Socket::SOCK_STREAM () != (unpack "I", $type) && defined $type; 616 if Socket::SOCK_STREAM () != (unpack "I", $type) && defined $type;
591 617
592 AnyEvent::Util::fh_nonblocking $self->{fh}, 1; 618 AnyEvent::fh_unblock $self->{fh};
593 619
594 $self->{_activity} = 620 $self->{_activity} =
595 $self->{_ractivity} = 621 $self->{_ractivity} =
596 $self->{_wactivity} = AE::now; 622 $self->{_wactivity} = AE::now;
597 623
617 if $self->{on_read} || @{ $self->{_queue} }; 643 if $self->{on_read} || @{ $self->{_queue} };
618 644
619 $self->_drain_wbuf; 645 $self->_drain_wbuf;
620} 646}
621 647
648=item $handle->error ($errno[, $fatal[, $message]])
649
650Generates an error event, just like AnyEvent::Handle itself would do, i.e.
651calls the C<on_error> callback.
652
653Te only rerquired parameter is C<$errno>, which sets C<$!>. C<$fatal>
654defaults to false and C<$message> defaults to the stringified version
655of C<$1>.
656
657Example: generate C<EIO> when you read unexpected data.
658
659 $handle->push_read (line => sub {
660 $_[1] eq "hello"
661 or return $handle->error (Errno::EIO);
662 });
663
664=cut
665
622sub _error { 666sub error {
623 my ($self, $errno, $fatal, $message) = @_; 667 my ($self, $errno, $fatal, $message) = @_;
624 668
625 $! = $errno; 669 $! = $errno;
626 $message ||= "$!"; 670 $message ||= "$!";
627 671
733 setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_OOBINLINE (), int $_[1] 777 setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_OOBINLINE (), int $_[1]
734 if $_[0]{fh}; 778 if $_[0]{fh};
735 }; 779 };
736} 780}
737 781
738=item $handle->keepalive ($boolean)
739
740Enables or disables the C<keepalive> setting (see constructor argument of
741the same name for details).
742
743=cut
744
745sub keepalive {
746 $_[0]{keepalive} = $_[1];
747
748 eval {
749 local $SIG{__DIE__};
750 setsockopt $_[0]{fh}, Socket::SOL_SOCKET (), Socket::SO_KEEPALIVE (), int $_[1]
751 if $_[0]{fh};
752 };
753}
754
755=item $handle->on_starttls ($cb) 782=item $handle->on_starttls ($cb)
756 783
757Replace the current C<on_starttls> callback (see the C<on_starttls> constructor argument). 784Replace the current C<on_starttls> callback (see the C<on_starttls> constructor argument).
758 785
759=cut 786=cut
857 $self->{$activity} = $NOW; 884 $self->{$activity} = $NOW;
858 885
859 if ($self->{$on_timeout}) { 886 if ($self->{$on_timeout}) {
860 $self->{$on_timeout}($self); 887 $self->{$on_timeout}($self);
861 } else { 888 } else {
862 $self->_error (Errno::ETIMEDOUT); 889 $self->error (Errno::ETIMEDOUT);
863 } 890 }
864 891
865 # callback could have changed timeout value, optimise 892 # callback could have changed timeout value, optimise
866 return unless $self->{$timeout}; 893 return unless $self->{$timeout};
867 894
947 $self->{on_drain}($self) 974 $self->{on_drain}($self)
948 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf}) 975 if $self->{low_water_mark} >= (length $self->{wbuf}) + (length $self->{_tls_wbuf})
949 && $self->{on_drain}; 976 && $self->{on_drain};
950 977
951 delete $self->{_ww} unless length $self->{wbuf}; 978 delete $self->{_ww} unless length $self->{wbuf};
952 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 979 } elsif ($! != EAGAIN && $! != EINTR && $! != EWOULDBLOCK && $! != WSAEWOULDBLOCK) {
953 $self->_error ($!, 1); 980 $self->error ($!, 1);
954 } 981 }
955 }; 982 };
956 983
957 # try to write data immediately 984 # try to write data immediately
958 $cb->() unless $self->{autocork}; 985 $cb->() unless $self->{autocork};
963 990
964 if ( 991 if (
965 defined $self->{wbuf_max} 992 defined $self->{wbuf_max}
966 && $self->{wbuf_max} < length $self->{wbuf} 993 && $self->{wbuf_max} < length $self->{wbuf}
967 ) { 994 ) {
968 $self->_error (Errno::ENOSPC, 1), return; 995 $self->error (Errno::ENOSPC, 1), return;
969 } 996 }
970 }; 997 };
971} 998}
972 999
973our %WH; 1000our %WH;
1045 1072
1046Encodes the given hash or array reference into a JSON object. Unless you 1073Encodes the given hash or array reference into a JSON object. Unless you
1047provide your own JSON object, this means it will be encoded to JSON text 1074provide your own JSON object, this means it will be encoded to JSON text
1048in UTF-8. 1075in UTF-8.
1049 1076
1077The default encoder might or might not handle every type of JSON value -
1078it might be limited to arrays and objects for security reasons. See the
1079C<json> constructor attribute for more details.
1080
1050JSON objects (and arrays) are self-delimiting, so you can write JSON at 1081JSON objects (and arrays) are self-delimiting, so if you only use arrays
1051one end of a handle and read them at the other end without using any 1082and hashes, you can write JSON at one end of a handle and read them at the
1052additional framing. 1083other end without using any additional framing.
1053 1084
1054The generated JSON text is guaranteed not to contain any newlines: While 1085The JSON text generated by the default encoder is guaranteed not to
1055this module doesn't need delimiters after or between JSON texts to be 1086contain any newlines: While this module doesn't need delimiters after or
1056able to read them, many other languages depend on that. 1087between JSON texts to be able to read them, many other languages depend on
1088them.
1057 1089
1058A simple RPC protocol that interoperates easily with others is to send 1090A simple RPC protocol that interoperates easily with other languages is
1059JSON arrays (or objects, although arrays are usually the better choice as 1091to send JSON arrays (or objects, although arrays are usually the better
1060they mimic how function argument passing works) and a newline after each 1092choice as they mimic how function argument passing works) and a newline
1061JSON text: 1093after each JSON text:
1062 1094
1063 $handle->push_write (json => ["method", "arg1", "arg2"]); # whatever 1095 $handle->push_write (json => ["method", "arg1", "arg2"]); # whatever
1064 $handle->push_write ("\012"); 1096 $handle->push_write ("\012");
1065 1097
1066An AnyEvent::Handle receiver would simply use the C<json> read type and 1098An AnyEvent::Handle receiver would simply use the C<json> read type and
1069 $handle->push_read (json => sub { my $array = $_[1]; ... }); 1101 $handle->push_read (json => sub { my $array = $_[1]; ... });
1070 1102
1071Other languages could read single lines terminated by a newline and pass 1103Other languages could read single lines terminated by a newline and pass
1072this line into their JSON decoder of choice. 1104this line into their JSON decoder of choice.
1073 1105
1106=item cbor => $perl_scalar
1107
1108Encodes the given scalar into a CBOR value. Unless you provide your own
1109L<CBOR::XS> object, this means it will be encoded to a CBOR string not
1110using any extensions, if possible.
1111
1112CBOR values are self-delimiting, so you can write CBOR at one end of
1113a handle and read them at the other end without using any additional
1114framing.
1115
1116A simple nd very very fast RPC protocol that interoperates with
1117other languages is to send CBOR and receive CBOR values (arrays are
1118recommended):
1119
1120 $handle->push_write (cbor => ["method", "arg1", "arg2"]); # whatever
1121
1122An AnyEvent::Handle receiver would simply use the C<cbor> read type:
1123
1124 $handle->push_read (cbor => sub { my $array = $_[1]; ... });
1125
1074=cut 1126=cut
1075 1127
1076sub json_coder() { 1128sub json_coder() {
1077 eval { require JSON::XS; JSON::XS->new->utf8 } 1129 eval { require JSON::XS; JSON::XS->new->utf8 }
1078 || do { require JSON; JSON->new->utf8 } 1130 || do { require JSON::PP; JSON::PP->new->utf8 }
1079} 1131}
1080 1132
1081register_write_type json => sub { 1133register_write_type json => sub {
1082 my ($self, $ref) = @_; 1134 my ($self, $ref) = @_;
1083 1135
1084 my $json = $self->{json} ||= json_coder; 1136 ($self->{json} ||= json_coder)
1085
1086 $json->encode ($ref) 1137 ->encode ($ref)
1138};
1139
1140sub cbor_coder() {
1141 require CBOR::XS;
1142 CBOR::XS->new
1143}
1144
1145register_write_type cbor => sub {
1146 my ($self, $scalar) = @_;
1147
1148 ($self->{cbor} ||= cbor_coder)
1149 ->encode ($scalar)
1087}; 1150};
1088 1151
1089=item storable => $reference 1152=item storable => $reference
1090 1153
1091Freezes the given reference using L<Storable> and writes it to the 1154Freezes the given reference using L<Storable> and writes it to the
1268 1331
1269 if (my $cb = shift @{ $self->{_queue} }) { 1332 if (my $cb = shift @{ $self->{_queue} }) {
1270 unless ($cb->($self)) { 1333 unless ($cb->($self)) {
1271 # no progress can be made 1334 # no progress can be made
1272 # (not enough data and no data forthcoming) 1335 # (not enough data and no data forthcoming)
1273 $self->_error (Errno::EPIPE, 1), return 1336 $self->error (Errno::EPIPE, 1), return
1274 if $self->{_eof}; 1337 if $self->{_eof};
1275 1338
1276 unshift @{ $self->{_queue} }, $cb; 1339 unshift @{ $self->{_queue} }, $cb;
1277 last; 1340 last;
1278 } 1341 }
1286 && !@{ $self->{_queue} } # and the queue is still empty 1349 && !@{ $self->{_queue} } # and the queue is still empty
1287 && $self->{on_read} # but we still have on_read 1350 && $self->{on_read} # but we still have on_read
1288 ) { 1351 ) {
1289 # no further data will arrive 1352 # no further data will arrive
1290 # so no progress can be made 1353 # so no progress can be made
1291 $self->_error (Errno::EPIPE, 1), return 1354 $self->error (Errno::EPIPE, 1), return
1292 if $self->{_eof}; 1355 if $self->{_eof};
1293 1356
1294 last; # more data might arrive 1357 last; # more data might arrive
1295 } 1358 }
1296 } else { 1359 } else {
1301 } 1364 }
1302 1365
1303 if ($self->{_eof}) { 1366 if ($self->{_eof}) {
1304 $self->{on_eof} 1367 $self->{on_eof}
1305 ? $self->{on_eof}($self) 1368 ? $self->{on_eof}($self)
1306 : $self->_error (0, 1, "Unexpected end-of-file"); 1369 : $self->error (0, 1, "Unexpected end-of-file");
1307 1370
1308 return; 1371 return;
1309 } 1372 }
1310 1373
1311 if ( 1374 if (
1312 defined $self->{rbuf_max} 1375 defined $self->{rbuf_max}
1313 && $self->{rbuf_max} < length $self->{rbuf} 1376 && $self->{rbuf_max} < length $self->{rbuf}
1314 ) { 1377 ) {
1315 $self->_error (Errno::ENOSPC, 1), return; 1378 $self->error (Errno::ENOSPC, 1), return;
1316 } 1379 }
1317 1380
1318 # may need to restart read watcher 1381 # may need to restart read watcher
1319 unless ($self->{_rw}) { 1382 unless ($self->{_rw}) {
1320 $self->start_read 1383 $self->start_read
1485 1548
1486register_read_type line => sub { 1549register_read_type line => sub {
1487 my ($self, $cb, $eol) = @_; 1550 my ($self, $cb, $eol) = @_;
1488 1551
1489 if (@_ < 3) { 1552 if (@_ < 3) {
1490 # this is more than twice as fast as the generic code below 1553 # this is faster then the generic code below
1491 sub { 1554 sub {
1492 $_[0]{rbuf} =~ s/^([^\015\012]*)(\015?\012)// or return; 1555 (my $pos = index $_[0]{rbuf}, "\012") >= 0
1556 or return;
1493 1557
1558 (my $str = substr $_[0]{rbuf}, 0, $pos + 1, "") =~ s/(\015?\012)\Z// or die;
1494 $cb->($_[0], "$1", "$2"); 1559 $cb->($_[0], $str, "$1");
1495 1 1560 1
1496 } 1561 }
1497 } else { 1562 } else {
1498 $eol = quotemeta $eol unless ref $eol; 1563 $eol = quotemeta $eol unless ref $eol;
1499 $eol = qr|^(.*?)($eol)|s; 1564 $eol = qr|^(.*?)($eol)|s;
1508}; 1573};
1509 1574
1510=item regex => $accept[, $reject[, $skip], $cb->($handle, $data) 1575=item regex => $accept[, $reject[, $skip], $cb->($handle, $data)
1511 1576
1512Makes a regex match against the regex object C<$accept> and returns 1577Makes a regex match against the regex object C<$accept> and returns
1513everything up to and including the match. 1578everything up to and including the match. All the usual regex variables
1579($1, %+ etc.) from the regex match are available in the callback.
1514 1580
1515Example: read a single line terminated by '\n'. 1581Example: read a single line terminated by '\n'.
1516 1582
1517 $handle->push_read (regex => qr<\n>, sub { ... }); 1583 $handle->push_read (regex => qr<\n>, sub { ... });
1518 1584
1563 return 1; 1629 return 1;
1564 } 1630 }
1565 1631
1566 # reject 1632 # reject
1567 if ($reject && $$rbuf =~ $reject) { 1633 if ($reject && $$rbuf =~ $reject) {
1568 $_[0]->_error (Errno::EBADMSG); 1634 $_[0]->error (Errno::EBADMSG);
1569 } 1635 }
1570 1636
1571 # skip 1637 # skip
1572 if ($skip && $$rbuf =~ $skip) { 1638 if ($skip && $$rbuf =~ $skip) {
1573 $data .= substr $$rbuf, 0, $+[0], ""; 1639 $data .= substr $$rbuf, 0, $+[0], "";
1589 my ($self, $cb) = @_; 1655 my ($self, $cb) = @_;
1590 1656
1591 sub { 1657 sub {
1592 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) { 1658 unless ($_[0]{rbuf} =~ s/^(0|[1-9][0-9]*)://) {
1593 if ($_[0]{rbuf} =~ /[^0-9]/) { 1659 if ($_[0]{rbuf} =~ /[^0-9]/) {
1594 $_[0]->_error (Errno::EBADMSG); 1660 $_[0]->error (Errno::EBADMSG);
1595 } 1661 }
1596 return; 1662 return;
1597 } 1663 }
1598 1664
1599 my $len = $1; 1665 my $len = $1;
1602 my $string = $_[1]; 1668 my $string = $_[1];
1603 $_[0]->unshift_read (chunk => 1, sub { 1669 $_[0]->unshift_read (chunk => 1, sub {
1604 if ($_[1] eq ",") { 1670 if ($_[1] eq ",") {
1605 $cb->($_[0], $string); 1671 $cb->($_[0], $string);
1606 } else { 1672 } else {
1607 $_[0]->_error (Errno::EBADMSG); 1673 $_[0]->error (Errno::EBADMSG);
1608 } 1674 }
1609 }); 1675 });
1610 }); 1676 });
1611 1677
1612 1 1678 1
1662=item json => $cb->($handle, $hash_or_arrayref) 1728=item json => $cb->($handle, $hash_or_arrayref)
1663 1729
1664Reads a JSON object or array, decodes it and passes it to the 1730Reads a JSON object or array, decodes it and passes it to the
1665callback. When a parse error occurs, an C<EBADMSG> error will be raised. 1731callback. When a parse error occurs, an C<EBADMSG> error will be raised.
1666 1732
1667If a C<json> object was passed to the constructor, then that will be used 1733If a C<json> object was passed to the constructor, then that will be
1668for the final decode, otherwise it will create a JSON coder expecting UTF-8. 1734used for the final decode, otherwise it will create a L<JSON::XS> or
1735L<JSON::PP> coder object expecting UTF-8.
1669 1736
1670This read type uses the incremental parser available with JSON version 1737This read type uses the incremental parser available with JSON version
16712.09 (and JSON::XS version 2.2) and above. You have to provide a 17382.09 (and JSON::XS version 2.2) and above.
1672dependency on your own: this module will load the JSON module, but
1673AnyEvent does not depend on it itself.
1674 1739
1675Since JSON texts are fully self-delimiting, the C<json> read and write 1740Since JSON texts are fully self-delimiting, the C<json> read and write
1676types are an ideal simple RPC protocol: just exchange JSON datagrams. See 1741types are an ideal simple RPC protocol: just exchange JSON datagrams. See
1677the C<json> write type description, above, for an actual example. 1742the C<json> write type description, above, for an actual example.
1678 1743
1682 my ($self, $cb) = @_; 1747 my ($self, $cb) = @_;
1683 1748
1684 my $json = $self->{json} ||= json_coder; 1749 my $json = $self->{json} ||= json_coder;
1685 1750
1686 my $data; 1751 my $data;
1687 my $rbuf = \$self->{rbuf};
1688 1752
1689 sub { 1753 sub {
1690 my $ref = eval { $json->incr_parse ($_[0]{rbuf}) }; 1754 my $ref = eval { $json->incr_parse ($_[0]{rbuf}) };
1691 1755
1692 if ($ref) { 1756 if ($ref) {
1700 $json->incr_skip; 1764 $json->incr_skip;
1701 1765
1702 $_[0]{rbuf} = $json->incr_text; 1766 $_[0]{rbuf} = $json->incr_text;
1703 $json->incr_text = ""; 1767 $json->incr_text = "";
1704 1768
1705 $_[0]->_error (Errno::EBADMSG); 1769 $_[0]->error (Errno::EBADMSG);
1706 1770
1707 () 1771 ()
1708 } else { 1772 } else {
1709 $_[0]{rbuf} = ""; 1773 $_[0]{rbuf} = "";
1710 1774
1775 ()
1776 }
1777 }
1778};
1779
1780=item cbor => $cb->($handle, $scalar)
1781
1782Reads a CBOR value, decodes it and passes it to the callback. When a parse
1783error occurs, an C<EBADMSG> error will be raised.
1784
1785If a L<CBOR::XS> object was passed to the constructor, then that will be
1786used for the final decode, otherwise it will create a CBOR coder without
1787enabling any options.
1788
1789You have to provide a dependency to L<CBOR::XS> on your own: this module
1790will load the L<CBOR::XS> module, but AnyEvent does not depend on it
1791itself.
1792
1793Since CBOR values are fully self-delimiting, the C<cbor> read and write
1794types are an ideal simple RPC protocol: just exchange CBOR datagrams. See
1795the C<cbor> write type description, above, for an actual example.
1796
1797=cut
1798
1799register_read_type cbor => sub {
1800 my ($self, $cb) = @_;
1801
1802 my $cbor = $self->{cbor} ||= cbor_coder;
1803
1804 my $data;
1805
1806 sub {
1807 my (@value) = eval { $cbor->incr_parse ($_[0]{rbuf}) };
1808
1809 if (@value) {
1810 $cb->($_[0], @value);
1811
1812 1
1813 } elsif ($@) {
1814 # error case
1815 $cbor->incr_reset;
1816
1817 $_[0]->error (Errno::EBADMSG);
1818
1819 ()
1820 } else {
1711 () 1821 ()
1712 } 1822 }
1713 } 1823 }
1714}; 1824};
1715 1825
1739 if ($format + $len <= length $_[0]{rbuf}) { 1849 if ($format + $len <= length $_[0]{rbuf}) {
1740 my $data = substr $_[0]{rbuf}, $format, $len; 1850 my $data = substr $_[0]{rbuf}, $format, $len;
1741 substr $_[0]{rbuf}, 0, $format + $len, ""; 1851 substr $_[0]{rbuf}, 0, $format + $len, "";
1742 1852
1743 eval { $cb->($_[0], Storable::thaw ($data)); 1 } 1853 eval { $cb->($_[0], Storable::thaw ($data)); 1 }
1744 or return $_[0]->_error (Errno::EBADMSG); 1854 or return $_[0]->error (Errno::EBADMSG);
1745 } else { 1855 } else {
1746 # remove prefix 1856 # remove prefix
1747 substr $_[0]{rbuf}, 0, $format, ""; 1857 substr $_[0]{rbuf}, 0, $format, "";
1748 1858
1749 # read remaining chunk 1859 # read remaining chunk
1750 $_[0]->unshift_read (chunk => $len, sub { 1860 $_[0]->unshift_read (chunk => $len, sub {
1751 eval { $cb->($_[0], Storable::thaw ($_[1])); 1 } 1861 eval { $cb->($_[0], Storable::thaw ($_[1])); 1 }
1752 or $_[0]->_error (Errno::EBADMSG); 1862 or $_[0]->error (Errno::EBADMSG);
1753 }); 1863 });
1754 } 1864 }
1755 1865
1756 1 1866 1
1757 } 1867 }
1868};
1869
1870=item tls_detect => $cb->($handle, $detect, $major, $minor)
1871
1872Checks the input stream for a valid SSL or TLS handshake TLSPaintext
1873record without consuming anything. Only SSL version 3 or higher
1874is handled, up to the fictituous protocol 4.x (but both SSL3+ and
1875SSL2-compatible framing is supported).
1876
1877If it detects that the input data is likely TLS, it calls the callback
1878with a true value for C<$detect> and the (on-wire) TLS version as second
1879and third argument (C<$major> is C<3>, and C<$minor> is 0..4 for SSL
18803.0, TLS 1.0, 1.1, 1.2 and 1.3, respectively). If it detects the input
1881to be definitely not TLS, it calls the callback with a false value for
1882C<$detect>.
1883
1884The callback could use this information to decide whether or not to start
1885TLS negotiation.
1886
1887In all cases the data read so far is passed to the following read
1888handlers.
1889
1890Usually you want to use the C<tls_autostart> read type instead.
1891
1892If you want to design a protocol that works in the presence of TLS
1893dtection, make sure that any non-TLS data doesn't start with the octet 22
1894(ASCII SYN, 16 hex) or 128-255 (i.e. highest bit set). The checks this
1895read type does are a bit more strict, but might losen in the future to
1896accomodate protocol changes.
1897
1898This read type does not rely on L<AnyEvent::TLS> (and thus, not on
1899L<Net::SSLeay>).
1900
1901=item tls_autostart => [$tls_ctx, ]$tls
1902
1903Tries to detect a valid SSL or TLS handshake. If one is detected, it tries
1904to start tls by calling C<starttls> with the given arguments.
1905
1906In practice, C<$tls> must be C<accept>, or a Net::SSLeay context that has
1907been configured to accept, as servers do not normally send a handshake on
1908their own and ths cannot be detected in this way.
1909
1910See C<tls_detect> above for more details.
1911
1912Example: give the client a chance to start TLS before accepting a text
1913line.
1914
1915 $hdl->push_read (tls_autostart => "accept");
1916 $hdl->push_read (line => sub {
1917 print "received ", ($_[0]{tls} ? "encrypted" : "cleartext"), " <$_[1]>\n";
1918 });
1919
1920=cut
1921
1922register_read_type tls_detect => sub {
1923 my ($self, $cb) = @_;
1924
1925 sub {
1926 # this regex matches a full or partial tls record
1927 if (
1928 # ssl3+: type(22=handshake) major(=3) minor(any) length_hi
1929 $self->{rbuf} =~ /^(?:\z| \x16 (\z| [\x03\x04] (?:\z| . (?:\z| [\x00-\x40] ))))/xs
1930 # ssl2 comapatible: len_hi len_lo type(1) major minor dummy(forlength)
1931 or $self->{rbuf} =~ /^(?:\z| [\x80-\xff] (?:\z| . (?:\z| \x01 (\z| [\x03\x04] (?:\z| . (?:\z| . ))))))/xs
1932 ) {
1933 return if 3 != length $1; # partial match, can't decide yet
1934
1935 # full match, valid TLS record
1936 my ($major, $minor) = unpack "CC", $1;
1937 $cb->($self, "accept", $major, $minor);
1938 } else {
1939 # mismatch == guaranteed not TLS
1940 $cb->($self, undef);
1941 }
1942
1943 1
1944 }
1945};
1946
1947register_read_type tls_autostart => sub {
1948 my ($self, @tls) = @_;
1949
1950 $RH{tls_detect}($self, sub {
1951 return unless $_[1];
1952 $_[0]->starttls (@tls);
1953 })
1758}; 1954};
1759 1955
1760=back 1956=back
1761 1957
1762=item custom read types - Package::anyevent_read_type $handle, $cb, @args 1958=item custom read types - Package::anyevent_read_type $handle, $cb, @args
1846 } elsif (defined $len) { 2042 } elsif (defined $len) {
1847 delete $self->{_rw}; 2043 delete $self->{_rw};
1848 $self->{_eof} = 1; 2044 $self->{_eof} = 1;
1849 $self->_drain_rbuf; 2045 $self->_drain_rbuf;
1850 2046
1851 } elsif ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK) { 2047 } elsif ($! != EAGAIN && $! != EINTR && $! != EWOULDBLOCK && $! != WSAEWOULDBLOCK) {
1852 return $self->_error ($!, 1); 2048 return $self->error ($!, 1);
1853 } 2049 }
1854 }; 2050 };
1855 } 2051 }
1856} 2052}
1857 2053
1859our $ERROR_WANT_READ; 2055our $ERROR_WANT_READ;
1860 2056
1861sub _tls_error { 2057sub _tls_error {
1862 my ($self, $err) = @_; 2058 my ($self, $err) = @_;
1863 2059
1864 return $self->_error ($!, 1) 2060 return $self->error ($!, 1)
1865 if $err == Net::SSLeay::ERROR_SYSCALL (); 2061 if $err == Net::SSLeay::ERROR_SYSCALL ();
1866 2062
1867 my $err = Net::SSLeay::ERR_error_string (Net::SSLeay::ERR_get_error ()); 2063 my $err = Net::SSLeay::ERR_error_string (Net::SSLeay::ERR_get_error ());
1868 2064
1869 # reduce error string to look less scary 2065 # reduce error string to look less scary
1872 if ($self->{_on_starttls}) { 2068 if ($self->{_on_starttls}) {
1873 (delete $self->{_on_starttls})->($self, undef, $err); 2069 (delete $self->{_on_starttls})->($self, undef, $err);
1874 &_freetls; 2070 &_freetls;
1875 } else { 2071 } else {
1876 &_freetls; 2072 &_freetls;
1877 $self->_error (Errno::EPROTO, 1, $err); 2073 $self->error (Errno::EPROTO, 1, $err);
1878 } 2074 }
1879} 2075}
1880 2076
1881# poll the write BIO and send the data if applicable 2077# poll the write BIO and send the data if applicable
1882# also decode read data if possible 2078# also decode read data if possible
1883# this is basiclaly our TLS state machine 2079# this is basically our TLS state machine
1884# more efficient implementations are possible with openssl, 2080# more efficient implementations are possible with openssl,
1885# but not with the buggy and incomplete Net::SSLeay. 2081# but not with the buggy and incomplete Net::SSLeay.
1886sub _dotls { 2082sub _dotls {
1887 my ($self) = @_; 2083 my ($self) = @_;
1888 2084
1889 my $tmp; 2085 my $tmp;
1890 2086
1891 if (length $self->{_tls_wbuf}) { 2087 while (length $self->{_tls_wbuf}) {
1892 while (($tmp = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) > 0) { 2088 if (($tmp = Net::SSLeay::write ($self->{tls}, $self->{_tls_wbuf})) <= 0) {
1893 substr $self->{_tls_wbuf}, 0, $tmp, ""; 2089 $tmp = Net::SSLeay::get_error ($self->{tls}, $tmp);
2090
2091 return $self->_tls_error ($tmp)
2092 if $tmp != $ERROR_WANT_READ
2093 && ($tmp != $ERROR_SYSCALL || $!);
2094
2095 last;
1894 } 2096 }
1895 2097
1896 $tmp = Net::SSLeay::get_error ($self->{tls}, $tmp); 2098 substr $self->{_tls_wbuf}, 0, $tmp, "";
1897 return $self->_tls_error ($tmp)
1898 if $tmp != $ERROR_WANT_READ
1899 && ($tmp != $ERROR_SYSCALL || $!);
1900 } 2099 }
1901 2100
1902 while (defined ($tmp = Net::SSLeay::read ($self->{tls}))) { 2101 while (defined ($tmp = Net::SSLeay::read ($self->{tls}))) {
1903 unless (length $tmp) { 2102 unless (length $tmp) {
1904 $self->{_on_starttls} 2103 $self->{_on_starttls}
1918 $self->{_tls_rbuf} .= $tmp; 2117 $self->{_tls_rbuf} .= $tmp;
1919 $self->_drain_rbuf; 2118 $self->_drain_rbuf;
1920 $self->{tls} or return; # tls session might have gone away in callback 2119 $self->{tls} or return; # tls session might have gone away in callback
1921 } 2120 }
1922 2121
1923 $tmp = Net::SSLeay::get_error ($self->{tls}, -1); 2122 $tmp = Net::SSLeay::get_error ($self->{tls}, -1); # -1 is not neccessarily correct, but Net::SSLeay doesn't tell us
1924 return $self->_tls_error ($tmp) 2123 return $self->_tls_error ($tmp)
1925 if $tmp != $ERROR_WANT_READ 2124 if $tmp != $ERROR_WANT_READ
1926 && ($tmp != $ERROR_SYSCALL || $!); 2125 && ($tmp != $ERROR_SYSCALL || $!);
1927 2126
1928 while (length ($tmp = Net::SSLeay::BIO_read ($self->{_wbio}))) { 2127 while (length ($tmp = Net::SSLeay::BIO_read ($self->{_wbio}))) {
1981 unless (defined $AnyEvent::TLS::VERSION) { 2180 unless (defined $AnyEvent::TLS::VERSION) {
1982 eval { 2181 eval {
1983 require Net::SSLeay; 2182 require Net::SSLeay;
1984 require AnyEvent::TLS; 2183 require AnyEvent::TLS;
1985 1 2184 1
1986 } or return $self->_error (Errno::EPROTO, 1, "TLS support not available on this system"); 2185 } or return $self->error (Errno::EPROTO, 1, "TLS support not available on this system");
1987 } 2186 }
1988 2187
1989 $self->{tls} = $tls; 2188 $self->{tls} = $tls;
1990 $self->{tls_ctx} = $ctx if @_ > 2; 2189 $self->{tls_ctx} = $ctx if @_ > 2;
1991 2190
2012 $self->{tls} = $tls = $self->{tls_ctx}->_get_session ($tls, $self, $self->{peername}); 2211 $self->{tls} = $tls = $self->{tls_ctx}->_get_session ($tls, $self, $self->{peername});
2013 2212
2014 # basically, this is deep magic (because SSL_read should have the same issues) 2213 # basically, this is deep magic (because SSL_read should have the same issues)
2015 # but the openssl maintainers basically said: "trust us, it just works". 2214 # but the openssl maintainers basically said: "trust us, it just works".
2016 # (unfortunately, we have to hardcode constants because the abysmally misdesigned 2215 # (unfortunately, we have to hardcode constants because the abysmally misdesigned
2017 # and mismaintained ssleay-module doesn't even offer them). 2216 # and mismaintained ssleay-module didn't offer them for a decade or so).
2018 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html 2217 # http://www.mail-archive.com/openssl-dev@openssl.org/msg22420.html
2019 # 2218 #
2020 # in short: this is a mess. 2219 # in short: this is a mess.
2021 # 2220 #
2022 # note that we do not try to keep the length constant between writes as we are required to do. 2221 # note that we do not try to keep the length constant between writes as we are required to do.
2023 # we assume that most (but not all) of this insanity only applies to non-blocking cases, 2222 # we assume that most (but not all) of this insanity only applies to non-blocking cases,
2024 # and we drive openssl fully in blocking mode here. Or maybe we don't - openssl seems to 2223 # and we drive openssl fully in blocking mode here. Or maybe we don't - openssl seems to
2025 # have identity issues in that area. 2224 # have identity issues in that area.
2026# Net::SSLeay::CTX_set_mode ($ssl, 2225# Net::SSLeay::set_mode ($ssl,
2027# (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1) 2226# (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ENABLE_PARTIAL_WRITE () } || 1)
2028# | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2)); 2227# | (eval { local $SIG{__DIE__}; Net::SSLeay::MODE_ACCEPT_MOVING_WRITE_BUFFER () } || 2));
2029 Net::SSLeay::CTX_set_mode ($tls, 1|2); 2228 Net::SSLeay::set_mode ($tls, 1|2);
2030 2229
2031 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 2230 $self->{_rbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
2032 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ()); 2231 $self->{_wbio} = Net::SSLeay::BIO_new (Net::SSLeay::BIO_s_mem ());
2033 2232
2034 Net::SSLeay::BIO_write ($self->{_rbio}, $self->{rbuf}); 2233 Net::SSLeay::BIO_write ($self->{_rbio}, $self->{rbuf});
2074 2273
2075 return unless $self->{tls}; 2274 return unless $self->{tls};
2076 2275
2077 $self->{tls_ctx}->_put_session (delete $self->{tls}) 2276 $self->{tls_ctx}->_put_session (delete $self->{tls})
2078 if $self->{tls} > 0; 2277 if $self->{tls} > 0;
2079 2278
2080 delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)}; 2279 delete @$self{qw(_rbio _wbio _tls_wbuf _on_starttls)};
2081} 2280}
2082 2281
2083=item $handle->resettls 2282=item $handle->resettls
2084 2283
2109 push @linger, AE::io $fh, 1, sub { 2308 push @linger, AE::io $fh, 1, sub {
2110 my $len = syswrite $fh, $wbuf, length $wbuf; 2309 my $len = syswrite $fh, $wbuf, length $wbuf;
2111 2310
2112 if ($len > 0) { 2311 if ($len > 0) {
2113 substr $wbuf, 0, $len, ""; 2312 substr $wbuf, 0, $len, "";
2114 } elsif (defined $len || ($! != EAGAIN && $! != EINTR && $! != WSAEWOULDBLOCK)) { 2313 } elsif (defined $len || ($! != EAGAIN && $! != EINTR && $! != EWOULDBLOCK && $! != WSAEWOULDBLOCK)) {
2115 @linger = (); # end 2314 @linger = (); # end
2116 } 2315 }
2117 }; 2316 };
2118 push @linger, AE::timer $linger, 0, sub { 2317 push @linger, AE::timer $linger, 0, sub {
2119 @linger = (); 2318 @linger = ();

Diff Legend

Removed lines
+ Added lines
< Changed lines
> Changed lines