lib/Mojo/Pipeline.pm t/mojolicious/templates/23.html.epl t/mojolicious/templates/layouts/default.html.epl @@ -4,6 +4,8 @@ This file documents the revision history for Perl extension Mojo. - Rewrote MojoX::Renderer, it is not backward compatible and templates need to be renamed in the following 3 part format "index.html.tt"! + - Added full HTTP 1.1 pipelining support to all Mojo layers. + - Added layout support to MojoX::Renderer. - Made render call optional. - Added format support to MojoX::Routes. - Added wildcard symbol support to MojoX::Routes and rewrote many Changes @@ -24,6 +24,7 @@ WriteMakefile( PREREQ_PM => { 'Carp' => 0, 'Cwd' => 0, + 'Data::Dumper' => 0, 'Digest::MD5' => 0, 'Encode' => 0, 'File::Basename' => 0, Makefile.PL @@ -9,7 +9,6 @@ use base 'Mojo::Base'; use IO::Socket::INET; use IO::Select; -use Mojo; use Mojo::Loader; use Mojo::Message::Response; use Socket; @@ -21,15 +20,7 @@ __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; - } + my ($host, $port) = $tx->client_info; # Try to get a cached connection my $connection = $self->withdraw_connection("$host:$port"); @@ -47,26 +38,13 @@ sub connect { my $address = sockaddr_in($port, scalar inet_aton($host)); $connection->connect($address); - $tx->{connect_timeout} = time + 5; + $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; + # State machine + $tx->client_connect; return $tx; } @@ -74,14 +52,11 @@ sub connect { sub disconnect { my ($self, $tx) = @_; - my $req = $tx->req; - my $host = $req->url->host; - my $port = $req->url->port || 80; - my $peer = "$host:$port"; + my ($host, $port) = $tx->client_info; # Deposit connection for later or kill socket $tx->keep_alive - ? $self->deposit_connection($peer, $tx->connection) + ? $self->deposit_connection("$host:$port", $tx->connection) : $tx->connection(undef); return $tx; @@ -167,30 +142,25 @@ sub spin { my $done = 0; for my $tx (@transactions) { - # Check for request/response errors - $tx->error('Request error.') if $tx->req->has_error; - $tx->error('Response error.') if $tx->res->has_error; - # Connect transaction $self->connect($tx) if $tx->is_state('start'); # Check connect status if (!$tx->connection->connected) { - if (time > $tx->{connect_timeout}) { - $tx->error("Can't connect to peer before timeout"); + if (time > $tx->{_connect_timeout}) { + $tx->error("Couldn't connect to peer before timeout."); $done++; } next; } + + # Connected elsif ($tx->is_state('connect')) { - # We might have to handle 100 Continue - $tx->{_continue} = $self->continue_timeout - if ($tx->req->headers->expect || '') =~ /100-continue/; + $tx->continue_timeout($self->continue_timeout); - # Ready for next state - $tx->state('write_start_line'); - $tx->{_to_write} = $tx->req->start_line_length; + # State machine + $tx->client_connected; # Store connection information my ($lport, $laddr) = sockaddr_in(getsockname($tx->connection)); @@ -201,40 +171,13 @@ sub spin { $tx->remote_port($rport); } + # State machine + $tx->client_spin; + # Map my $name = $self->_socket_name($tx->connection); $transaction{$name} = $tx; - # Request start line written - if ($tx->is_state('write_start_line')) { - if ($tx->{_to_write} <= 0) { - $tx->state('write_headers'); - $tx->{_offset} = 0; - $tx->{_to_write} = $tx->req->header_length; - } - } - - # Request headers written - if ($tx->is_state('write_headers')) { - if ($tx->{_to_write} <= 0) { - $tx->{_continue} - ? $tx->state('read_continue') - : $tx->state('write_body'); - $tx->{_offset} = 0; - $tx->{_to_write} = $tx->req->body_length; - } - } - - # 100 Continue timeout - if ($tx->is_state('read_continue')) { - $tx->state('write_body') unless $tx->{_continue}; - } - - # Request body written - if ($tx->is_state('write_body')) { - $tx->state('read_response') if $tx->{_to_write} <= 0; - } - # Done? if ($tx->is_done) { $done++; @@ -278,16 +221,6 @@ sub spin { IO::Select->select($read_select, $write_select, undef, $self->select_timeout); - # Make sure we don't wait longer than 5 seconds for a 100 Continue - for my $tx (@transactions) { - next unless $tx->{_continue}; - my $continue = $tx->{_continue}; - $tx->{_started} ||= time; - $continue -= time - $tx->{_started}; - $continue = 0 if $continue < 0; - $tx->{_continue} = $continue; - } - $read ||= []; $write ||= []; @@ -300,26 +233,15 @@ sub spin { # Write if ($do == 1) { - my ($tx, $req, $chunk); + my ($tx, $chunk); - # Check for content + # Check for content randomly for my $connection (sort { int(rand(3)) - 1 } @$write) { my $name = $self->_socket_name($connection); - $tx = $transaction{$name}; - $req = $tx->req; - - # Body - $chunk = $req->get_body_chunk($tx->{_offset} || 0) - if $tx->is_state('write_body'); - - # Headers - $chunk = $req->get_header_chunk($tx->{_offset} || 0) - if $tx->is_state('write_headers'); + $tx = $transaction{$name}; - # Start line - $chunk = $req->get_start_line_chunk($tx->{_offset} || 0) - if $tx->is_state('write_start_line'); + $chunk = $tx->client_get_chunk; # Content generator ready? last if defined $chunk; @@ -327,11 +249,10 @@ sub spin { # Write chunk my $written = $tx->connection->syswrite($chunk, length $chunk); - $tx->error("Can't write request: $!") unless defined $written; + $tx->error("Can't write to socket: $!") unless defined $written; return 1 if $tx->has_error; - $tx->{_to_write} -= $written; - $tx->{_offset} += $written; + $tx->client_written($written); } # Read @@ -340,43 +261,13 @@ sub spin { my $connection = $read->[rand(@$read)]; my $name = $self->_socket_name($connection); my $tx = $transaction{$name}; - my $res = $tx->res; - - # Early response, most likely an error - $tx->state('read_response') - if $tx->is_state(qw/write_start_line write_headers write_body/); my $buffer; my $read = $connection->sysread($buffer, 1024, 0); $tx->error("Can't read from socket: $!") unless defined $read; return 1 if $tx->has_error; - # Read 100 Continue - if ($tx->is_state('read_continue')) { - $res->done if $read == 0; - $res->parse($buffer); - - # We got a 100 Continue response - if ($res->is_done && $res->code == 100) { - $tx->res(Mojo::Message::Response->new); - $tx->continued(1); - $tx->{_continue} = 0; - } - - # We got something else - elsif ($res->is_done) { - $tx->res($res); - $tx->continued(0); - $tx->done; - } - } - - # Read response - elsif ($tx->is_state('read_response')) { - $tx->done if $read == 0; - $res->parse($buffer); - $tx->done if $res->is_done; - } + $tx->client_read($buffer); } return $done; lib/Mojo/Client.pm @@ -36,7 +36,8 @@ __PACKAGE__->attr( default => sub { Mojo::Headers->new } ) ); -__PACKAGE__->attr(raw_header_length => (chained => 1, default => 0)); +__PACKAGE__->attr( + [qw/raw_header_length relaxed/] => (chained => 1, default => 0)); sub build_body { my $self = shift; @@ -125,6 +126,16 @@ sub is_multipart { return $type =~ /multipart/i ? 1 : 0; } +sub leftovers { + my $self = shift; + + # Chunked leftovers are in the filter buffer + return $self->filter_buffer->to_string if $self->is_chunked; + + # Normal leftovers + return $self->buffer->to_string; +} + sub parse { my $self = shift; @@ -177,15 +188,29 @@ sub parse { return Mojo::Content::MultiPart->new($self)->parse; } - # Parse body - $self->file->add_chunk($self->buffer->empty); + # Chunked body or relaxed content + if ($self->is_chunked || $self->relaxed) { + $self->file->add_chunk($self->buffer->empty); + } + + # Normal body + else { - # Done - unless ($self->is_chunked) { + # Slurp my $length = $self->headers->content_length || 0; + my $need = $length - $self->file->length; + $self->file->add_chunk($self->buffer->remove($need)) if $need > 0; + + # Done $self->done if $length <= $self->raw_body_length; } + # With leftovers, maybe pipelined + if ($self->is_done) { + $self->state('done_with_leftovers') + if $self->buffer->length || $self->filter_buffer->length; + } + return $self; } @@ -306,6 +331,11 @@ implements the following new ones. my $raw_body_length = $content->raw_body_length; +=head2 C<relaxed> + + my $relaxed = $content->relaxed; + $content = $content->relaxed(1); + =head1 METHODS L<Mojo::Content> inherits all methods from L<Mojo::Stateful> and implements @@ -339,6 +369,10 @@ the following new ones. my $multipart = $content->is_multipart; +=head2 C<leftovers> + + my $bytes = $content->leftovers; + =head2 C<parse> $content = $content->parse("Content-Length: 12\r\n\r\nHello World!"); lib/Mojo/Content.pm @@ -206,14 +206,15 @@ sub _parse_multipart_boundary { # Begin if (index($self->buffer->{buffer}, "\x0d\x0a--$boundary\x0d\x0a") == 0) { substr $self->buffer->{buffer}, 0, length($boundary) + 6, ''; - push @{$self->parts}, Mojo::Content->new; + push @{$self->parts}, Mojo::Content->new(relaxed => 1); $self->state('multipart_body'); return 1; } # End - if (index($self->buffer->{buffer}, "\x0d\x0a--$boundary--") == 0) { - $self->buffer->empty; + my $end = "\x0d\x0a--$boundary--"; + if (index($self->buffer->{buffer}, $end) == 0) { + $self->buffer->{buffer} =~ s/^$end//; $self->done; } lib/Mojo/Content/MultiPart.pm @@ -75,7 +75,6 @@ sub parse { # Done else { $self->_remove_chunked_encoding; - $filter->empty; $self->done; } last; lib/Mojo/Filter/Chunked.pm @@ -270,6 +270,8 @@ sub is_chunked { shift->content->is_chunked } sub is_multipart { shift->content->is_multipart } +sub leftovers { shift->content->leftovers } + sub param { my $self = shift; $self->{body_params} ||= $self->body_params; @@ -287,16 +289,27 @@ sub parse { $self->parser_progress_cb->($self) if $self->parser_progress_cb; # Content - if ($self->is_state(qw/content done/)) { + if ($self->is_state(qw/content done done_with_leftovers/)) { my $content = $self->content; + + # HTTP 0.9 has no headers $content->state('body') if $self->version eq '0.9'; + + # Parse $content->filter_buffer($self->buffer); $self->content($content->parse); + + # HTTP 0.9 has no defined length + $content->state('done') if $self->version eq '0.9'; } # Done $self->done if $self->content->is_done; + # Done with leftovers, maybe pipelined + $self->state('done_with_leftovers') + if $self->content->is_state('done_with_leftovers'); + return $self; } @@ -529,6 +542,13 @@ Defaults to 1. L<Mojo::Message> inherits all methods from L<Mojo::Stateful> and implements the following new ones. +=head2 C<at_least_version> + + my $success = $message->at_least_version('1.1'); + +Returns true if the HTTP version is greater than or equal to the version +passed in. + =head2 C<body> my $string = $message->body; @@ -608,12 +628,9 @@ followed HTTP version are set. my $is_multipart = $message->is_multipart; -=head2 C<at_least_version> +=head2 C<leftovers> - my $success = $message->at_least_version('1.1'); - -Returns true if the HTTP version is greater than or equal to the version -passed in. + my $bytes = $message->leftovers; =head2 C<param> lib/Mojo/Message.pm @@ -252,6 +252,10 @@ sub _parse_start_line { $self->major_version(0); $self->minor_version(9); $self->done; + + # HTTP 0.9 has no headers or body and does not support + # pipelining + $self->buffer->empty; } } else { $self->error('Parser error: Invalid request line') } lib/Mojo/Message/Request.pm @@ -158,6 +158,7 @@ sub _parse_start_line { $self->major_version(0); $self->minor_version(9); $self->state('content'); + $self->content->relaxed(1); return 1; } } lib/Mojo/Message/Response.pm @@ -10,6 +10,7 @@ use base 'Mojo::Server'; use Carp 'croak'; use IO::Select; use IO::Socket; +use Mojo::Pipeline; __PACKAGE__->attr(keep_alive_timeout => (chained => 1, default => 15)); __PACKAGE__->attr(listen_queue_size => (chained => 1, default => SOMAXCONN)); @@ -126,7 +127,7 @@ sub _prepare_select { my $connection = $self->{_connections}->{$name}; # Transaction - my $tx = $connection->{tx}; + my $p = $connection->{pipeline}; # Keep alive timeout my $timeout = time - $connection->{time}; @@ -136,7 +137,7 @@ sub _prepare_select { } # No transaction in progress - unless ($tx) { + unless ($p) { # Keep alive request limit if ($connection->{requests} >= $self->max_keep_alive_requests) { @@ -149,12 +150,12 @@ sub _prepare_select { next; } - # Read - if ($tx->is_state('read')) { unshift @read, $tx->connection } + # We always try to read as sugegsted by the HTTP spec + unshift @read, $p->connection; # Write - if ($tx->is_state(qw/write_start_line write_headers write_body/)) { - unshift @write, $tx->connection; + if ($p->is_state(qw/write_start_line write_headers write_body/)) { + unshift @write, $p->connection; } } @@ -179,58 +180,57 @@ sub _prepare_transactions { } # Transaction - my $tx = $connection->{tx}; + my $p = $connection->{pipeline}; # Just a keep alive, no transaction - next unless $tx; - - # Writing - if ($tx->is_state('write')) { - - # Connection header - unless ($tx->res->headers->connection) { - if ($tx->keep_alive) { - $tx->res->headers->connection('Keep-Alive'); - } - else { - $tx->res->headers->connection('Close'); - } - } + next unless $p; - # Ready for next state - $tx->state('write_start_line'); - $tx->{_to_write} = $tx->res->start_line_length; - } + # Expect 100 Continue? + if ($p->is_state('handle_continue')) { + + # Continue handler + $self->continue_handler_cb->($self, $p->server_tx); - # Response start line - if ($tx->is_state('write_start_line') && $tx->{_to_write} <= 0) { - $tx->state('write_headers'); - $tx->{_offset} = 0; - $tx->{_to_write} = $tx->res->header_length; + # Handled + $p->server_handled; } - # Response headers - if ($tx->is_state('write_headers') && $tx->{_to_write} <= 0) { - $tx->state('write_body'); - $tx->{_offset} = 0; - $tx->{_to_write} = $tx->res->body_length; + # EOF + if ($p->is_state('handle_request')) { + + # Handler + $self->handler_cb->($self, $p->server_tx); + + # Handled + $p->server_handled; } - # Response body - if ($tx->is_state('write_body') && $tx->{_to_write} <= 0) { + # State machine + $p->server_spin; - # Continue done - if (defined $tx->continued && $tx->continued == 0) { - $tx->continued(1); - $tx->state('read'); - $tx->done unless $tx->res->code == 100; - $tx->res->code(0); - next; - } + # Add transactions to the pipe for leftovers + if (my $leftovers = $p->server_leftovers) { + + # New transaction + my $tx = $self->build_tx_cb->($self); + $tx->local_address($p->local_address); + $tx->local_port($p->local_port); + $tx->remote_address($p->remote_address); + $tx->remote_port($p->remote_port); + + # Add to pipeline + $p->server_accept($tx); + + # Read leftovers + $p->server_read($leftovers); + } + + # Pipeline finished? + elsif ($p->is_finished) { - # Done - delete $connection->{tx}; - $self->_drop_connection($name) unless $tx->keep_alive; + # Drop + delete $connection->{pipeline}; + $self->_drop_connection($name) unless $p->keep_alive; } } } @@ -256,28 +256,29 @@ sub _read { return 0 unless my $name = $self->_socket_name($socket); my $connection = $self->{_connections}->{$name}; - unless ($connection->{tx}) { - my $tx = $connection->{tx} ||= $self->build_tx_cb->($self); - $tx->connection($socket); - $tx->state('read'); + unless ($connection->{pipeline}) { + + # New pipeline + my $p = $connection->{pipeline} + ||= Mojo::Pipeline->new->server_accept($self->build_tx_cb->($self)); + $p->connection($socket); $connection->{requests}++; - $tx->kept_alive(1) if $connection->{requests} > 1; + $p->kept_alive(1) if $connection->{requests} > 1; # Last keep alive request? - $tx->res->headers->connection('Close') + $p->server_tx->res->headers->connection('Close') if $connection->{requests} >= $self->max_keep_alive_requests; # Store connection information - my ($lport, $laddr) = sockaddr_in(getsockname($tx->connection)); - $tx->local_address(inet_ntoa($laddr)); - $tx->local_port($lport); - my ($rport, $raddr) = sockaddr_in(getpeername($tx->connection)); - $tx->remote_address(inet_ntoa($raddr)); - $tx->remote_port($rport); + my ($lport, $laddr) = sockaddr_in(getsockname($p->connection)); + $p->local_address(inet_ntoa($laddr)); + $p->local_port($lport); + my ($rport, $raddr) = sockaddr_in(getpeername($p->connection)); + $p->remote_address(inet_ntoa($raddr)); + $p->remote_port($rport); } - my $tx = $connection->{tx}; - my $req = $tx->req; + my $p = $connection->{pipeline}; # Read request my $read = $socket->sysread(my $buffer, 4096, 0); @@ -288,25 +289,8 @@ sub _read { return 1; } - # Parse - $req->parse($buffer); - - # Expect 100 Continue? - if ($req->content->is_state('body') && !defined $tx->continued) { - if (($req->headers->expect || '') =~ /100-continue/i) { - $tx->state('write'); - $tx->continued(0); - $self->continue_handler_cb->($self, $tx); - } - } - - # EOF - if (($read == 0) || $req->is_finished) { - $tx->state('write'); - - # Handle - $self->handler_cb->($self, $tx); - } + # Read + $p->server_read($buffer); $connection->{time} = time; } @@ -330,26 +314,15 @@ sub _socket_name { sub _write { my ($self, $sockets) = @_; - my ($name, $tx, $res, $chunk); + my ($name, $p, $chunk); # Check for content for my $socket (@$sockets) { next unless $name = $self->_socket_name($socket); my $connection = $self->{_connections}->{$name}; - $tx = $connection->{tx}; - $res = $tx->res; - - # Body - $chunk = $res->get_body_chunk($tx->{_offset} || 0) - if $tx->is_state('write_body'); + $p = $connection->{pipeline}; - # Headers - $chunk = $res->get_header_chunk($tx->{_offset} || 0) - if $tx->is_state('write_headers'); - - # Start line - $chunk = $res->get_start_line_chunk($tx->{_offset} || 0) - if $tx->is_state('write_start_line'); + $chunk = $p->server_get_chunk; # Content generator ready? last if defined $chunk; @@ -357,13 +330,13 @@ sub _write { return 0 unless $name; # Write chunk - return 0 unless $tx->connection->connected; - my $written = $tx->connection->syswrite($chunk, length $chunk); - $tx->error("Can't write request: $!") unless defined $written; - return 1 if $tx->has_error; + return 0 unless $p->connection->connected; + + my $written = $p->connection->syswrite($chunk, length $chunk); + $p->error("Can't write request: $!") unless defined $written; + return 1 if $p->has_error; - $tx->{_to_write} -= $written; - $tx->{_offset} += $written; + $p->server_written($written); $self->{_connections}->{$name}->{time} = time; } lib/Mojo/Server/Daemon.pm @@ -24,7 +24,7 @@ sub has_error { return defined shift->{error} } sub is_done { return shift->state eq 'done' } -sub is_finished { return shift->is_state(qw/done error/) } +sub is_finished { return shift->is_state(qw/done done_with_leftovers error/) } sub is_state { my ($self, @states) = @_; lib/Mojo/Stateful.pm @@ -7,13 +7,14 @@ use warnings; use base 'Mojo::Stateful'; +use Mojo; use Mojo::Message::Request; use Mojo::Message::Response; __PACKAGE__->attr( [ qw/ - continued connection + continued kept_alive local_address local_port @@ -22,6 +23,7 @@ __PACKAGE__->attr( / ] => (chained => 1) ); +__PACKAGE__->attr(continue_timeout => (chained => 1, default => 3)); __PACKAGE__->attr( req => ( chained => 1, @@ -37,10 +39,200 @@ __PACKAGE__->attr( # What's a wedding? Webster's dictionary describes it as the act of removing # weeds from one's garden. +sub client_connect { + my $self = shift; + + # Connect + $self->state('connect'); + + # Connection header + unless ($self->req->headers->connection) { + if ($self->keep_alive || $self->kept_alive) { + $self->req->headers->connection('Keep-Alive'); + } + else { + $self->req->headers->connection('Close'); + } + } + + # We identify ourself + my $version = $Mojo::VERSION; + $self->req->headers->user_agent( + "Mozilla/5.0 (compatible; Mojo/$version; Perl)") + unless $self->req->headers->user_agent; + + return $self; +} + +sub client_connected { + my $self = shift; + + # We might have to handle 100 Continue + $self->{_continue} = $self->continue_timeout + if ($self->req->headers->expect || '') =~ /100-continue/; + + # Ready for next state + $self->state('write_start_line'); + $self->{_to_write} = $self->req->start_line_length; + + return $self; +} + +sub client_get_chunk { + my $self = shift; + + my $chunk; + + # Body + $chunk = $self->req->get_body_chunk($self->{_offset} || 0) + if $self->is_state('write_body'); + + # Headers + $chunk = $self->req->get_header_chunk($self->{_offset} || 0) + if $self->is_state('write_headers'); + + # Start line + $chunk = $self->req->get_start_line_chunk($self->{_offset} || 0) + if $self->is_state('write_start_line'); + + return $chunk; +} + +sub client_info { + my $self = shift; + + my $host = $self->req->url->host; + my $port = $self->req->url->port || 80; + + # Proxy + if (my $proxy = $self->req->proxy) { + $host = $proxy->host; + $port = $proxy->port || 80; + } + + return ($host, $port); +} + +sub client_leftovers { + my $self = shift; + + # No leftovers + return undef unless $self->is_state('done_with_leftovers'); + + # Leftovers + my $leftovers = $self->res->leftovers; + $self->done; + + return $leftovers; +} + +sub client_read { + my ($self, $chunk) = @_; + + # Length + my $read = length $chunk; + + # Early response, most likely an error + $self->state('read_response') + if $self->is_state(qw/write_start_line write_headers write_body/); + + # Read 100 Continue + if ($self->is_state('read_continue')) { + $self->res->done if $read == 0; + $self->res->parse($chunk); + + # We got a 100 Continue response + if ($self->res->is_done && $self->res->code == 100) { + $self->res($self->res->new); + $self->continued(1); + $self->{_continue} = 0; + } + + # We got something else + elsif ($self->res->is_done) { + $self->continued(0); + $self->done; + } + } + + # Read response + elsif ($self->is_state('read_response')) { + $self->done if $read == 0; + $self->res->parse($chunk); + $self->done if $self->res->is_done; + $self->state('done_with_leftovers') + if $self->res->is_state('done_with_leftovers'); + } + + return $self; +} + +sub client_spin { + my $self = shift; + + # Check for request/response errors + $self->error('Request error.') if $self->req->has_error; + $self->error('Response error.') if $self->res->has_error; + + # Make sure we don't wait longer than 5 seconds for a 100 Continue + if ($self->{_continue}) { + my $continue = $self->{_continue}; + $self->{_started} ||= time; + $continue -= time - $self->{_started}; + $continue = 0 if $continue < 0; + $self->{_continue} = $continue; + } + + # Request start line written + if ($self->is_state('write_start_line')) { + if ($self->{_to_write} <= 0) { + $self->state('write_headers'); + $self->{_offset} = 0; + $self->{_to_write} = $self->req->header_length; + } + } + + # Request headers written + if ($self->is_state('write_headers')) { + if ($self->{_to_write} <= 0) { + $self->{_continue} + ? $self->state('read_continue') + : $self->state('write_body'); + $self->{_offset} = 0; + $self->{_to_write} = $self->req->body_length; + } + } + + # 100 Continue timeout + if ($self->is_state('read_continue')) { + $self->state('write_body') unless $self->{_continue}; + } + + # Request body written + if ($self->is_state('write_body')) { + $self->state('read_response') if $self->{_to_write} <= 0; + } + + return $self; +} + +sub client_written { + my ($self, $written) = @_; + + # Written + $self->{_to_write} -= $written; + $self->{_offset} += $written; + + return $self; +} + sub keep_alive { my ($self, $keep_alive) = @_; - $self->{keep_alive} = $keep_alive if $keep_alive; + if ($keep_alive) { + $self->{keep_alive} = $keep_alive; + return $self; + } my $req = $self->req; my $res = $self->res; @@ -74,6 +266,146 @@ sub new_head { shift->_builder('HEAD', @_) } sub new_post { shift->_builder('POST', @_) } sub new_put { shift->_builder('PUT', @_) } +sub server_accept { + my $self = shift; + + # Reading + $self->state('read'); + + return $self; +} + +sub server_get_chunk { + my $self = shift; + + my $chunk; + + # Body + $chunk = $self->res->get_body_chunk($self->{_offset} || 0) + if $self->is_state('write_body'); + + # Headers + $chunk = $self->res->get_header_chunk($self->{_offset} || 0) + if $self->is_state('write_headers'); + + # Start line + $chunk = $self->res->get_start_line_chunk($self->{_offset} || 0) + if $self->is_state('write_start_line'); + + return $chunk; +} + +sub server_handled { + my $self = shift; + + # Handled and writing now + $self->state('write'); + + return $self; +} + +sub server_read { + my ($self, $chunk) = @_; + + # Parse + $self->req->parse($chunk); + + # Expect 100 Continue? + if ($self->req->content->is_state('body') && !defined $self->continued) { + if (($self->req->headers->expect || '') =~ /100-continue/i) { + $self->state('handle_continue'); + $self->continued(0); + } + } + + # EOF + if ((length $chunk == 0) || $self->req->is_finished) { + $self->state('handle_request'); + } + + return $self; +} + +sub server_spin { + my $self = shift; + + # Initialize + $self->{_to_write} ||= 0; + + # Writing + if ($self->is_state('write')) { + + # Connection header + unless ($self->res->headers->connection) { + if ($self->keep_alive) { + $self->res->headers->connection('Keep-Alive'); + } + else { + $self->res->headers->connection('Close'); + } + } + + # Ready for next state + $self->state('write_start_line'); + $self->{_to_write} = $self->res->start_line_length; + } + + # Response start line + if ($self->is_state('write_start_line') && $self->{_to_write} <= 0) { + $self->state('write_headers'); + $self->{_offset} = 0; + $self->{_to_write} = $self->res->header_length; + } + + # Response headers + if ($self->is_state('write_headers') && $self->{_to_write} <= 0) { + $self->state('write_body'); + $self->{_offset} = 0; + $self->{_to_write} = $self->res->body_length; + } + + # Response body + if ($self->is_state('write_body') && $self->{_to_write} <= 0) { + + # Continue done + if (defined $self->continued && $self->continued == 0) { + $self->continued(1); + $self->state('read'); + + # Continue + if ($self->res->code == 100) { + $self->res($self->res->new); + } + + # Don't continue + else { $self->done } + } + + # Everything done + elsif (!defined $self->continued) { + $self->req->is_state('done_with_leftovers') + ? $self->state('done_with_leftovers') + : $self->state('done'); + } + } + + return $self; +} + +sub server_written { + my ($self, $written) = @_; + + # Written + $self->{_to_write} -= $written; + $self->{_offset} += $written; + + # Done early + $self->state('done') + if $self->is_state('write_body') && $self->{_to_write} <= 0; + + return $self; +} + sub _builder { my $class = shift; my $self = $class->new; @@ -126,6 +458,11 @@ implements the following new ones. my $connection = $tx->connection; $tx = $tx->connection($connection); +=head2 C<continue_timeout> + + my $continue_timeout = $tx->continue_timeout; + $tx = $tx->continue_timeout(5); + =head2 C<continued> my $continued = $tx->continued; @@ -134,7 +471,7 @@ implements the following new ones. =head2 C<keep_alive> my $keep_alive = $tx->keep_alive; - my $keep_alive = $tx->keep_alive(1); + $tx = $tx->keep_alive(1); =head2 C<kept_alive> @@ -182,6 +519,38 @@ Returns the invocant if called with arguments. L<Mojo::Transaction> inherits all methods from L<Mojo::Stateful> and implements the following new ones. +=head2 C<client_connect> + + $tx = $tx->client_connect; + +=head2 C<client_connected> + + $tx = $tx->client_connected; + +=head2 C<client_get_chunk> + + my $chunk = $tx->client_get_chunk; + +=head2 C<client_info> + + my ($host, $port) = $tx->client_info; + +=head2 C<client_leftovers> + + my $leftovers = $tx->client_leftovers; + +=head2 C<client_read> + + $tx = $tx->client_read($chunk); + +=head2 C<client_spin> + + $tx = $tx->client_spin; + +=head2 C<client_written> + + $tx = $tx->client_written($length); + =head2 C<new_delete> my $tx = Mojo::Transaction->new_delete('http://127.0.0.1', @@ -227,5 +596,28 @@ implements the following new ones. User-Agent => 'Mojo' }); +=head2 C<server_accept> + + $tx = $tx->server_accept; + +=head2 C<server_get_chunk> + + my $chunk = $tx->server_get_chunk; + +=head2 C<server_handled> + + $tx = $tx->server_handled; + +=head2 C<server_read> + + $tx = $tx->server_read($chunk); + +=head2 C<server_spin> + + $tx = $tx->server_spin; + +=head2 C<server_written> + + $tx = $tx->server_written($bytes); =cut lib/Mojo/Transaction.pm @@ -11,8 +11,6 @@ __PACKAGE__->attr(app => (chained => 1, weak => 1)); __PACKAGE__->attr(tx => (chained => 1)); # This is my first visit to the Galaxy of Terror and I'd like it to be a pleasant one. -sub render { } - sub req { return shift->tx->req } sub res { return shift->tx->res } @@ -94,11 +92,6 @@ Returns the invocant if called with a hashref or multiple arguments. =head1 METHODS -L<MojoX::Context> inherits all methods from L<Mojo::Base> and implements the -following new ones. - -=head2 C<render> - - $c->render; +L<MojoX::Context> inherits all methods from L<Mojo::Base>. =cut lib/MojoX/Context.pm @@ -13,7 +13,7 @@ use Mojo::Loader; __PACKAGE__->attr( disallow => ( chained => 1, - default => sub { [qw/new app attr render req res stash/] } + default => sub { [qw/new app attr meta render req res stash/] } ) ); __PACKAGE__->attr(namespace => (chained => 1)); lib/MojoX/Dispatcher/Routes.pm @@ -11,6 +11,7 @@ __PACKAGE__->attr(match => (chained => 1)); # Just make a simple cake. And this time, if someone's going to jump out of # it make sure to put them in *after* you cook it. +sub render { } 1; __END__ @@ -41,6 +42,10 @@ L<MojoX::Context> and implements the following new ones. =head1 METHODS L<MojoX::Dispatcher::Routes::Context> inherits all methods from -L<MojoX::Context>. +L<MojoX::Context> and implements the following new ones. + +=head2 C<render> + + $c->render; =cut lib/MojoX/Dispatcher/Routes/Context.pm @@ -36,67 +36,49 @@ sub add_handler { sub render { my ($self, $c) = @_; - my $format = $c->stash->{format}; - my $handler = $c->stash->{handler}; - my $template = $c->stash->{template}; - my $template_path = $c->stash->{template_path}; + my ($template, $template_path); + my $is_layout = 0; + + # Layout first + if ($c->stash->{layout} || $c->stash->{layout_path}) { + $template = $c->stash->{layout}; + $template_path = $c->stash->{layout_path}; + $is_layout = 1; + } + + # Normal template + else { + $template = $c->stash->{template}; + $template_path = $c->stash->{template_path}; + } - # Not enough informations + # Not enough information return undef unless $template || $template_path; # Handler precedence $self->precedence([sort keys %{$self->handler}]) unless $self->precedence; - # Template path - $template_path = File::Spec->catfile($self->root, $template) - if $template && !$template_path; - - # Format - unless ($template_path =~ /\.\w+(?:\.\w+)?$/) { - if ($format) { $template_path .= ".$format" } - else { - $c->app->log->debug('Template format missing'); - return undef; - } - } + # Inner + local $c->stash->{inner_template} = $c->stash->{template} if $is_layout; + local $c->stash->{inner_template_path} = $c->stash->{template_path} + if $is_layout; - # Handler - unless ($template_path =~ /\.\w+\.\w+$/) { - - if ($handler) { $template_path .= ".$handler" } - - # Detect - else { - my $found = 0; - for my $ext (@{$self->precedence}) { - - # Try - my $path = "$template_path.$ext"; - if (-f $path) { - $found++; - $template_path = $path; - $c->app->log->debug(qq/Template found "$template_path"/); - last; - } - } + # Template has priority + $template_path = undef if $template; - # Nothing found - unless ($found) { - $c->app->log->debug( - qq/Template not found "$template_path.*"/); - return undef; - } - } - } + # Path + return undef + unless $template_path = + $self->_fix_path($c, $template, $template_path, $is_layout); # Store for handler usage local $c->stash->{template_path} = $template_path; # Extract $template_path =~ /\.(\w+)\.(\w+)$/; - $format = $1; - $handler = $2; + my $format = $1; + my $handler = $2; # Renderer my $r = $self->handler->{$handler}; @@ -110,6 +92,11 @@ sub render { # Partial? my $partial = $c->stash->{partial}; + # Clean + local $c->stash->{layout} = undef; + local $c->stash->{layout_path} = undef; + local $c->stash->{template} = undef; + # Render my $output; return undef unless $r->($self, $c, \$output); @@ -132,6 +119,84 @@ sub render { return 1; } +sub _fix_format { + my ($self, $c, $path) = @_; + + # Format ok + return $path if $path =~ /\.\w+(?:\.\w+)?$/; + + my $format = $c->stash->{format}; + + # Append format + if ($format) { $path .= ".$format" } + + # Missing format + else { + $c->app->log->debug('Template format missing'); + return undef; + } + + return $path; +} + +sub _fix_handler { + my ($self, $c, $path) = @_; + + # Handler ok + return $path if $path =~ /\.\w+\.\w+$/; + + my $handler = $c->stash->{handler}; + + # Append handler + if ($handler) { $path .= ".$handler" } + + # Detect + else { + my $found = 0; + for my $ext (@{$self->precedence}) { + + # Try + my $p = "$path.$ext"; + if (-f $p) { + $found++; + $path = $p; + $c->app->log->debug(qq/Template found "$path"/); + last; + } + } + + # Nothing found + unless ($found) { + $c->app->log->debug(qq/Template not found "$path.*"/); + return undef; + } + } + + return $path; +} + +sub _fix_path { + my ($self, $c, $template, $template_path, $is_layout) = @_; + + # Root + my $root = + $is_layout ? File::Spec->catfile($self->root, 'layouts') : $self->root; + + # Path + $template_path = File::Spec->catfile($root, $template) + if $template && !$template_path; + + # Format + return undef + unless $template_path = $self->_fix_format($c, $template_path); + + # Handler + return undef + unless $template_path = $self->_fix_handler($c, $template_path); + + return $template_path; +} + 1; __END__ lib/MojoX/Renderer.pm @@ -19,7 +19,7 @@ sub render { $self->{stash} = {%{$self->stash}, %$args}; # Template - unless ($self->stash->{template}) { + unless ($self->stash->{template} || $self->stash->{template_path}) { # Default template my $controller = $self->stash->{controller}; @@ -38,6 +38,19 @@ sub render { return $self->app->renderer->render($self); } +sub render_inner { + my $self = shift; + local $self->stash->{template} = $self->stash->{inner_template}; + local $self->stash->{template_path} = $self->stash->{inner_template_path}; + return $self->render_partial(@_); +} + +sub render_partial { + my $self = shift; + local $self->stash->{partial} = 1; + return $self->render(@_); +} + sub url_for { my $self = shift; @@ -90,6 +103,16 @@ L<MojoX::Dispatcher::Routes::Context> and implements the following new ones. $c->render; $c->render(action => 'foo'); +=head2 C<render_inner> + + $c->render_inner; + $c->render_inner(action => 'foo'); + +=head2 C<render_partial> + + $c->render_partial; + $c->render_partial(action => 'foo'); + =head2 C<url_for> my $url = $c->url_for; lib/Mojolicious/Context.pm @@ -48,10 +48,12 @@ sub run { $self->render_to_rel_file('404', "$name/public/404.html"); $self->render_to_rel_file('static', "$name/public/index.html"); - # Template + # Layout and Template $self->renderer->line_start('%%'); $self->renderer->tag_start('<%%'); $self->renderer->tag_end('%%>'); + $self->render_to_rel_file('layout', + "$name/templates/layouts/default.html.epl"); $self->render_to_rel_file('welcome', "$name/templates/example/welcome.html.epl"); } @@ -180,8 +182,11 @@ use base 'Mojolicious::Controller'; sub welcome { my $self = shift; - # Render template "example/welcome.html.epl" with welcome message - $self->render(message => 'Welcome to the Mojolicious Web Framework!'); + # Render template "example/welcome.html.epl" with message and layout + $self->render( + layout => 'default', + message => 'Welcome to the Mojolicious Web Framework!' + ); } 1; @@ -229,21 +234,21 @@ $client->process_local('<%= $class %>', $tx); is($tx->res->code, 200); is($tx->res->headers->content_type, 'text/html'); like($tx->res->content->file->slurp, qr/Mojolicious Web Framework/i); -__welcome__ +__layout__ % my $self = shift; <!doctype html> - <head><title><%= $self->stash('message') %></title></head> + <head><title>Welcome</title></head> <body> - <h2><%= $self->stash('message') %></h2> - This page was generated from the template - "templates/example/welcome.html.epl", - <a href="<%= $self->url_for %>"> - click here - </a> - to reload the page or - <a href="/index.html"> - here - </a> - to move forward to a static page. + <%= $self->render_inner %> </body> </html> +__welcome__ +% my $self = shift; +<h2><%= $self->stash('message') %></h2> +This page was generated from the template +"templates/example/welcome.html.epl" and the layout +"templates/layouts/default.html.epl", +<a href="<%= $self->url_for %>">click here</a> +to reload the page or +<a href="/index.html">here</a> +to move forward to a static page. lib/Mojolicious/Script/Generate/App.pm @@ -9,11 +9,12 @@ use Test::More; plan skip_all => 'set TEST_CLIENT to enable this test' unless $ENV{TEST_CLIENT}; -plan tests => 16; +plan tests => 34; # So then I said to the cop, "No, you're driving under the influence... # of being a jerk". use_ok('Mojo::Client'); +use_ok('Mojo::Pipeline'); use_ok('Mojo::Transaction'); # Parallel async io @@ -62,3 +63,35 @@ ok($tx->local_address); ok($tx->local_port > 0); is($tx->remote_address, '88.198.25.164'); is($tx->remote_port, 80); + +# Pipelined +$tx = Mojo::Transaction->new_get('http://labs.kraih.com'); +$tx2 = Mojo::Transaction->new_get('http://mojolicious.org'); +my $tx3 = Mojo::Transaction->new_get('http://kraih.com'); +$client->process_all(Mojo::Pipeline->new($tx, $tx2), $tx3); +ok($tx->is_done); +ok($tx2->is_done); +ok($tx3->is_done); +is($tx->res->code, 301); +is($tx2->res->code, 200); +is($tx3->res->code, 200); +like($tx2->res->content->file->slurp, qr/Mojolicious/); + +# Pipelined with 100 Continue +$tx = Mojo::Transaction->new_get('http://labs.kraih.com'); +$tx2 = Mojo::Transaction->new_get('http://mojolicious.org'); +$tx2->req->headers->expect('100-continue'); +$tx2->req->body('foo bar baz'); +$tx3 = Mojo::Transaction->new_get('http://labs.kraih.com/blog/'); +my $tx4 = Mojo::Transaction->new_get('http://labs.kraih.com/blog'); +$client->process_all(Mojo::Pipeline->new($tx, $tx2, $tx3, $tx4)); +ok($tx->is_done); +ok($tx2->is_done); +ok($tx3->is_done); +ok($tx4->is_done); +is($tx->res->code, 301); +is($tx2->res->code, 200); +is($tx2->continued, 1); +is($tx3->res->code, 200); +is($tx4->res->code, 301); +like($tx2->res->content->file->slurp, qr/Mojolicious/); t/mojo/client.t @@ -8,12 +8,13 @@ use warnings; use Test::More; use Mojo::Client; +use Mojo::Pipeline; use Mojo::Transaction; use Test::Mojo::Server; plan skip_all => 'set TEST_DAEMON to enable this test (developer only!)' unless $ENV{TEST_DAEMON}; -plan tests => 16; +plan tests => 31; # Daddy, I'm scared. Too scared to even wet my pants. # Just relax and it'll come, son. @@ -61,5 +62,34 @@ is($tx->kept_alive, 1); like($tx->res->headers->connection, qr/Keep-Alive/i); like($tx->res->body, qr/Mojo is working/); +# Pipelined +$tx = Mojo::Transaction->new_get("http://127.0.0.1:$port/"); +my $tx2 = Mojo::Transaction->new_get("http://127.0.0.1:$port/"); +$client->process_all(Mojo::Pipeline->new($tx, $tx2)); +ok($tx->is_done); +ok($tx2->is_done); +is($tx->res->code, 200); +is($tx2->res->code, 200); +like($tx2->res->content->file->slurp, qr/Mojo is working/); + +# Pipelined with 100 Continue +$tx = Mojo::Transaction->new_get("http://127.0.0.1:$port/"); +$tx2 = Mojo::Transaction->new_get("http://127.0.0.1:$port/"); +$tx2->req->headers->expect('100-continue'); +$tx2->req->body('foo bar baz'); +my $tx3 = Mojo::Transaction->new_get("http://127.0.0.1:$port/"); +my $tx4 = Mojo::Transaction->new_get("http://127.0.0.1:$port/"); +$client->process_all(Mojo::Pipeline->new($tx, $tx2, $tx3, $tx4)); +ok($tx->is_done); +ok($tx2->is_done); +ok($tx3->is_done); +ok($tx4->is_done); +is($tx->res->code, 200); +is($tx2->res->code, 200); +is($tx2->continued, 1); +is($tx3->res->code, 200); +is($tx4->res->code, 200); +like($tx2->res->content->file->slurp, qr/Mojo is working/); + # Stop $server->stop_server_ok; t/mojo/daemon.t @@ -5,7 +5,7 @@ use strict; use warnings; -use Test::More tests => 234; +use Test::More tests => 240; use Mojo::Filter::Chunked; use Mojo::Headers; @@ -32,6 +32,16 @@ is($req->major_version, 1); is($req->minor_version, 1); is($req->url, '/'); +# Parse pipelined HTTP 1.1 start line, no headers and body +$req = Mojo::Message::Request->new; +$req->parse("GET / HTTP/1.1\x0d\x0a\x0d\x0aGET / HTTP/1.1\x0d\x0a\x0d\x0a"); +is($req->state, 'done_with_leftovers'); +is($req->method, 'GET'); +is($req->major_version, 1); +is($req->minor_version, 1); +is($req->url, '/'); +is($req->leftovers, "GET / HTTP/1.1\x0d\x0a\x0d\x0a"); + # Parse HTTP 1.0 start line and headers, no body $req = Mojo::Message::Request->new; $req->parse("GET /foo/bar/baz.html HTTP/1.0\x0d\x0a"); t/mojo/message.t @@ -93,7 +93,7 @@ is($url->to_abs, 'http://kraih.com/bar/baz/../foo?foo=bar#23'); is($url->to_abs->to_rel, '../foo?foo=bar#23'); is($url->to_abs->base, 'http://kraih.com/bar/baz/'); -# Real world test +# Real world tests $url = Mojo::URL->new('http://acme.s3.amazonaws.com' . '/mojo%2Fg%2B%2B-4%2E2_4%2E2%2E3-2ubuntu7_i386%2Edeb'); is($url->is_abs, 1); t/mojo/url.t @@ -34,7 +34,7 @@ $tx = Mojo::Transaction->new_get('/foo', 'X-Test' => 'Hi there!'); $client->process_local('MojoliciousTest', $tx); is($tx->res->code, 200); is($tx->res->headers->content_type, 'text/html'); -like($tx->res->body, qr/Hello Mojo from the template \/foo! Hello World!/); +like($tx->res->body, qr/<body>\n23Hello Mojo from the template \/foo! He/); # Foo::Bar::index $tx = Mojo::Transaction->new_get('/foo-bar', 'X-Test' => 'Hi there!'); t/mojolicious/app.t @@ -11,7 +11,7 @@ use base 'Mojolicious::Controller'; # Let me check my program... Yep. sub index { my $self = shift; - $self->stash(msg => 'Hello World!'); + $self->stash(layout => 'default', msg => 'Hello World!'); } sub templateless { t/mojolicious/lib/MojoliciousTest/Foo.pm 685a370c882b1e7f22fde88f00eb222c14cbb2c2 Sebastian Riedel kraih sri@labs.kraih.com http://github.com/kraih/mojo/commit/3114e47ddcbe132a1f0298572caf18231c8313fc 3114e47ddcbe132a1f0298572caf18231c8313fc 2009-04-18T17:15:24-07:00 2009-04-18T17:15:24-07:00 Added HTTP 1.1 pipelining support to Mojo and layouts to Mojolicious 1c823870140e6ef20b820ac8ecf6b5a81bcc4a1a Sebastian Riedel kraih sri@labs.kraih.com