Acme-Parataxis
view release on metacpan or search on metacpan
## Cooperative Parallelism
This example demonstrates how to perform multiple HTTP requests concurrently on a single interpretation thread.
```perl
use Acme::Parataxis;
# ... (See My::HTTP implementation in INTEGRATING SYNCHRONOUS MODULES) ...
Acme::Parataxis::run(sub {
my $http = My::HTTP->new(verify_SSL => 0);
my @urls = qw[http://example.com http://perl.org];
# Spawn tasks for each URL
my @futures = map {
my $url = $_;
Acme::Parataxis->spawn(sub { $http->get($url)->{status} })
} @urls;
# Collect results as they become ready
say "Status for $urls[$_]: " . $futures[$_]->await( ) for 0..$#urls;
eg/http_tiny.pl view on Meta::CPAN
#
package My::HTTP {
use parent 'HTTP::Tiny';
sub _open_handle( $self, $request, $scheme, $host, $port, $peer ) {
My::HTTP::Handle->new(
timeout => $self->{timeout},
keep_alive => $self->{keep_alive},
keep_alive_timeout => $self->{keep_alive_timeout},
SSL_options => $self->{SSL_options},
verify_SSL => $self->{verify_SSL}
)->connect( $scheme, $host, $port, $peer );
}
# Override request to ensure we capture content correctly in non-blocking mode
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 ) {
eg/http_tiny.pl view on Meta::CPAN
}
}
}
$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
https://www.perl.org
https://metacpan.org
];
my @futures;
for my $url (@urls) {
push @futures, fiber {
lib/Acme/Parataxis.pod view on Meta::CPAN
=head1 EXAMPLES
=head2 Cooperative Parallelism
This example demonstrates how to perform multiple HTTP requests concurrently on a single interpretation thread.
use Acme::Parataxis;
# ... (See My::HTTP implementation in INTEGRATING SYNCHRONOUS MODULES) ...
Acme::Parataxis::run(sub {
my $http = My::HTTP->new(verify_SSL => 0);
my @urls = qw[http://example.com http://perl.org];
# Spawn tasks for each URL
my @futures = map {
my $url = $_;
Acme::Parataxis->spawn(sub { $http->get($url)->{status} })
} @urls;
# Collect results as they become ready
say "Status for $urls[$_]: " . $futures[$_]->await( ) for 0..$#urls;
t/009_http_tiny.t view on Meta::CPAN
$|++;
#
package Acme::Parataxis::Test::HTTP {
use parent 'HTTP::Tiny';
sub _open_handle {
my ( $self, $request, $scheme, $host, $port, $peer ) = @_;
my $handle = Acme::Parataxis::Test::HTTP::Handle->new(
timeout => $self->{timeout},
SSL_options => $self->{SSL_options},
verify_SSL => $self->{verify_SSL},
);
return $handle->connect( $scheme, $host, $port, $peer );
}
sub request {
my ( $self, $method, $url, $args ) = @_;
$args //= {};
my $orig_cb = $args->{data_callback};
my $content = '';
$args->{data_callback} = sub {
t/013_real_http.t view on Meta::CPAN
{
package Acme::Parataxis::Test::RealHTTP;
use parent 'HTTP::Tiny';
sub _open_handle {
my ( $self, $request, $scheme, $host, $port, $peer ) = @_;
my $handle = Acme::Parataxis::Test::RealHTTP::Handle->new(
timeout => $self->{timeout},
SSL_options => $self->{SSL_options},
verify_SSL => $self->{verify_SSL},
keep_alive => $self->{keep_alive},
keep_alive_timeout => $self->{keep_alive_timeout},
);
return $handle->connect( $scheme, $host, $port, $peer );
}
sub request {
my ( $self, $method, $url, $args ) = @_;
no warnings 'uninitialized';
$method //= 'GET';
t/013_real_http.t view on Meta::CPAN
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/
https://www.cpan.org/
https://github.com/
];
my @futures;
t/014_http_pool.t view on Meta::CPAN
{
package Acme::Parataxis::Test::PoolHTTP;
use parent 'HTTP::Tiny';
sub _open_handle {
my ( $self, $request, $scheme, $host, $port, $peer ) = @_;
my $handle = Acme::Parataxis::Test::PoolHTTP::Handle->new(
timeout => $self->{timeout},
SSL_options => $self->{SSL_options},
verify_SSL => $self->{verify_SSL},
keep_alive => $self->{keep_alive},
keep_alive_timeout => $self->{keep_alive_timeout},
);
return $handle->connect( $scheme, $host, $port, $peer );
}
sub request {
my ( $self, $method_in, $url, $args ) = @_;
my $method = "$method_in"; # Local copy
my %new_args = %{ $args // {} };
t/014_http_pool.t view on Meta::CPAN
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/
https://metacpan.org/
https://www.cpan.org/
https://github.com/
];
my %results;
my $worker_count = 3;
t/015_http_mock_pool.t view on Meta::CPAN
use POSIX ();
package Acme::Parataxis::Test::MockPoolHTTP {
use parent 'HTTP::Tiny';
sub _open_handle {
my ( $self, $request, $scheme, $host, $port, $peer ) = @_;
my $handle = Acme::Parataxis::Test::MockPoolHTTP::Handle->new(
timeout => $self->{timeout},
SSL_options => $self->{SSL_options},
verify_SSL => $self->{verify_SSL},
keep_alive => $self->{keep_alive},
keep_alive_timeout => $self->{keep_alive_timeout}
);
return $handle->connect( $scheme, $host, $port, $peer );
}
sub request {
my ( $self, $method, $url, $args ) = @_;
no warnings 'uninitialized';
$method //= 'GET';
t/015_http_mock_pool.t view on Meta::CPAN
}
}
$client->shutdown(1);
$client->close();
}
}
}
);
# Testing reentrancy: Use a SINGLE HTTP::Tiny object across multiple concurrent fibers
my $http = Acme::Parataxis::Test::MockPoolHTTP->new( timeout => 5, verify_SSL => 0 );
my $url_base = "http://127.0.0.1:$server_port/";
my @queue = ($url_base) x 10;
my @results;
my $worker_count = 3;
my @workers;
diag "Main: Starting worker pool with $worker_count fibers to process " . scalar(@queue) . " requests...";
for my $w_id ( 1 .. $worker_count ) {
push @workers, Acme::Parataxis->spawn(
sub {
( run in 1.711 second using v1.01-cache-2.11-cpan-39bf76dae61 )