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

package Mojo::Pipeline;

use strict;
use warnings;

use base 'Mojo::Transaction';

use Mojo::Transaction;

__PACKAGE__->attr(txs => (chained => 1, default => sub { [] }));

# No children have ever meddled with the Republican Party and lived to tell
# about it.
sub new {
    my $self = shift->SUPER::new();
    $self->add_tx(@_);
    return $self;
}

sub add_tx {
    my $self = shift;
    push @{$self->txs}, @_;
    return $self;
}

sub client_info { shift->_proxy('client_info', @_) }

sub client_connect {
    my $self = shift;

    # Initialize
    $self->{_writer} = 0;
    $self->{_reader} = 0;

    # Connect all
    $_->client_connect for @{$self->txs};
    $self->state('connect');

    return $self;
}

sub client_connected {
    my $self = shift;

    # All connected
    $_->client_connected for @{$self->txs};
    $self->state('write_start_line');

    return $self;
}

sub client_get_chunk {
    my $self = shift;

    # Get chunk from current writer
    return $self->_writer->client_get_chunk;
}

sub client_read {
    my ($self, $chunk) = @_;

    # Read with current reader
    $self->_reader->client_read($chunk);

    # Transaction finished
    if ($self->_reader->is_finished) {

        # All done
        unless ($self->_next_reader) {

            $self->{_reader} = $#{$self->txs};
            $self->{_writer} = $#{$self->txs};

            $self->done;
            return $self;
        }
    }

    # Inherit state
    $self->_client_inherit_state;

    return $self;
}

sub client_leftovers {
    my $self = shift;

    # Previous reader
    my $previous = $self->{_reader} - 1;

    # No previous reader
    return undef unless $previous >= 0;

    # Leftovers
    return $self->txs->[$previous]->client_leftovers;
}

sub client_spin {
    my $self = shift;

    # Spin all
    $_->client_spin for @{$self->txs};

    # Transaction finished
    if (!$self->{_all_written} && $self->_writer->is_state('read_response')) {

        # All written
        $self->{_all_written} = 1 unless $self->_next_writer;
    }

    # Take care of leftovers
    if (my $leftovers = $self->client_leftovers) {
        $self->_reader->client_read($leftovers);
    }

    # Inherit state
    $self->_client_inherit_state;

    return $self;
}

sub client_written {
    my ($self, $length) = @_;

    # Written
    $self->_writer->client_written($length);

    return $self;
}

sub connection       { shift->_proxy('connection',       @_) }
sub continue_timeout { shift->_proxy('continue_timeout', @_) }
sub continued        { shift->_proxy('continued',        @_) }
sub keep_alive       { shift->_proxy('keep_alive',       @_) }
sub kept_alive       { shift->_proxy('kept_alive',       @_) }
sub local_address    { shift->_proxy('local_address',    @_) }
sub local_port       { shift->_proxy('local_port',       @_) }
sub remote_address   { shift->_proxy('remote_address',   @_) }
sub remote_port      { shift->_proxy('remote_port',      @_) }

sub req { shift->_writer->req(@_) }

sub res { shift->_writer->res(@_) }

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

    # Accept<commit_msg>cleaned up Mojo::Pipeline<commit_after># Copyright (C) 2008-2009, <NAME>.

package Mojo::Pipeline;

use strict;
use warnings;

use base 'Mojo::Transaction';

use Mojo::Transaction;

__PACKAGE__->attr(txs => (chained => 1, default => sub { [] }));

# No children have ever meddled with the Republican Party and lived to tell
# about it.
sub new {
    my $self = shift->SUPER::new();
    $self->add_tx(@_);
    return $self;
}

sub add_tx {
    my $self = shift;
    push @{$self->txs}, @_;
    return $self;
}

sub client_info { shift->_proxy('client_info', @_) }

sub client_connect {
    my $self = shift;

    # Initialize
    $self->{_writer} = 0;
    $self->{_reader} = 0;

    # Connect all
    $_->client_connect for @{$self->txs};
    $self->state('connect');

    return $self;
}

sub client_connected {
    my $self = shift;

    # All connected
    $_->client_connected for @{$self->txs};
    $self->state('write_start_line');

    return $self;
}

sub client_get_chunk {
    my $self = shift;

    # Get chunk from current writer
    return $self->_writer->client_get_chunk;
}

sub client_read {
    my ($self, $chunk) = @_;

    # Read with current reader
    $self->_reader->client_read($chunk);

    # Transaction finished
    if ($self->_reader->is_finished) {

        # All done
        unless ($self->_next_reader) {

            $self->{_reader} = $#{$self->txs};
            $self->{_writer} = $#{$self->txs};

            $self->done;
            return $self;
        }
    }

    # Inherit state
    $self->_client_inherit_state;

    return $self;
}

sub client_leftovers {
    my $self = shift;

    # Previous reader
    my $previous = $self->{_reader} - 1;

    # No previous reader
    return undef unless $previous >= 0;

    # Leftovers
    return $self->txs->[$previous]->client_leftovers;
}

sub client_spin {
    my $self = shift;

    # Spin all
    $_->client_spin for @{$self->txs};

    # Transaction finished
    if (!$self->{_all_written} && $self->_writer->is_state('read_response')) {

        # All written
        $self->{_all_written} = 1 unless $self->_next_writer;
    }

    # Take care of leftovers
    if (my $leftovers = $self->client_leftovers) {
        $self->_reader->client_read($leftovers);
    }

    # Inherit state
    $self->_client_inherit_state;

    return $self;
}

sub client_written {
    my ($self, $length) = @_;

    # Written
    $self->_writer->client_written($length);

    return $self;
}

sub connection       { shift->_proxy('connection',       @_) }
sub continue_timeout { shift->_proxy('continue_timeout', @_) }
sub continued        { shift->_proxy('continued',        @_) }
sub keep_alive       { shift->_proxy('keep_alive',       @_) }
sub kept_alive       { shift->_proxy('kept_alive',       @_) }
sub local_address    { shift->_proxy('local_address',    @_) }
sub local_port       { shift->_proxy('local_port',       @_) }
sub remote_address   { shift->_proxy('remote_address',   @_) }
sub remote_port      { shift->_proxy('remote_port',      @_) }
sub req              { shift->_proxy('req',              @_) }
sub res              { shift->_proxy('res',              @_) }

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

    # Accept
{'fraction_non_alphanumeric': 0.1564070351758794, 'fraction_numerical': 0.0043969849246231155, 'mean_word_length': 2.6332002281802622, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 123, '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': '2600810', 'n_tokens_mistral': 2171, 'n_tokens_neox': 2056, 'n_words': 469}