File Coverage

blib/lib/Future/Buffer.pm

Criterion	Covered	Total	%
statement	110	112	98.2
branch	28	38	73.6
condition	22	55	40.0
subroutine	25	25	100.0
pod	9	9	100.0
total	194	239	81.1

#  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, 2020-2022 -- [e-mail]

package Future::Buffer;

use 5.010; # //
use strict;
use warnings;

our $VERSION = '0.03';

use Future;

use Scalar::Util qw( weaken );

=head1 NAME

C - a string buffer that uses Futures

=head1 SYNOPSIS

   use Future::Buffer;

   use Future::AsyncAwait;
   use Future::IO;

   my $buffer = Future::Buffer->new(
      fill => sub { Future::IO->sysread( $socket, 8192 ) }
   );

   async sub print_lines
   {
      while(1) {
         my $line = await $buffer->read_until( "\n" );
         chomp $line;

         say "Got a line: $line";
      }
   }

   print_lines()->get;

=head1 DESCRIPTION

Objects in this class provide a string buffer, on which operations return
L instances which will complete when data is available. Data can be
inserted into the buffer either in a push-based manner by calling the C
method, or in a pull-based manner by providing it with a C callback by
which it can request data itself. This flexibility allows the buffer to act as
an adapter between push- and pull-based providers and consumers.

Each C-like method returns a L which will complete once there
are enough bytes in the buffer to satisfy the required condition. The buffer
behaves somewhat like a pipe, where bytes provided at the writing end (either
by the C method or the C callback) are eventually consumed at the
reading end by one of the C futures.

Multiple C futures can remain pending at once, and will be completed in
the order they were created when more data is eventually available. Thus, any
call to the C method to provide more data can potentially result in
multiple futures becoming ready.

=cut

=head1 CONSTRUCTOR

=cut

=head2 new

   $buffer = Future::Buffer->new( %args )

Returns a new L instance.

Takes the following named arguments:

=over 4

=item fill => CODE

   $f = $fill->()

      $data = $f->get

Optional callback which the buffer will invoke when it needs more data.

Any read futures which are waiting on the fill future are constructed by using
the fill future as a prototype, ensuring they have the correct type.

=back

=cut

sub new
{
   my $class = shift;
   my %args = @_;

   return bless {
      pending => [],
      data    => "",
      fill    => $args{fill},
   }, $class;
}

=head1 METHODS

=cut

sub _fill
{
   my $self = shift;
   return $self->{fill_f} //= do {
      weaken( my $weakself = $self );
      my $fill = $self->{fill};

      # Arm the fill loop
      $fill->() # TODO: give it a size hint?
         ->on_done( sub {
            my ( $data ) = @_;
            $weakself or return;

            $weakself->{data} .= $data;
            undef $self->{fill_f};

            $weakself->_invoke_pending;

            $weakself->_fill if @{ $self->{pending} };
         });
   };
}

sub _new_read_future
{
   my $self = shift;
   my ( $code ) = @_;

   my $pending = $self->{pending};

   # First see if the buffer is already sufficient;
   if( !@$pending and
         ( my @ret = $code->( \$self->{data} ) ) ) {
      return Future->done( @ret );
   }

   my $f;
   if( $self->{fill} and my $fill_f = $self->_fill ) {
      $f = $fill_f->new;
   }
   else {
      $f = Future->new;
   }

   push @$pending, [ $code, $f ];

   $self->_invoke_pending if length $self->{data};

   $f->on_cancel( sub {
      shift @$pending while @$pending and $pending->[0]->[1]->is_cancelled;
      return if @$pending or !$self->{fill_f};

      $self->{fill_f}->cancel;
      undef $self->{fill_f};
   } );

   return $f;
}

sub _invoke_pending
{
   my $self = shift;

   my $pending = $self->{pending};

   while( @$pending and length $self->{data} ) {
      my $p = $pending->[0];
      shift @$pending and next if $p->[1]->is_cancelled;

      defined( my $ret = $p->[0]->( \$self->{data} ) )
         or last;

      shift @$pending;
      $p->[1]->done( $ret );
   }
}

=head2 length

   $len = $buffer->length

Returns the length of the currently-stored data; that is, data that has been
provided by C calls or the C callback but not yet consumed by a
C future.

=cut

sub length :method { length $_[0]->{data} }

=head2 is_empty

   $empty = $buffer->is_empty

Returns true if the stored length is zero.

=cut

sub is_empty { shift->length == 0 }

=head2 write

   $f = $buffer->write( $data )

Appends to the stored data, invoking any pending C futures that are
outstanding and can now complete.

Currently this method returns an already-completed C. Some later
version may implement a buffer maximum size, and choose not to complete this
future until there is enough space to accept the new data. For now it is safe
for the caller to ignore the return value, but it may become not so.

=cut

sub write
{
   my $self = shift;
   $self->{data} .= $_[0];

   $self->_invoke_pending if @{ $self->{pending} };

   return Future->done;
}

=head2 read_atmost

   $f = $buffer->read_atmost( $len )

      $data = $f->get

Returns a future which will complete when there is some data available in the
buffer and will yield I the given length. Note that, analogous to
calling the C IO method on a filehandle, this can still complete and
yield a shorter length if less is currently available.

=cut

sub read_atmost
{
   my $self = shift;
   my ( $maxlen ) = @_;

   return $self->_new_read_future(
      sub {
         my ( $dref ) = @_;
         return unless length $$dref;

         return substr( $$dref, 0, $maxlen, "" );
      }
   );
}

=head2 read_exactly

   $f = $buffer->read_exactly( $len )

      $data = $f->get

Returns a future which will complete when there is enough data available in
the buffer to yield exactly the length given.

=cut

sub read_exactly
{
   my $self = shift;
   my ( $len ) = @_;

   return $self->_new_read_future(
      sub {
         my ( $dref ) = @_;
         return unless length $$dref >= $len;

         return substr( $$dref, 0, $len, "" );
      }
   );
}

=head2 read_until

   $f = $buffer->read_until( $pattern )

      $data = $f->get

Returns a future which will complete when the buffer contains a match for the
given pattern (which may either be a plain string or a compiled C).
The future will yield the contents of the buffer up to and including this
match.

For example, a C-like operation can be performed by

   $f = $buffer->read_until( "\x0d\x0a" );

=cut

sub read_until
{
   my $self = shift;
   my ( $pattern ) = @_;

   $pattern = qr/\Q$pattern/ unless ref $pattern eq "Regexp";

   return $self->_new_read_future(
      sub {
         my ( $dref ) = @_;
         return unless $$dref =~ m/$pattern/;

         return substr( $$dref, 0, $+[0], "" );
      }
   );
}

=head2 read_unpacked

   $f = $buffer->read_unpacked( $pack_format )

      @fields = $f->get

I

Returns a future which will complete when the buffer contains enough data to
unpack all of the requested fields using the given C format. The
future will yield a list of all the fields extracted by the format.

Note that because the implementation is shamelessly stolen from
L the same limitations on what pack formats are
recognized will apply.

=cut

# Gratuitously stolen from IO::Handle::Packable

use constant {
   BYTES_FMT_i => length( pack "i", 0 ),
   BYTES_FMT_f => length( pack "f", 0 ),
   BYTES_FMT_d => length( pack "d", 0 ),
};

sub _length_of_packformat
{
   my ( $format ) = @_;
   local $_ = $format;

   my $bytes = 0;
   while( length ) {
      s/^\s+//;
      length or last;

      my $this;

      # Basic template
      s/^[aAcC]// and $this = 1 or
      s/^[sSnv]// and $this = 2 or
      s/^[iI]//   and $this = BYTES_FMT_i or
      s/^[lLNV]// and $this = 4 or
      s/^[qQ]//   and $this = 8 or
      s/^f//      and $this = BYTES_FMT_f or
      s/^d//      and $this = BYTES_FMT_d or
         die "TODO: unrecognised template char ${\substr $_, 0, 1}\n";

      # Ignore endian specifiers
      s/^[<>]//;

      # Repeat count
      s/^(\d+)// and $this *= $1;

      $bytes += $this;
   }

   return $bytes;
}

sub read_unpacked
{
   my $self = shift;
   my ( $format ) = @_;

   my $len = _length_of_packformat $format;
   return $self->_new_read_future(
      sub {
         my ( $dref ) = @_;
         return unless length $$dref >= $len;

         return unpack $format, substr( $$dref, 0, $len, "" );
      }
   );
}

=head2 unread

   $buffer->unread( $data )

I

Prepends more data back into the buffer,

It is uncommon to need this method, but it may be useful in certain situations
such as when it is hard to determine upfront how much data needs to be read
for a single operation, and it turns out too much was read. The trailing
content past what is needed can be put back for a later operation.

Note that use of this method causes an inherent race condition between
outstanding read futures and existing data in the buffer. If there are no
pending futures then this is safe. If there is no existing data already in the
buffer this is also safe. If neither of these is true then a warning is
printed indicating that the logic of the caller is not well-defined.

=cut

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

   if( @{ $self->{pending} } and length $self->{data} ) {
      warn "Racy use of ->unread with both pending read futures and existing data";
   }

   $self->{data} = $data . $self->{data};
   $self->_invoke_pending if @{ $self->{pending} };

   return Future->done;
}

=head1 TODO

=over 4

=item *

An "on-read" event, taking maybe inspiration from L. This
would allow both pull- and push-based consumers.

=item *

Size limitation. Allow an upper bound of stored data, make C calls
return pending futures until buffer can accept it. Needs consideration of
unbounded C though.

=item *

Consider extensions of the L method to handle more situations.
This may require building a shared CPAN module for doing streaming-unpack
along with C and other situations.

=item *

Consider what happens at EOF. Add a C method for producers to call.
Understand what C would do there. Have all the pending C futures
yield an empty list maybe?

=back

=head1 AUTHOR

Paul Evans

Inspired by L by Tom Molesworth

=cut

0x55AA;