<commit_before># Copyright (C) 2008, <NAME>.

package Mojo::Client;

use strict;
use warnings;

use base 'Mojo::Base';

use IO::Socket::INET;
use IO::Select;
use Mojo;
use Mojo::Loader;
use Mojo::Message::Response;

__PACKAGE__->attr(continue_timeout   => (chained => 1, default => 3));
__PACKAGE__->attr(keep_alive_timeout => (chained => 1, default => 15));
__PACKAGE__->attr(select_timeout     => (chained => 1, default => 5));

sub connect {
    my ($self, $tx) = @_;

    my $req  = $tx->req;
    my $host = $req->url->host;
    my $port = $req->url->port || 80;

    # Proxy
    if (my $proxy = $req->proxy) {
        $host = $proxy->host;
        $port = $proxy->port || 80;
    }

    # Try to get a cached connection
    my $connection = $self->withdraw_connection("$host:$port");
    $tx->kept_alive(1) if $connection;

    # Non blocking connect
    unless ($connection) {
        $connection = IO::Socket::INET->new(
            Proto => 'tcp',
            Type  => SOCK_STREAM
        );

        # Non blocking
        $connection->blocking(0);

        my $address = sockaddr_in($port, scalar inet_aton($host));
        $connection->connect($address);
        $tx->{connect_timeout} = time + 5;

    }
    $tx->connection($connection);
    $tx->state('connect');

    # Connection header
    unless ($req->headers->connection) {
        if ($tx->keep_alive || $tx->kept_alive) {
            $req->headers->connection('Keep-Alive');
        }
        else {
            $req->headers->connection('Close');
        }
    }

    # We identify ourself
    my $version = $Mojo::VERSION;
    $req->headers->user_agent("Mozilla/5.0 (compatible; Mojo/$version; Perl)")
      unless $req->headers->user_agent;

    return $tx;
}

sub disconnect {
    my ($self, $tx) = @_;

    my $req  = $tx->req;
    my $host = $req->url->host;
    my $port = $req->url->port || 80;
    my $peer = "$host:$port";

    # Deposit connection for later or kill socket
    $tx->keep_alive
      ? $self->deposit_connection($peer, $tx->connection)
      : $tx->connection(undef);

    return $tx;
}

sub deposit_connection {
    my ($self, $name, $connection, $timeout) = @_;

    # Drop connections after 30 seconds from queue
    $timeout ||= $self->keep_alive_timeout;

    $self->{_connections} ||= [];

    # Store socket if it is in a good state
    if ($self->test_connection($connection)) {
        push @{$self->{_connections}}, [$name, $connection, time + $timeout];
        return 1;
    }
    return 0;
}

# Marge, I'm going to Moe's. Send the kids to the neighbors,
# I'm coming back loaded!
sub process {
    my ($self, @transactions) = @_;

    # Parallel async io main loop... woot!
    while (1) { last if $self->spin(@transactions) }

    # Finished transactions should be returned first
    my @sorted;
    while (my $tx = shift @transactions) {
        $tx->is_state(qw/done error/)
          ? unshift(@sorted, $tx)
          : push(@sorted, $tx);
    }

    return @sorted;
}

sub process_all {
    my ($self, @transactions) = @_;

    my @finished;
    my @progress = @transactions;

    # Process until all transactions are finished
    while (1) {
        my @done = $self->process(@progress);
        @progress = ();
        for my $tx (@done) {
            $tx->is_state(qw/done error/)
              ? push(@finished, $tx)
              : push(@progress, $tx);
        }
        last unless @progress;
    }

    return @finished;
}

sub process_local {
    my ($self, $class, $tx) = @_;

    # Remote server
    if (my $authority = $ENV{MOJO_AUTHORITY}) {
        $tx->req->url->authority($authority);
        return $self->process($tx);
    }

    my $app = Mojo::Loader->load_build($class);

    $tx->req->fix_headers;
    $app->handler($tx);
    $tx->res->fix_headers;

    return $tx;
}<commit_msg>Bug fix: Quit assuming that a handler() returns the same $tx it is invoked with.     The docs in other places show that handler() must return a $tx object, but     not necessarily the same one provided to it as input.<commit_after># Copyright (C) 2008, <NAME>.

package Mojo::Client;

use strict;
use warnings;

use base 'Mojo::Base';

use IO::Socket::INET;
use IO::Select;
use Mojo;
use Mojo::Loader;
use Mojo::Message::Response;

__PACKAGE__->attr(continue_timeout   => (chained => 1, default => 3));
__PACKAGE__->attr(keep_alive_timeout => (chained => 1, default => 15));
__PACKAGE__->attr(select_timeout     => (chained => 1, default => 5));

sub connect {
    my ($self, $tx) = @_;

    my $req  = $tx->req;
    my $host = $req->url->host;
    my $port = $req->url->port || 80;

    # Proxy
    if (my $proxy = $req->proxy) {
        $host = $proxy->host;
        $port = $proxy->port || 80;
    }

    # Try to get a cached connection
    my $connection = $self->withdraw_connection("$host:$port");
    $tx->kept_alive(1) if $connection;

    # Non blocking connect
    unless ($connection) {
        $connection = IO::Socket::INET->new(
            Proto => 'tcp',
            Type  => SOCK_STREAM
        );

        # Non blocking
        $connection->blocking(0);

        my $address = sockaddr_in($port, scalar inet_aton($host));
        $connection->connect($address);
        $tx->{connect_timeout} = time + 5;

    }
    $tx->connection($connection);
    $tx->state('connect');

    # Connection header
    unless ($req->headers->connection) {
        if ($tx->keep_alive || $tx->kept_alive) {
            $req->headers->connection('Keep-Alive');
        }
        else {
            $req->headers->connection('Close');
        }
    }

    # We identify ourself
    my $version = $Mojo::VERSION;
    $req->headers->user_agent("Mozilla/5.0 (compatible; Mojo/$version; Perl)")
      unless $req->headers->user_agent;

    return $tx;
}

sub disconnect {
    my ($self, $tx) = @_;

    my $req  = $tx->req;
    my $host = $req->url->host;
    my $port = $req->url->port || 80;
    my $peer = "$host:$port";

    # Deposit connection for later or kill socket
    $tx->keep_alive
      ? $self->deposit_connection($peer, $tx->connection)
      : $tx->connection(undef);

    return $tx;
}

sub deposit_connection {
    my ($self, $name, $connection, $timeout) = @_;

    # Drop connections after 30 seconds from queue
    $timeout ||= $self->keep_alive_timeout;

    $self->{_connections} ||= [];

    # Store socket if it is in a good state
    if ($self->test_connection($connection)) {
        push @{$self->{_connections}}, [$name, $connection, time + $timeout];
        return 1;
    }
    return 0;
}

# Marge, I'm going to Moe's. Send the kids to the neighbors,
# I'm coming back loaded!
sub process {
    my ($self, @transactions) = @_;

    # Parallel async io main loop... woot!
    while (1) { last if $self->spin(@transactions) }

    # Finished transactions should be returned first
    my @sorted;
    while (my $tx = shift @transactions) {
        $tx->is_state(qw/done error/)
          ? unshift(@sorted, $tx)
          : push(@sorted, $tx);
    }

    return @sorted;
}

sub process_all {
    my ($self, @transactions) = @_;

    my @finished;
    my @progress = @transactions;

    # Process until all transactions are finished
    while (1) {
        my @done = $self->process(@progress);
        @progress = ();
        for my $tx (@done) {
            $tx->is_state(qw/done error/)
              ? push(@finished, $tx)
              : push(@progress, $tx);
        }
        last unless @progress;
    }

    return @finished;
}

sub process_local {
    my ($self, $class, $tx) = @_;

    # Remote server
    if (my $authority = $ENV{MOJO_AUTHORITY}) {
        $tx->req->url->authority($authority);
        return $self->process($tx);
    }

    my $app = Mojo::Loader->load_build($class);

    $tx->req->fix_headers;
    my $new_tx = $app->handler($tx);
    $new_tx->res->fix_headers;

    return $new_tx;
}
{'fraction_non_alphanumeric': 0.16418480336006108, 'fraction_numerical': 0.007127402316405753, 'mean_word_length': 2.520609318996416, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 143, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3468459', 'n_tokens_mistral': 2673, 'n_tokens_neox': 2545, 'n_words': 652}