This file is indexed.

/usr/share/perl5/IO/Async/Channel.pm is in libio-async-perl 0.71-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
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
#  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-2016 -- leonerd@leonerd.org.uk

package IO::Async::Channel;

use strict;
use warnings;
use base qw( IO::Async::Notifier );

our $VERSION = '0.71';

use Carp;

use IO::Async::Stream;

use IO::Async::OS;
use constant DEFAULT_CODEC => IO::Async::OS->HAVE_SEREAL ? "Sereal" : "Storable";

=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
L<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.

=head1 PARAMETERS

The following named parameters may be passed to C<new> or C<configure>:

=head2 codec => STR

Gives the name of the encoding method used to represent values over the
channel.

This can be set to C<Storable> to use the core L<Storable> module. As this
only supports references, to pass a single scalar value, C<send> a SCALAR
reference to it, and dereference the result of C<recv>.

If the L<Sereal::Encoder> and L<Sereal::Decoder> modules are installed, this
can be set to C<Sereal> instead, and will use those to perform the encoding
and decoding. This optional dependency may give higher performance than using
C<Storable>. If these modules are available, then this option is picked by
default.

=cut

=head1 CONSTRUCTOR

=cut

=head2 new

   $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>, it should
not be added to a Loop object directly; event management will be handled by
its containing L<IO::Async::Routine> object.

=cut

=head1 METHODS

The following methods documented with a trailing call to C<< ->get >> return
L<Future> instances.

=cut

=head2 configure

   $channel->configure( %params )

Similar to the standard C<configure> method on L<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 _init
{
   my $self = shift;
   my ( $params ) = @_;

   defined $params->{codec} or $params->{codec} = DEFAULT_CODEC;

   $self->SUPER::_init( $params );
}

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;
   }

   if( my $codec = delete $params{codec} ) {
      @{ $self }{qw( encode decode )} = (
         $self->can( "_make_codec_$codec" ) or croak "Unrecognised codec name '$codec'"
      )->();
   }

   $self->SUPER::configure( %params );
}

sub _make_codec_Storable
{
   require Storable;

   return
      \&Storable::freeze,
      \&Storable::thaw;
}

sub _make_codec_Sereal
{
   require Sereal::Encoder;
   require Sereal::Decoder;

   my $encoder;
   my $decoder;

   # "thread safety" to Sereal::{Encoder,Decoder} means that the variables get
   # reset to undef in new threads. We should defend against that.

   return
      sub { ( $encoder ||= Sereal::Encoder->new )->encode( $_[0] ) },
      sub { ( $decoder ||= Sereal::Decoder->new )->decode( $_[0] ) };
}

=head2 send

   $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 ) = @_;

   $self->send_encoded( $self->{encode}->( $data ) );
}

=head2 send_encoded

   $channel->send_encoded( $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<encode>.

=cut

sub send_encoded
{
   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 encode

   $record = $channel->encode( $data )

Takes a Perl reference and returns a serialised string that can be passed to
C<send_encoded>. The following two forms are equivalent

 $channel->send( $data )
 $channel->send_encoded( $channel->encode( $data ) )

This is provided for the use-case where data needs to be serialised into a
fixed string to "snapshot it" but not sent yet; the returned string can be
saved and sent at a later time.

   $record = IO::Async::Channel->encode( $data )

This can also be used as a class method, in case it is inconvenient to operate
on a particular object instance, or when one does not exist yet. In this case
it will encode using whatever is the default codec for C<IO::Async::Channel>.

=cut

my $default_encode;
sub encode
{
   my $self = shift;
   my ( $data ) = @_;

   return ( ref $self ?
      $self->{encode} :
      $default_encode ||= do { ( $self->can( "_make_codec_".DEFAULT_CODEC )->() )[0] }
   )->( $data );
}

=head2 send_frozen

   $channel->send_frozen( $record )

Legacy name for C<send_encoded>. This is no longer preferred as it expects
the data to be encoded using C<Storable>, which prevents (or at least makes
more awkward) the use of other codecs on a channel by default. This method
should not be used in new code and may be removed in a later version.

=cut

*send_frozen = \&send_encoded;

=head2 recv

   $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.

   $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.

   $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 close

   $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 $self->{decode}->( $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 = $self->{decode}->( 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;