File Coverage

blib/lib/Mojo/IOLoop/Stream/Role/LineBuffer.pm

Criterion	Covered	Total	%
statement	29	29	100.0
branch	8	10	80.0
condition	1	2	50.0
subroutine	5	5	100.0
pod	2	2	100.0
total	45	48	93.7

package Mojo::IOLoop::Stream::Role::LineBuffer;

use Mojo::Base -role;

our $VERSION = '0.006';

has 'read_line_separator' => sub { qr/\x0D?\x0A/ };
has 'write_line_separator' => "\x0D\x0A";

requires qw(on emit write);

sub watch_lines {
  my $self = shift;
  return $self if $self->{_read_line_read_cb};
  $self->{_read_line_read_cb} = $self->on(read => sub {
    my ($self, $bytes) = @_;
    $self->{_read_line_buffer} .= $bytes;
    my $sep = $self->read_line_separator;
    my $pos;
    while ($self->{_read_line_buffer} =~ m/\G(.*?)($sep)/gs) {
      $pos = pos $self->{_read_line_buffer};
      $self->emit(read_line => "$1", "$2");
    } continue {
      $sep = $self->read_line_separator;
    }
    $self->{_read_line_buffer} = substr($self->{_read_line_buffer}, $pos)
      if $pos;
  });
  $self->{_read_line_close_cb} = $self->on(close => sub {
    my $self = shift;
    if (length(my $buffer = delete $self->{_read_line_buffer} // '')) {
      my $sep = $self->read_line_separator;
      my $pos = 0;
      while ($buffer =~ m/\G(.*?)($sep)/gs) {
        $pos = pos $buffer;
        $self->emit(read_line => "$1", "$2");
      } continue {
        $sep = $self->read_line_separator;
      }
      $self->emit(read_line => $pos ? substr($buffer, $pos) : $buffer)
        if $pos < length $buffer;
    }
  });
  return $self;
}

sub write_line {
  my ($self, $line) = (shift, shift);
  my $sep = $self->write_line_separator;
  $self->write("$line$sep", @_);
}

1;

=head1 NAME

Mojo::IOLoop::Stream::Role::LineBuffer - Read and write streams by lines

=head1 SYNOPSIS

  use Mojo::IOLoop;
  use Mojo::IOLoop::Stream;
  my $output_stream = Mojo::IOLoop::Stream->with_roles('+LineBuffer')->new($handle);
  Mojo::IOLoop->client({port => 3000} => sub {
    my ($loop, $err, $stream) = @_;
    $stream->with_roles('+LineBuffer')->watch_lines->on(read_line => sub {
      my ($stream, $line) = @_;
      say "Received line: $line";
      $output_stream->write_line('Got it!');
    });
  });

=head1 DESCRIPTION

L composes the method
L which causes a L object to emit the
L event for each line received. The L method is
also provided to add a line separator to the passed data before writing.

=head1 EVENTS

L can emit the following events.

=head2 read_line

  $stream->on(read_line => sub {
    my ($stream, $line, $separator) = @_;
    ...
  });

Emitted when a line ending in L arrives on the stream,
and when the stream closes if data is still buffered. The separator is passed
as a separate argument if present.

=head1 ATTRIBUTES

L composes the following attributes.

=head2 read_line_separator

  my $separator = $stream->read_line_separator;
  $stream       = $stream->read_line_separator(qr/\x0D\x0A/);

Regular expression to indicate new lines in received bytes. Defaults to a
newline (LF) character optionally preceded by a CR character (C<\x0D?\x0A>).
Note that if you set this to L or
C<\v> (vertical whitespace), this may match the CR character of a CR/LF
sequence and consider the LF as a separate line if they are read separately.

=head2 write_line_separator

  my $separator = $stream->write_line_separator;
  $stream       = $stream->write_line_separator("\x0A");

Byte sequence to indicate new lines in data written with L.
Defaults to the network newline CR/LF (C<\x0D\x0A>).

=head1 METHODS

L composes the following methods.

=head2 watch_lines

  $stream = $stream->watch_lines;

Subscribe to the L and
L events, to buffer received bytes and emit
L when L is encountered or the stream is
closed with buffered data.

=head2 write_line

  $stream = $stream->write_line($bytes);
  $stream = $stream->write_line($bytes => sub {...});

Write a line to the stream by appending L to the data.
The optional drain callback will be executed once all data has been written.

=head1 BUGS

Report any issues on the public bugtracker.

=head1 AUTHOR

Dan Book

=head1 COPYRIGHT AND LICENSE

This software is Copyright (c) 2018 by Dan Book.

This is free software, licensed under:

  The Artistic License 2.0 (GPL Compatible)

=head1 SEE ALSO

L, L, L,
L