Acme-Parataxis
view release on metacpan or search on metacpan
my ($self, $method, $url, $args) = @_;
my %new_args = %{ $args // {} };
my $orig_cb = $new_args{data_callback};
my $content = '';
$new_args{data_callback} = sub {
my ($data, $response) = @_;
if ($orig_cb) { return $orig_cb->($data, $response) }
$content .= $data;
return 1;
};
my $res = $self->SUPER::request($method, $url, \%new_args);
$res->{content} = $content unless $orig_cb;
return $res;
}
}
{
package My::HTTP::Handle;
use parent -norequire, 'HTTP::Tiny::Handle';
use Time::HiRes qw[time];
sub _do_timeout {
my ($self, $type, $timeout) = @_;
$timeout //= $self->{timeout} // 60;
my $start = time;
while (1) {
# Check for readiness NOW (0 timeout)
return 1 if $self->SUPER::_do_timeout($type, 0);
# Check for overall timeout
my $elapsed = time - $start;
return 0 if $elapsed > $timeout;
# Suspend fiber and wait for background I/O check
my $wait = ($timeout - $elapsed) > 0.5 ? 0.5 : ($timeout - $elapsed);
if ($type eq 'read') {
Acme::Parataxis->await_read($self->{fh}, int($wait * 1000));
} else {
Acme::Parataxis->await_write($self->{fh}, int($wait * 1000));
}
eg/http_tiny.pl view on Meta::CPAN
sub request ( $self, $method, $url, $args ) {
$method //= 'GET';
my %new_args = %{ $args // {} };
my $orig_cb = $new_args{data_callback};
my $content = '';
$new_args{data_callback} = sub ( $data, $response ) {
return $orig_cb->( $data, $response ) if $orig_cb;
$content .= $data;
return 1;
};
my $res = $self->SUPER::request( $method, $url, \%new_args );
$res->{content} = $content unless $orig_cb;
return $res;
}
}
#
package My::HTTP::Handle {
use parent -norequire, 'HTTP::Tiny::Handle';
use Time::HiRes qw[time];
use Acme::Parataxis qw[await_read await_write];
sub _do_timeout ( $self, $type, $timeout //= $self->{timeout} // 60 ) {
if ( $self->{fh} ) {
my $start = time;
while (1) {
# Check for readiness NOW (0 timeout)
return 1 if $self->SUPER::_do_timeout( $type, 0 );
# Check for overall timeout
my $elapsed = time() - $start;
return 0 if $elapsed > $timeout;
# Suspend fiber and wait for background I/O check.
# This is where the magic happens: the fiber yields control back
# to the scheduler while waiting for the socket to be ready.
my $wait = ( $timeout - $elapsed ) > 0.5 ? 0.5 : ( $timeout - $elapsed );
if ( $type eq 'read' ) {
await_read( $self->{fh}, int( $wait * 1000 ) );
}
else {
await_write( $self->{fh}, int( $wait * 1000 ) );
}
}
}
$self->SUPER::_do_timeout( $type, 0 );
}
}
# Use the integrated scheduler to run concurrent fetches
async {
say 'Starting concurrent fetches...';
my $http = My::HTTP->new( timeout => 10, verify_SSL => 0 );
my @urls = qw[
http://www.google.com
http://www.example.com
lib/Acme/Parataxis.pod view on Meta::CPAN
my ($self, $method, $url, $args) = @_;
my %new_args = %{ $args // {} };
my $orig_cb = $new_args{data_callback};
my $content = '';
$new_args{data_callback} = sub {
my ($data, $response) = @_;
if ($orig_cb) { return $orig_cb->($data, $response) }
$content .= $data;
return 1;
};
my $res = $self->SUPER::request($method, $url, \%new_args);
$res->{content} = $content unless $orig_cb;
return $res;
}
}
{
package My::HTTP::Handle;
use parent -norequire, 'HTTP::Tiny::Handle';
use Time::HiRes qw[time];
sub _do_timeout {
my ($self, $type, $timeout) = @_;
$timeout //= $self->{timeout} // 60;
my $start = time;
while (1) {
# Check for readiness NOW (0 timeout)
return 1 if $self->SUPER::_do_timeout($type, 0);
# Check for overall timeout
my $elapsed = time - $start;
return 0 if $elapsed > $timeout;
# Suspend fiber and wait for background I/O check
my $wait = ($timeout - $elapsed) > 0.5 ? 0.5 : ($timeout - $elapsed);
if ($type eq 'read') {
Acme::Parataxis->await_read($self->{fh}, int($wait * 1000));
} else {
Acme::Parataxis->await_write($self->{fh}, int($wait * 1000));
}
t/009_http_tiny.t view on Meta::CPAN
$args->{data_callback} = sub {
my ( $data, $response ) = @_;
# diag 'Progress: Received ' . length($data) . " bytes for $url";
if ($orig_cb) {
return $orig_cb->( $data, $response );
}
$content .= $data;
return 1;
};
my $res = $self->SUPER::request( $method, $url, $args );
$res->{content} = $content unless $orig_cb;
return $res;
}
}
{
package Acme::Parataxis::Test::HTTP::Handle;
use parent -norequire, 'HTTP::Tiny::Handle';
sub _do_timeout {
my ( $self, $type, $timeout ) = @_;
$timeout //= $self->{timeout};
if ( $self->{fh} ) {
my $start = time();
while (1) {
# Immediate check using original select (0 timeout)
return 1 if $self->SUPER::_do_timeout( $type, 0 );
# Check for overall timeout
return 0 if ( time() - $start ) > $timeout;
# Suspend fiber and wait for background I/O check.
# await_* submits a job and yields 'WAITING'.
if ( $type eq 'read' ) {
Acme::Parataxis->await_read( $self->{fh}, 500 );
}
else {
Acme::Parataxis->await_write( $self->{fh}, 500 );
}
}
}
return $self->SUPER::_do_timeout( $type, 0 );
}
}
Acme::Parataxis::run(
sub {
my $listener = IO::Socket::INET->new(
Listen => 10,
LocalAddr => '127.0.0.1', # Force IPv4
LocalPort => 0, # Auto-assign port
Proto => 'tcp',
ReuseAddr => 1,
t/013_real_http.t view on Meta::CPAN
my $orig_cb = $new_args{data_callback};
my $content = '';
$new_args{data_callback} = sub {
my ( $data, $response ) = @_;
if ($orig_cb) {
return $orig_cb->( $data, $response );
}
$content .= $data;
return 1;
};
my $res = $self->SUPER::request( $method, $url, \%new_args );
$res->{content} = $content unless $orig_cb;
return $res;
}
}
{
package Acme::Parataxis::Test::RealHTTP::Handle;
use parent -norequire, 'HTTP::Tiny::Handle';
sub _do_timeout {
my ( $self, $type, $timeout ) = @_;
$timeout //= $self->{timeout} // 60;
if ( $self->{fh} ) {
my $start = time();
while (1) {
return 1 if $self->SUPER::_do_timeout( $type, 0 );
my $elapsed = time() - $start;
return 0 if $elapsed > $timeout;
my $wait = ( $timeout - $elapsed ) > 0.5 ? 0.5 : ( $timeout - $elapsed );
if ( $type eq 'read' ) {
Acme::Parataxis->await_read( $self->{fh}, int( $wait * 1000 ) );
}
else {
Acme::Parataxis->await_write( $self->{fh}, int( $wait * 1000 ) );
}
}
}
return $self->SUPER::_do_timeout( $type, 0 );
}
}
Acme::Parataxis::run(
sub {
my $http = Acme::Parataxis::Test::RealHTTP->new( timeout => 10, verify_SSL => 0 );
my @urls = qw[
http://example.com
https://www.google.com/
https://www.perl.org/
https://metacpan.org/
t/014_http_pool.t view on Meta::CPAN
my $content = '';
$new_args{data_callback} = sub {
my ( $data, $response ) = @_;
if ($orig_cb) {
return $orig_cb->( $data, $response );
}
$content .= $data;
return 1;
};
no warnings 'uninitialized';
my $res = $self->SUPER::request( $method, $url, \%new_args );
$res->{content} = $content unless $orig_cb;
return $res;
}
}
{
package Acme::Parataxis::Test::PoolHTTP::Handle;
use parent -norequire, 'HTTP::Tiny::Handle';
sub _do_timeout {
my ( $self, $type, $timeout ) = @_;
$timeout //= $self->{timeout} // 60;
if ( $self->{fh} ) {
my $start = time();
while (1) {
return 1 if $self->SUPER::_do_timeout( $type, 0 );
my $elapsed = time() - $start;
return 0 if $elapsed > $timeout;
my $wait = ( $timeout - $elapsed ) > 0.5 ? 0.5 : ( $timeout - $elapsed );
if ( $type eq 'read' ) {
Acme::Parataxis->await_read( $self->{fh}, int( $wait * 1000 ) );
}
else {
Acme::Parataxis->await_write( $self->{fh}, int( $wait * 1000 ) );
}
}
}
return $self->SUPER::_do_timeout( $type, 0 );
}
}
Acme::Parataxis::run(
sub {
# Testing reentrancy: Use a SINGLE HTTP::Tiny object across multiple concurrent fibers
my $http = Acme::Parataxis::Test::PoolHTTP->new( timeout => 10, verify_SSL => 0 );
my @queue = qw[
http://example.com
https://www.google.com/
https://www.perl.org/
t/015_http_mock_pool.t view on Meta::CPAN
my $orig_cb = $new_args{data_callback};
my $content = '';
$new_args{data_callback} = sub {
my ( $data, $response ) = @_;
if ($orig_cb) {
return $orig_cb->( $data, $response );
}
$content .= $data;
return 1;
};
my $res = $self->SUPER::request( $method, $url, \%new_args );
$res->{content} = $content unless $orig_cb;
return $res;
}
}
package Acme::Parataxis::Test::MockPoolHTTP::Handle {
use parent -norequire, 'HTTP::Tiny::Handle';
sub _do_timeout {
my ( $self, $type, $timeout ) = @_;
$timeout //= $self->{timeout} // 60;
if ( $self->{fh} ) {
my $start = time();
while (1) {
return 1 if $self->SUPER::_do_timeout( $type, 0 );
my $elapsed = time() - $start;
return 0 if $elapsed > $timeout;
my $wait = ( $timeout - $elapsed ) > 0.5 ? 0.5 : ( $timeout - $elapsed );
if ( $type eq 'read' ) {
Acme::Parataxis->await_read( $self->{fh}, int( $wait * 1000 ) );
}
else {
Acme::Parataxis->await_write( $self->{fh}, int( $wait * 1000 ) );
}
}
}
return $self->SUPER::_do_timeout( $type, 0 );
}
}
Acme::Parataxis::run(
sub {
# Start a mock HTTP server in a fiber
my $listener = IO::Socket::INET->new( LocalAddr => '127.0.0.1', Listen => 10, Reuse => 1, Blocking => 0 ) or
die 'Could not create listener: ' . $!;
my $server_port = $listener->sockport;
diag "Mock server listening on port $server_port";
Acme::Parataxis->spawn(
( run in 0.926 second using v1.01-cache-2.11-cpan-acf6aa7dc9e )