view release on metacpan or  search on metacpan

lib/Catalyst/Engine/HTTP/POE.pm  view on Meta::CPAN

        FlushedEvent => 'client_flushed',
        ErrorEvent   => 'client_error',
        HighMark     => 128 * 1024,
        HighEvent    => sub {}, # useless, never gets called
        LowMark      => 8 * 1024,
        LowEvent     => sub {}, # also useless, we can use FlushedEvent
    );

    # get the local connection information
    my $local_sockaddr = getsockname($socket);
    my ( undef, $localiaddr ) = sockaddr_in($local_sockaddr);
    my $localaddr = inet_ntoa($localiaddr) || '127.0.0.1';
    my $localname = gethostbyaddr( $localiaddr, AF_INET ) || 'localhost';
    
    my $ID = $wheel->ID;
    
    $self->{clients}->{$ID} = {
        wheel     => $wheel,
        socket    => $socket,
        peeraddr  => $peeraddr,
        peerport  => $peerport,
        localaddr => $localaddr,
        localname => $localname,
        
        requests  => 0,
        inputbuf  => '',
        written   => 0,
        
        stopwatch => $stopwatch,
    };
    
    DEBUG && warn "[$ID] [$$] New connection (wheel $ID from $peeraddr:$peerport)\n";

    # Wait for some data to read
    $poe_kernel->select_read( $socket, 'read_input', $ID );
}

sub accept_failed {
    my ( $kernel, $self, $op, $errnum, $errstr ) = @_[ KERNEL, OBJECT, ARG0 .. ARG2 ];
    
    warn "Unable to start server: $op error $errnum: $errstr\n";
    
    $kernel->yield('shutdown');
}

sub client_error {
    my ( $kernel, $self, $op, $errnum, $errstr, $ID ) = @_[ KERNEL, OBJECT, ARG0 .. ARG3 ];
    
    DEBUG && warn "[$ID] [$$] Wheel generated $op error $errnum: $errstr\n";
    
    delete $self->{clients}->{$ID};
}

sub read_input {
    my ( $kernel, $self, $handle, $ID ) = @_[ KERNEL, OBJECT, ARG0, ARG2 ];
    
    my $client = $self->{clients}->{$ID} || return;
    
    BENCH && $client->{stopwatch}->lap('read_input');
    
    # Clear the keepalive timeout timer if set
    if ( my $timer = delete $client->{_timeout_timer} ) {
        $kernel->alarm_remove( $timer );
    }
    
    # Read some data from the driver
    my $driver = $client->{wheel}->[ $client->{wheel}->DRIVER_BOTH ];
    my $buffer_ref = $driver->get( $handle );
        
    if ( !$buffer_ref ) {
        # Error, stop reading and shut down this client
        DEBUG && warn "[$ID] [$$] Error reading, disconnecting\n";
        $kernel->select_read( $handle );
        delete $self->{clients}->{$ID};
        return;
    }
        
    $client->{inputbuf} .= join '', @{$buffer_ref};
        
    DEBUG && warn "[$ID] [$$] read_input (" . length( $client->{inputbuf} ) . " bytes in buffer)\n";
    
    $kernel->yield( 'process_input', $ID );
}

sub process_input {
    my ( $kernel, $self, $ID ) = @_[ KERNEL, OBJECT, ARG0 ];
    
    my $client = $self->{clients}->{$ID} || return;
    
    # Have we started processing the body?
    if ( exists $client->{_read} ) {
        
        _process_chunk( $client );
                
        return;
    }
    
    # Have we already parsed headers
    return if $client->{_headers};
    
    # Have we read enough to include all headers?
    if ( $client->{inputbuf} =~ /(\x0D\x0A?\x0D\x0A?|\x0A\x0D?\x0A\x0D?)/s ) {
        $client->{_headers} = 1;

        # Copy the buffer for header parsing, and remove the header block
        # from the content buffer.
        my $buf = $client->{inputbuf};
        $client->{inputbuf} =~ s/.*?(\x0D\x0A?\x0D\x0A?|\x0A\x0D?\x0A\x0D?)//s;
        
        # Parse the request line.
        if ( $buf !~ s/^(\w+)[ \t]+(\S+)(?:[ \t]+(HTTP\/\d+\.\d+))?[^\012]*\012// ) {
            # Invalid request
            DEBUG && warn "[$ID] [$$] Bad request: $buf\n";

            my $status   = 400;
            my $message  = HTTP::Status::status_message($status);
            my $response = HTTP::Response->new( $status => $message );
            $response->content_type( 'text/plain' );
            $response->content( "$status $message" );
            # XXX: fix to use CRLF
            $client->{wheel}->put( $response->as_string );
            return;
        }

lib/Catalyst/Engine/HTTP/POE.pm  view on Meta::CPAN

    # lot of outgoing data.  Don't return until it's been sent
    while ( $client && $client->{_highmark_reached} ) {
        $poe_kernel->run_one_timeslice();
    }

    # always return 1, we can't detect failures here
    return 1;
}

# client_flushed is called when all data is done being written to the browser
sub client_flushed {
    my ( $kernel, $self, $ID ) = @_[ KERNEL, OBJECT, ARG0 ];

    my $client = $self->{clients}->{$ID} || return;
    
    BENCH && $client->{stopwatch}->lap('client_flushed');

    # Are we done writing?
    if ( $client->{context} ) {
        my $cl = $client->{context}->response->content_length;
        if ( $cl && $client->{_written} >= $cl ) {
            DEBUG && warn "[$ID] [$$] client_flushed, written full content-length\n";
            $kernel->yield( 'client_done', $ID );
            return;
        }
    }

    # if we get this event because of the highmark being reached
    # don't clean up but reset the highmark value to 0
    if ( $client->{_highmark_reached} ) {
        $client->{_highmark_reached} = 0;
        return;
    }

    # we may have not had a content-length...
    $kernel->yield( 'client_done', $ID );
}

sub client_done {
    my ( $kernel, $self, $ID ) = @_[ KERNEL, OBJECT, ARG0 ];
    
    my $client = $self->{clients}->{$ID} || return;
    
    BENCH && warn "[$ID] [$$] Stopwatch:\n" . $client->{stopwatch}->stop->summary;

    # clean up everything about this client unless we are using keepalive
    if ( $client->{_keepalive} ) {
        DEBUG && warn "[$ID] [$$] client_done, keepalive enabled, waiting for more requests\n";
        $client->{requests}++;
        
        # Clear important variables from the previous state
        delete $client->{_headers};
        delete $client->{_written};
        delete $client->{_read};
        
        if ( BENCH ) {
            $client->{stopwatch} = Benchmark::Stopwatch->new->start;
        }

        # timeout idle connection after some seconds
        $client->{_timeout_timer} = $kernel->delay_set( 'keepalive_timeout', KEEPALIVE_TIMEOUT, $ID );
    }
    else {
        DEBUG && warn "[$ID] [$$] client_done, closing connection\n";
        delete $self->{clients}->{$ID};
    }
}

sub keepalive_timeout {
    my ( $kernel, $self, $ID ) = @_[ KERNEL, OBJECT, ARG0 ];
    
    DEBUG && warn "[$ID] [$$] Timing out idle keepalive connection\n";
    
    delete $self->{clients}->{$ID};
}

# Process a chunk of body data
sub _process_chunk {
    my $client = shift;
    
    # Read no more than content-length
    my $cl = $client->{env}->{CONTENT_LENGTH} || length( $client->{inputbuf} ) || 0;

    my $buf  = substr $client->{inputbuf}, 0, $cl, '';
    my $read = length($buf);
    
    return unless $read;
        
    $client->{context}->prepare_body_chunk( $buf );

    $client->{_read} += $read;
    
    if ( DEBUG ) {
        my $ID   = $client->{wheel}->ID;
        my $togo = $client->{_read_length} - $client->{_read};
        warn "[$ID] [$$] prepare_body: Read $read bytes ($togo to go)\n";
    }
    
    # Is that all the body data?
    if ( $client->{_read} >= $client->{_read_length} ) {
        # Some browsers (like MSIE 5.01) send extra CRLFs after the content
        # so we need to strip it away
        $client->{inputbuf} =~ s/^\s+//;
        
        $client->{_prepare_body_done} = 1;
    }
}

1;
__END__

=head1 NAME

Catalyst::Engine::HTTP::POE - Single-threaded multi-tasking Catalyst engine (deprecated in favor of HTTP::Prefork)

=head1 SYNOPIS

    CATALYST_ENGINE='HTTP::POE' script/yourapp_server.pl
    
    # Prefork 5 children
    CATALYST_POE_MAX_PROC=6 CATALYST_ENGINE='HTTP::POE' script/yourapp_server.pl