Non-blocking read lines from a slow standard input (mojolicious) (perl)

slow_stdin_

#!/Users/brian/bin/perls/perl-latest
use v5.28;
use utf8;
use strict;
use warnings;
use feature qw(signatures);
no warnings qw(experimental::signatures);

# non-blocking stream reading of a slow standard input
# perl -pe '$|++; sleep 1' /etc/hosts | perl-latest read_slowly.pl

use Mojo::UserAgent;
my $ua = Mojo::UserAgent->new;

use Mojo::IOLoop;

STDOUT->autoflush(1);

my $urls = [];
END { say join "\n", $urls->@* }

my $stream = make_stdin_stream(
	sub ($stream, $line) {
		say "Line [$line]";
		push @$urls, $line;
		},
	sub ($stream) { say "Closing" },
	sub ($stream, $message) { warn "Error: $message" },
	);


$stream->start;
$stream->reactor->start unless $stream->reactor->is_running;


sub make_stdin_stream ( $line, $close, $error ) {
	state $rc = require Mojo::IOLoop::Stream;
	$line  //= sub {};
	$close //= sub {};
	$error //= sub {};

	my $stream = Mojo::IOLoop::Stream->new( *STDIN{IO} );
	my $sid = Mojo::IOLoop->stream( $stream );

	$stream->on(
		error => $error
		);
	$stream->on(
		read  => sub ( $stream, $octets ) {
			state $buffer = '';
			next unless length $octets;
			$buffer .= $octets;
			my @lines = split /\n/, $buffer;
			$buffer = '';
			unless( $octets =~ /\n/ ) {
				$buffer = pop @lines;
				}

			$line->( $stream, $_ ) for @lines;
			});
	$stream->on(close => sub {
		$close->( $stream );
		Mojo::IOLoop->remove($sid);
		});

	$stream;
	}