/usr/share/perl5/IO/Async/Channel.pm is in libio-async-perl 0.64-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373 374 375 376 377 378 379 380 381 382 383 384 385 386 387 388 389 390 391 392 393 394 395 396 397 398 399 400 401 402 403 404 405 406 407 408 409 410 411 412 413 414 415 416 417 418 419 420 421 422 423 424 425 426 427 428 429 430 | # You may distribute under the terms of either the GNU General Public License
# or the Artistic License (the same terms as Perl itself)
#
# (C) Paul Evans, 2011-2014 -- leonerd@leonerd.org.uk
package IO::Async::Channel;
use strict;
use warnings;
use base qw( IO::Async::Notifier ); # just to get _capture_weakself
our $VERSION = '0.64';
use Carp;
use Storable qw( freeze thaw );
use IO::Async::Stream;
=head1 NAME
C<IO::Async::Channel> - pass values into or out from an L<IO::Async::Routine>
=head1 DESCRIPTION
A C<IO::Async::Channel> object allows Perl values to be passed into or out of
an L<IO::Async::Routine>. It is intended to be used primarily with a Routine
object rather than independently. For more detail and examples on how to use
this object see also the documentation for L<IO::Async::Routine>.
A Channel object is shared between the main process of the program and the
process running within the Routine. In the main process it will be used in
asynchronous mode, and in the Routine process it will be used in synchronous
mode. In asynchronous mode all methods return immediately and use
C<IO::Async>-style futures or callback functions. In synchronous within the
Routine process the methods block until they are ready and may be used for
flow-control within the routine. Alternatively, a Channel may be shared
between two different Routine objects, and not used directly by the
controlling program.
The channel itself represents a FIFO of Perl reference values. New values may
be put into the channel by the C<send> method in either mode. Values may be
retrieved from it by the C<recv> method. Values inserted into the Channel are
snapshot by the C<send> method. Any changes to referred variables will not be
observed by the other end of the Channel after the C<send> method returns.
Since the channel uses L<Storable> to serialise values to write over the
communication filehandle only reference values may be passed. To pass a single
scalar value, C<send> a SCALAR reference to it, and dereference the result of
C<recv>.
=cut
=head1 CONSTRUCTOR
=cut
=head2 $channel = IO::Async::Channel->new
Returns a new C<IO::Async::Channel> object. This object reference itself
should be shared by both sides of a C<fork()>ed process. After C<fork()> the
two C<setup_*> methods may be used to configure the object for operation on
either end.
While this object does in fact inherit from L<IO::Async::Notifier> for
implementation reasons it is not intended that this object be used as a
Notifier. It should not be added to a Loop object directly; event management
will be handled by its containing C<IO::Async::Routine> object.
=cut
sub new
{
my $class = shift;
return bless {
mode => "",
}, $class;
}
=head1 METHODS
The following methods documented with a trailing call to C<< ->get >> return
L<Future> instances.
=cut
=head2 $channel->configure( %params )
Similar to the standard C<configure> method on C<IO::Async::Notifier>, this is
used to change details of the Channel's operation.
=over 4
=item on_recv => CODE
May only be set on an async mode channel. If present, will be invoked whenever
a new value is received, rather than using the C<recv> method.
$on_recv->( $channel, $data )
=item on_eof => CODE
May only be set on an async mode channel. If present, will be invoked when the
channel gets closed by the peer.
$on_eof->( $channel )
=back
=cut
sub configure
{
my $self = shift;
my %params = @_;
foreach (qw( on_recv on_eof )) {
next unless exists $params{$_};
$self->{mode} and $self->{mode} eq "async" or
croak "Can only configure $_ in async mode";
$self->{$_} = delete $params{$_};
$self->_build_stream;
}
$self->SUPER::configure( %params );
}
=head2 $channel->send( $data )
Pushes the data stored in the given Perl reference into the FIFO of the
Channel, where it can be received by the other end. When called on a
synchronous mode Channel this method may block if a C<write()> call on the
underlying filehandle blocks. When called on an asynchronous mode channel this
method will not block.
=cut
sub send
{
my $self = shift;
my ( $data ) = @_;
my $record = freeze $data;
$self->send_frozen( $record );
}
=head2 $channel->send_frozen( $record )
A variant of the C<send> method; this method pushes the byte record given.
This should be the result of a call to C<Storable::freeze()>.
=cut
sub send_frozen
{
my $self = shift;
my ( $record ) = @_;
my $bytes = pack( "I", length $record ) . $record;
defined $self->{mode} or die "Cannot ->send without being set up";
return $self->_send_sync( $bytes ) if $self->{mode} eq "sync";
return $self->_send_async( $bytes ) if $self->{mode} eq "async";
}
=head2 $data = $channel->recv
When called on a synchronous mode Channel this method will block until a Perl
reference value is available from the other end and then return it. If the
Channel is closed this method will return C<undef>. Since only references may
be passed and all Perl references are true the truth of the result of this
method can be used to detect that the channel is still open and has not yet
been closed.
=head2 $data = $channel->recv->get
When called on an asynchronous mode Channel this method returns a future which
will eventually yield the next Perl reference value that becomes available
from the other end. If the Channel is closed, the future will fail with an
C<eof> failure.
=head2 $channel->recv( %args )
When not returning a future, takes the following named arguments:
=over 8
=item on_recv => CODE
Called when a new Perl reference value is available. Will be passed the
Channel object and the reference data.
$on_recv->( $channel, $data )
=item on_eof => CODE
Called if the Channel was closed before a new value was ready. Will be passed
the Channel object.
$on_eof->( $channel )
=back
=cut
sub recv
{
my $self = shift;
defined $self->{mode} or die "Cannot ->recv without being set up";
return $self->_recv_sync( @_ ) if $self->{mode} eq "sync";
return $self->_recv_async( @_ ) if $self->{mode} eq "async";
}
=head2 $channel->close
Closes the channel. Causes a pending C<recv> on the other end to return undef
or the queued C<on_eof> callbacks to be invoked.
=cut
sub close
{
my $self = shift;
return $self->_close_sync if $self->{mode} eq "sync";
return $self->_close_async if $self->{mode} eq "async";
}
# Leave this undocumented for now
sub setup_sync_mode
{
my $self = shift;
( $self->{fh} ) = @_;
$self->{mode} = "sync";
# Since we're communicating binary structures and not Unicode text we need to
# enable binmode
binmode $self->{fh};
$self->{fh}->autoflush(1);
}
sub _read_exactly
{
$_[1] = "";
while( length $_[1] < $_[2] ) {
my $n = read( $_[0], $_[1], $_[2]-length $_[1], length $_[1] );
defined $n or return undef;
$n or return "";
}
return $_[2];
}
sub _recv_sync
{
my $self = shift;
my $n = _read_exactly( $self->{fh}, my $lenbuffer, 4 );
defined $n or die "Cannot read - $!";
length $n or return undef;
my $len = unpack( "I", $lenbuffer );
$n = _read_exactly( $self->{fh}, my $record, $len );
defined $n or die "Cannot read - $!";
length $n or return undef;
return thaw $record;
}
sub _send_sync
{
my $self = shift;
my ( $bytes ) = @_;
$self->{fh}->print( $bytes );
}
sub _close_sync
{
my $self = shift;
$self->{fh}->close;
}
# Leave this undocumented for now
sub setup_async_mode
{
my $self = shift;
my %args = @_;
exists $args{$_} and $self->{$_} = delete $args{$_} for qw( read_handle write_handle );
keys %args and croak "Unrecognised keys for setup_async_mode: " . join( ", ", keys %args );
$self->{mode} = "async";
}
sub _build_stream
{
my $self = shift;
return $self->{stream} ||= do {
$self->{on_result_queue} = [];
my $stream = IO::Async::Stream->new(
read_handle => $self->{read_handle},
write_handle => $self->{write_handle},
autoflush => 1,
on_read => $self->_capture_weakself( '_on_stream_read' )
);
$self->add_child( $stream );
$stream;
};
}
sub _send_async
{
my $self = shift;
my ( $bytes ) = @_;
$self->_build_stream->write( $bytes );
}
sub _recv_async
{
my $self = shift;
my %args = @_;
my $on_recv = $args{on_recv};
my $on_eof = $args{on_eof};
my $stream = $self->_build_stream;
my $f;
$f = $stream->loop->new_future unless !defined wantarray;
push @{ $self->{on_result_queue} }, sub {
my ( $self, $type, $result ) = @_;
if( $type eq "recv" ) {
$f->done( $result ) if $f and !$f->is_cancelled;
$on_recv->( $self, $result ) if $on_recv;
}
else {
$f->fail( "EOF waiting for Channel recv", eof => ) if $f and !$f->is_cancelled;
$on_eof->( $self ) if $on_eof;
}
};
return $f;
}
sub _close_async
{
my $self = shift;
if( my $stream = $self->{stream} ) {
$stream->close_when_empty;
}
else {
$_ and $_->close for $self->{read_handle}, $self->{write_handle};
}
undef $_ for $self->{read_handle}, $self->{write_handle};
}
sub _on_stream_read
{
my $self = shift or return;
my ( $stream, $buffref, $eof ) = @_;
if( $eof ) {
while( my $on_result = shift @{ $self->{on_result_queue} } ) {
$on_result->( $self, eof => );
}
$self->{on_eof}->( $self ) if $self->{on_eof};
return;
}
return 0 unless length( $$buffref ) >= 4;
my $len = unpack( "I", $$buffref );
return 0 unless length( $$buffref ) >= 4 + $len;
my $record = thaw( substr( $$buffref, 4, $len ) );
substr( $$buffref, 0, 4 + $len ) = "";
if( my $on_result = shift @{ $self->{on_result_queue} } ) {
$on_result->( $self, recv => $record );
}
else {
$self->{on_recv}->( $self, $record );
}
return 1;
}
sub _extract_read_handle
{
my $self = shift;
return undef if !$self->{mode};
croak "Cannot extract filehandle" if $self->{mode} ne "async";
$self->{mode} = "dead";
return $self->{read_handle};
}
sub _extract_write_handle
{
my $self = shift;
return undef if !$self->{mode};
croak "Cannot extract filehandle" if $self->{mode} ne "async";
$self->{mode} = "dead";
return $self->{write_handle};
}
=head1 AUTHOR
Paul Evans <leonerd@leonerd.org.uk>
=cut
0x55AA;
|