POE::Component::Client::HTTP - a HTTP user-agent component

  use POE qw(Component::Client::HTTP);

  POE::Component::Client::HTTP->spawn(
    Agent     => 'SpiffCrawler/0.90',   # defaults to something long
    Alias     => 'ua',                  # defaults to 'weeble'
    From      => 'spiffster@perl.org',  # defaults to undef (no header)
    Protocol  => 'HTTP/0.9',            # defaults to 'HTTP/1.1'
    Timeout   => 60,                    # defaults to 180 seconds
    MaxSize   => 16384,                 # defaults to entire response
    Streaming => 4096,                  # defaults to 0 (off)
    FollowRedirects => 2,               # defaults to 0 (off)
    Proxy     => "http://localhost:80", # defaults to HTTP_PROXY env. variable
    NoProxy   => [ "localhost", "127.0.0.1" ], # defs to NO_PROXY env. variable
    BindAddr  => "12.34.56.78",         # defaults to INADDR_ANY
  );

  $kernel->post(
    'ua',        # posts to the 'ua' alias
    'request',   # posts to ua's 'request' state
    'response',  # which of our states will receive the response
    $request,    # an HTTP::Request object
  );

  # This is the sub which is called when the session receives a
  # 'response' event.
  sub response_handler {
    my ($request_packet, $response_packet) = @_[ARG0, ARG1];

    # HTTP::Request
    my $request_object  = $request_packet->[0];

    # HTTP::Response
    my $response_object = $response_packet->[0];

    my $stream_chunk;
    if (! defined($response_object->content)) {
      $stream_chunk = $response_packet->[1];
    }

    print(
      "*" x 78, "\n",
      "*** my request:\n",
      "-" x 78, "\n",
      $request_object->as_string(),
      "*" x 78, "\n",
      "*** their response:\n",
      "-" x 78, "\n",
      $response_object->as_string(),
    );

    if (defined $stream_chunk) {
      print "-" x 40, "\n", $stream_chunk, "\n";
    }

    print "*" x 78, "\n";
  }

  $kernel->post(
    'ua', 'request',           # http session alias & state
    'response',                # my state to receive responses
    GET 'http://poe.perl.org', # a simple HTTP request
    'unique id',               # a tag to identify the request
    'progress',                # an event to indicate progress
    'http://1.2.3.4:80/'       # proxy to use for this request
  );

  my $count = $kernel->call('ua' => 'pending_requests_count');

  $kernel->post( component => cancel => $http_request );

  $kernel->post( component => cancel => "$http_request" );

  my ($request_packet, $response_packet) = @_[ARG0, ARG1];

  my $http_request_object = $request_packet->[0];
  my $http_request_tag    = $request_packet->[1]; # from the 'request' post

  my $http_response_object = $response_packet->[0];

  sub progress_handler {
    my $gen_args  = $_[ARG0];    # args passed to all calls
    my $call_args = $_[ARG1];    # args specific to the call

    my $req = $gen_args->[0];    # HTTP::Request object being serviced
    my $tag = $gen_args->[1];    # Request ID tag from.
    my $got = $call_args->[0];   # Number of bytes retrieved so far.
    my $tot = $call_args->[1];   # Total bytes to be retrieved.
    my $oct = $call_args->[2];   # Chunk of raw octets received this time.

    my $percent = $got / $tot * 100;

    printf(
      "-- %.0f%% [%d/%d]: %s\n", $percent, $got, $tot, $req->uri()
    );

    # To cancel the request:
    # $_[KERNEL]->post( component => cancel => $req );
  }

  my $request = HTTP::Request->new( PUT => 'http://...' );

  my $file = '/path/to/large_file';

  open my $fh, '<', $file;

  my $upload_cb = sub {
    if ( sysread $fh, my $buf, 4096 ) {
      return $buf;
    }
    else {
      close $fh;
      return '';
    }
  };

  $request->content_length( -s $file );

  $request->content( $upload_cb );

  $kernel->post( ua => request, 'response', $request );

  my $request = HTTP::Request->new(
    GET => "http://www.yahoo.com/", [
      'Accept-Encoding' => 'gzip'
    ]
  );

  # ... time passes ...

  my $content = $response->decoded_content();

package POE::Component::Client::HTTP;
BEGIN {
  $POE::Component::Client::HTTP::VERSION = '0.943';
}
# vim: ts=2 sw=2 expandtab

# {{{ INIT

use strict;
#use bytes; # for utf8 compatibility

use constant DEBUG      => 0;
use constant DEBUG_DATA => 0;

use Carp qw(croak);
use HTTP::Response;
use Net::HTTP::Methods;
use Socket qw(sockaddr_in inet_ntoa);

use POE::Component::Client::HTTP::RequestFactory;
use POE::Component::Client::HTTP::Request qw(:states :fields);

BEGIN {
  local $SIG{'__DIE__'} = 'DEFAULT';
  #TODO: move this to Client::Keepalive?
  # Allow more finely grained timeouts if Time::HiRes is available.
  eval {
    require Time::HiRes;
    Time::HiRes->import("time");
  };
}

use Socket::GetAddrInfo qw(:newapi getnameinfo NI_NUMERICHOST NI_NUMERICSERV);

use POE qw(
  Driver::SysRW Filter::Stream
  Filter::HTTPHead Filter::HTTPChunk
  Component::Client::Keepalive
);

# The Internet Assigned Numbers Authority (IANA) acts as a registry
# for transfer-coding value tokens. Initially, the registry contains
# the following tokens: "chunked" (section 3.6.1), "identity" (section
# 3.6.2), "gzip" (section 3.5), "compress" (section 3.5), and
# "deflate" (section 3.5).

# FIXME - Haven't been able to test the compression options.
# Comments for each filter are what HTTP::Message use.  Methods
# without packages are from Compress::Zlib.

# FIXME - Is it okay to be mixing content and transfer encodings in
# this one table?

my %te_possible_filters = (
  'chunked'  => 'POE::Filter::HTTPChunk',
  'identity' => 'POE::Filter::Stream',
#  'gzip'     => 'POE::Filter::Zlib::Stream',  # Zlib: memGunzip
#  'x-gzip'   => 'POE::Filter::Zlib::Stream',  # Zlib: memGunzip
#  'x-bzip2'  => 'POE::Filter::Bzip2',         # Compress::BZip2::decompress
#  'deflate'  => 'POE::Filter::Zlib::Stream',  # Zlib: uncompress / inflate
#  'compress' => 'POE::Filter::LZW',           # unsupported
  # FIXME - base64 = MIME::Base64::decode
  # FIXME - quoted-printable = Mime::QuotedPrint::decode
);

my %te_filters;

while (my ($encoding, $filter) = each %te_possible_filters) {
  eval "use $filter";
  next if $@;
  $te_filters{$encoding} = $filter;
}

# The following defaults to 'chunked,identity' which is technically
# correct but arguably useless.  It also stomps on gzip'd transport
# because in the World Wild Web, Accept-Encoding is used to indicate
# gzip readiness, but the server responds with 'Content-Encoding:
# gzip', completely outside of TE encoding.
#
# Done this way so they appear in order of preference.
# FIXME - Is the order important here?

#my $accept_encoding = join(
#  ",",
#  grep { exists $te_filters{$_} }
#  qw(x-bzip2 gzip x-gzip deflate compress chunked identity)
#);

my %supported_schemes = (
  http  => 1,
  https => 1,
);

# }}} INIT

#------------------------------------------------------------------------------
# Spawn a new PoCo::Client::HTTP session.  This basically is a
# constructor, but it isn't named "new" because it doesn't create a
# usable object.  Instead, it spawns the object off as a separate
# session.
# {{{ spawn

sub spawn {
  my $type = shift;

  croak "$type requires an even number of parameters" if @_ % 2;

  my %params = @_;

  my $alias = delete $params{Alias};
  $alias = 'weeble' unless defined $alias and length $alias;

  my $bind_addr = delete $params{BindAddr};
  my $cm = delete $params{ConnectionManager};

  my $request_factory = POE::Component::Client::HTTP::RequestFactory->new(
    \%params
  );

  croak(
    "$type doesn't know these parameters: ",
    join(', ', sort keys %params)
  ) if scalar keys %params;

  POE::Session->create(
    inline_states => {
      _start  => \&_poco_weeble_start,
      _stop   => \&_poco_weeble_stop,
      _child  => sub { },

      # Public interface.
      request                => \&_poco_weeble_request,
      pending_requests_count => \&_poco_weeble_pending_requests_count,
      'shutdown'             => \&_poco_weeble_shutdown,
      cancel                 => \&_poco_weeble_cancel,

      # Client::Keepalive interface.
      got_connect_done  => \&_poco_weeble_connect_done,

      # ReadWrite interface.
      got_socket_input  => \&_poco_weeble_io_read,
      got_socket_flush  => \&_poco_weeble_io_flushed,
      got_socket_error  => \&_poco_weeble_io_error,

      # I/O timeout.
      got_timeout       => \&_poco_weeble_timeout,
      remove_request    => \&_poco_weeble_remove_request,
    },
    heap => {
      alias        => $alias,
      factory      => $request_factory,
      cm           => $cm,
      is_shut_down => 0,
      bind_addr    => $bind_addr,
    },
  );

  undef;
}

# }}} spawn
# ------------------------------------------------------------------------------
# {{{ _poco_weeble_start

sub _poco_weeble_start {
  my ($kernel, $heap) = @_[KERNEL, HEAP];

  $kernel->alias_set($heap->{alias});

  # have to do this here because it wants a current_session
  $heap->{cm} = POE::Component::Client::Keepalive->new(
    timeout => $heap->{factory}->timeout,
    ($heap->{bind_addr} ? (bind_address => $heap->{bind_addr}) : ()),
  ) unless ($heap->{cm});
}

# }}} _poco_weeble_start
#------------------------------------------------------------------------------
# {{{ _poco_weeble_stop

sub _poco_weeble_stop {
  my $heap = $_[HEAP];
  my $request = delete $heap->{request};

  foreach my $request_rec (values %$request) {
    $request_rec->remove_timeout();
    delete $heap->{ext_request_to_int_id}->{$request_rec->[REQ_HTTP_REQUEST]};
  }

  DEBUG and warn "Client::HTTP (alias=$heap->{alias}) stopped.";
}

# }}} _poco_weeble_stop
# {{{ _poco_weeble_pending_requests_count

sub _poco_weeble_pending_requests_count {
  my ($heap) = $_[HEAP];
  my $r = $heap->{request} || {};
  return scalar keys %$r;
}

# }}} _poco_weeble_pending_requests_count
#------------------------------------------------------------------------------
# {{{ _poco_weeble_request

sub _poco_weeble_request {
  my (
    $kernel, $heap, $sender,
    $response_event, $http_request, $tag, $progress_event,
    $proxy_override
  ) = @_[KERNEL, HEAP, SENDER, ARG0, ARG1, ARG2, ARG3, ARG4];

  unless (
    defined($http_request->uri->scheme) and
    length($http_request->uri->scheme) and
    $supported_schemes{$http_request->uri->scheme} and
    defined($http_request->uri->host) and
    length($http_request->uri->host)
  ) {
    my $rsp = HTTP::Response->new(
       400 => 'Bad Request', [],
       "<html>\n"
       . "<HEAD><TITLE>Error: Bad Request</TITLE></HEAD>\n"
       . "<BODY>\n"
       . "<H1>Error: Bad Request</H1>\n"
       . "Unsupported URI scheme\n"
       . "</BODY>\n"
       . "</HTML>\n"
    );
    $rsp->request($http_request);
    if (ref $response_event) {
      $response_event->postback->($rsp);
    } else {
      $kernel->post($sender, $response_event, [$http_request, $tag], [$rsp]);
    }
    return;
  }

  if ($heap->{is_shut_down}) {
    my $rsp = HTTP::Response->new(
       408 => 'Request timed out (component shut down)', [],
       "<html>\n"
       . "<HEAD><TITLE>Error: Request timed out (component shut down)"
       . "</TITLE></HEAD>\n"
       . "<BODY>\n"
       . "<H1>Error: Request Timeout</H1>\n"
       . "Request timed out (component shut down)\n"
       . "</BODY>\n"
       . "</HTML>\n"
      );
    $rsp->request($http_request);
    if (ref $response_event) {
      $response_event->postback->($rsp);
    } else {
      $kernel->post($sender, $response_event, [$http_request, $tag], [$rsp]);
    }
    return;
  }

  if (defined $proxy_override) {
    POE::Component::Client::HTTP::RequestFactory->parse_proxy($proxy_override);
  }

  my $request = $heap->{factory}->create_request(
    $http_request, $response_event, $tag, $progress_event,
    $proxy_override, $sender
  );
  $heap->{request}->{$request->ID} = $request;
  $heap->{ext_request_to_int_id}->{$http_request} = $request->ID;

  my @timeout;
  if ($heap->{factory}->timeout()) {
    @timeout = (
      timeout => $heap->{factory}->timeout()
    );
  }

  eval {
    # get a connection from Client::Keepalive
    #
    # TODO CONNECT - We must ask PCC::Keepalive to establish an http
    # socket, not https.  The initial proxy interactin is plaintext?

    $request->[REQ_CONN_ID] = $heap->{cm}->allocate(
      scheme  => $request->scheme,
      addr    => $request->host,
      port    => $request->port,
      context => $request->ID,
      event   => 'got_connect_done',
      @timeout,
    );
  };
  if ($@) {
    delete $heap->{request}->{$request->ID};
    delete $heap->{ext_request_to_int_id}->{$http_request};

    # we can reach here for things like host being invalid.
    $request->error(400, $@);
  }
}

# }}} _poco_weeble_request

#------------------------------------------------------------------------------
# {{{ _poco_weeble_connect_done

sub _poco_weeble_connect_done {
  my ($heap, $response) = @_[HEAP, ARG0];

  my $connection = $response->{'connection'};
  my $request_id = $response->{'context'};

  # Can't handle connections if we're shut down.
  # TODO - How do we still get these?  Were they previously queued or
  # something?
  if ($heap->{is_shut_down}) {
    _internal_cancel(
      $heap, $request_id, 408, "Request timed out (request canceled)"
    );
    return;
  }

  if (defined $connection) {
    DEBUG and warn "CON: request $request_id connected ok...";

    my $request = $heap->{request}->{$request_id};
    unless (defined $request) {
      DEBUG and warn "CON: ignoring connection for canceled request";
      return;
    }

    my $block_size = $heap->{factory}->block_size;

    # get wheel from the connection
    my $new_wheel = $connection->start(
      Driver       => POE::Driver::SysRW->new(BlockSize => $block_size),
      InputFilter  => POE::Filter::HTTPHead->new(),
      OutputFilter => POE::Filter::Stream->new(),
      InputEvent   => 'got_socket_input',
      FlushedEvent => 'got_socket_flush',
      ErrorEvent   => 'got_socket_error',
    );

    DEBUG and warn "CON: request $request_id uses wheel ", $new_wheel->ID;

    # Add the new wheel ID to the lookup table.
    $heap->{wheel_to_request}->{ $new_wheel->ID() } = $request_id;

    $request->[REQ_CONNECTION] = $connection;

    # SSLify needs us to call it's function to get the "real" socket
    my $peer_addr;
    if ( $request->scheme eq 'http' ) {
      $peer_addr = getpeername($new_wheel->get_input_handle());
    } else {
      my $socket = $new_wheel->get_input_handle();
      $peer_addr = getpeername(POE::Component::SSLify::SSLify_GetSocket($socket));
    }

    if (defined $peer_addr) {
      my ($error, $address, $port) = getnameinfo(
        $peer_addr, NI_NUMERICHOST | NI_NUMERICSERV
      );

      if ($error) {
        $request->[REQ_PEERNAME] = "error: $error";
      }
      else {
        $request->[REQ_PEERNAME] = "$address.$port";
      }
    }
    else {
      $request->[REQ_PEERNAME] = "error: $!";
    }

    $request->create_timer($heap->{factory}->timeout);
    $request->send_to_wheel;
  }
  else {
    DEBUG and warn(
      "CON: Error connecting for request $request_id --- ", $_[SENDER]->ID
    );

    my ($operation, $errnum, $errstr) = (
      $response->{function},
      $response->{error_num} || '??',
      $response->{error_str}
    );

    DEBUG and warn(
      "CON: request $request_id encountered $operation error " .
      "$errnum: $errstr"
    );

    DEBUG and warn "I/O: removing request $request_id";
    my $request = delete $heap->{request}->{$request_id};
    $request->remove_timeout();
    delete $heap->{ext_request_to_int_id}->{$request->[REQ_HTTP_REQUEST]};

    # Post an error response back to the requesting session.
    $request->connect_error($operation, $errnum, $errstr);
  }
}

# }}} _poco_weeble_connect_done

# {{{ _poco_weeble_timeout

sub _poco_weeble_timeout {
  my ($kernel, $heap, $request_id) = @_[KERNEL, HEAP, ARG0];

  DEBUG and warn "T/O: request $request_id timed out";

  # Discard the request.  Keep a copy for a few bits of cleanup.
  DEBUG and warn "I/O: removing request $request_id";
  my $request = delete $heap->{request}->{$request_id};

  unless (defined $request) {
    die(
      "T/O: unexpectedly undefined request for id $request_id\n",
      "T/O: known request IDs: ", join(", ", keys %{$heap->{request}}), "\n",
      "...",
    );
  }

  DEBUG and warn "T/O: request $request_id has timer ", $request->timer;
  $request->remove_timeout();
  delete $heap->{ext_request_to_int_id}->{$request->[REQ_HTTP_REQUEST]};

  # There's a wheel attached to the request.  Shut it down.
  if (defined(my $wheel = $request->wheel())) {
    my $wheel_id = $wheel->ID();
    DEBUG and warn "T/O: request $request_id is wheel $wheel_id";

    # Shut down the connection so it's not reused.
    $wheel->shutdown_input();
    delete $heap->{wheel_to_request}->{$wheel_id};
  }


  DEBUG and do {
    die( "T/O: request $request_id is unexpectedly zero" )
      unless $request->[REQ_STATE];
    warn "T/O: request_state = " . sprintf("%#04x\n", $request->[REQ_STATE]);
  };

  # Hey, we haven't sent back a response yet!
  unless ($request->[REQ_STATE] & (RS_REDIRECTED | RS_POSTED)) {

    # Well, we have a response.  Isn't that nice?  Let's send it.
    if ($request->[REQ_STATE] & (RS_IN_CONTENT | RS_DONE)) {
      _finish_request($heap, $request);
      return;
    }

    # Post an error response back to the requesting session.
    DEBUG and warn "I/O: Disconnect, keepalive timeout or HTTP/1.0.";
    $request->error(408, "Request timed out") if $request->[REQ_STATE];
    return;
  }
}

# }}} _poco_weeble_timeout
#------------------------------------------------------------------------------
# {{{ _poco_weeble_io_flushed

sub _poco_weeble_io_flushed {
  my ($heap, $wheel_id) = @_[HEAP, ARG0];

  # We sent the request.  Now we're looking for a response.  It may be
  # bad to assume we won't get a response until a request has flushed.
  my $request_id = $heap->{wheel_to_request}->{$wheel_id};
  if (not defined $request_id) {
    DEBUG and warn "!!!: unexpectedly undefined request ID";
    return;
  }

  DEBUG and warn(
    "I/O: wheel $wheel_id (request $request_id) flushed its request..."
  );

  my $request = $heap->{request}->{$request_id};

  # Read content to send from a callback
  if ( ref $request->[REQ_HTTP_REQUEST]->content() eq 'CODE' ) {
    my $callback = $request->[REQ_HTTP_REQUEST]->content();

    my $buf = eval { $callback->() };

    if ( $buf ) {
      $request->[REQ_CONNECTION]->wheel->put($buf);

      # reset the timeout
      # Have to also reset REQ_START_TIME or timer ends early
      $request->remove_timeout;
      $request->[REQ_START_TIME] = time();
      $request->create_timer($heap->{factory}->timeout);

      return;
    }
  }

  $request->[REQ_STATE] ^= RS_SENDING;
  $request->[REQ_STATE] = RS_IN_HEAD;

  # XXX - Removed a second time.  The first time was in version 0.53,
  # because the EOF generated by shutdown_output() causes some servers
  # to disconnect rather than send their responses.
  # $request->wheel->shutdown_output();
}

# }}} _poco_weeble_io_flushed
#------------------------------------------------------------------------------
# {{{ _poco_weeble_io_error

sub _poco_weeble_io_error {
  my ($kernel, $heap, $operation, $errnum, $errstr, $wheel_id) =
    @_[KERNEL, HEAP, ARG0..ARG3];

  DEBUG and warn(
    "I/O: wheel $wheel_id encountered $operation error $errnum: $errstr"
  );

  # Drop the wheel.
  my $request_id = delete $heap->{wheel_to_request}->{$wheel_id};

  # There was no corresponding request?  Nothing left to do here.
  # We might have got here because the server sent EOF after we were done processing
  # the request, and deleted it from our cache. ( notes for RT#50231 )
  return unless $request_id;

  DEBUG and warn "I/O: removing request $request_id";
  my $request = delete $heap->{request}->{$request_id};
  $request->remove_timeout;
  delete $heap->{ext_request_to_int_id}{$request->[REQ_HTTP_REQUEST]};

  # Otherwise the remote end simply closed.  If we've got a pending
  # response, then post it back to the client.
  DEBUG and warn "STATE is ", $request->[REQ_STATE];

  # Except when we're redirected.  In this case, the connection was but
  # one step towards our destination.
  return if ($request->[REQ_STATE] & RS_REDIRECTED);

  # If there was a non-zero error, then something bad happened.  Post
  # an error response back, if we haven't posted anything before.
  if ($errnum) {
    if ($operation eq "connect") {
      $request->connect_error($operation, $errnum, $errstr);
      return;
    }

    unless ($request->[REQ_STATE] & RS_POSTED) {
      $request->error(400, "$operation error $errnum: $errstr");
    }
    return;
  }

  # We seem to have finished with the request.  Send back a response.
  if (
    $request->[REQ_STATE] & (RS_IN_CONTENT | RS_DONE) and
    not $request->[REQ_STATE] & RS_POSTED
  ) {
    _finish_request($heap, $request);
    return;
  }

  # We have already posted a response, so this is a remote keepalive
  # timeout or other delayed socket shutdown.  Nothing left to do.
  if ($request->[REQ_STATE] & RS_POSTED) {
    DEBUG and warn "I/O: Disconnect, remote keepalive timeout or HTTP/1.0.";
    return;
  }

  # We never received a response.
  if (not defined $request->[REQ_RESPONSE]) {
    # Check for pending data indicating a LF-free HTTP 0.9 response.
    my $lines = $request->wheel->get_input_filter()->get_pending();
    my $text = join '' => @$lines;
    DEBUG and warn "Got ", length($text), " bytes of data without LF.";

    # If we have data, build and return a response from it.
    if ($text =~ /\S/) {
      DEBUG and warn(
        "Generating HTTP response for HTTP/0.9 response without LF."
      );
      $request->[REQ_RESPONSE] = HTTP::Response->new(
        200, 'OK', [
          'Content-Type'  => 'text/html',
          'X-PCCH-Peer'   => $request->[REQ_PEERNAME],
        ], $text
      );
      $request->[REQ_RESPONSE]->protocol('HTTP/0.9');
      $request->[REQ_RESPONSE]->request($request->[REQ_HTTP_REQUEST]);
      $request->[REQ_STATE] = RS_DONE;
      $request->return_response;
      return;
    }

    # No data received.  This is an incomplete response.
    $request->error(400, "Incomplete response - $request_id");
    return;
  }

  # We haven't built a proper response, and nothing returned by the
  # server can be turned into a proper response.  Send back an error.
  # Changed to 406 after considering rt.cpan.org 20975.
  #
  # 10.4.7 406 Not Acceptable
  #
  # The resource identified by the request is only capable of
  # generating response entities which have content characteristics
  # not acceptable according to the accept headers sent in the
  # request.

  $request->error(406, "Server response is Not Acceptable - $request_id");
}

# }}} _poco_weeble_io_error
#------------------------------------------------------------------------------
# Read a chunk of response.  This code is directly adapted from Artur
# Bergman's nifty POE::Filter::HTTPD, which does pretty much the same
# in the other direction.
# {{{ _poco_weeble_io_read

sub _poco_weeble_io_read {
  my ($kernel, $heap, $input, $wheel_id) = @_[KERNEL, HEAP, ARG0, ARG1];
  my $request_id = $heap->{wheel_to_request}->{$wheel_id};

  DEBUG and warn "I/O: wheel $wheel_id got input...";
  DEBUG_DATA and warn (ref($input) ? $input->as_string : _hexdump($input));

  # There was no corresponding request?  Nothing left to do here.
  # We might have got here because the server sent EOF after we were done processing
  # the request, and deleted it from our cache. ( notes for RT#50231 )
  return unless defined $request_id;

  my $request = $heap->{request}->{$request_id};
  return unless defined $request;
  DEBUG and warn(
    "REQUEST $request_id is $request <" . $request->[REQ_HTTP_REQUEST]->uri . ">"
  );

  # Reset the timeout if we get data.
  $kernel->delay_adjust($request->timer, $heap->{factory}->timeout);

  if ($request->[REQ_STATE] & RS_REDIRECTED) {
    DEBUG and warn "input for request that was redirected";
    return;
  }

# {{{ HEAD

  # The very first line ought to be status.  If it's not, then it's
  # part of the content.
  if ($request->[REQ_STATE] & RS_IN_HEAD) {
    if (defined $input) {
      $input->request ($request->[REQ_HTTP_REQUEST]);
      #warn(
      #  "INPUT for ", $request->[REQ_HTTP_REQUEST]->uri,
      #  " is \n",$input->as_string
      #)
    }
    else {
      #warn "NO INPUT";
    }

    # FIXME: LordVorp gets here without $input being a HTTP::Response.
    # FIXME: This happens when the response is HTTP/0.9 and doesn't
    # include a status line.  See t/53_response_parser.t.
    $request->[REQ_RESPONSE] = $input;
    $input->header("X-PCCH-Peer", $request->[REQ_PEERNAME]);

    # TODO CONNECT - If we've got the headers to a CONNECT request,
    # then we can switch to the actual request.  This is like a faux
    # redirect where the socket gets reused.
    #
    # 1. Switch the socket to SSL.
    # 2. Switch the request from CONNECT mode to regular mode, using
    #    the method proposed in PCCH::Request.
    # 3. Send the original request via PCCH::Request->send_to_wheel().
    #    This puts the client back into the RS_SENDING state.
    # 4. Reset any data/state so it appears we never went through
    #    CONNECT.
    # 5. Make sure that PCC::Keepalive will discard the socket when
    #    we're done with it.
    # 6. Return.  The connection should proceed as normal.
    #
    # I think the normal handling for HTTP errors will cover the case
    # of CONNECT failure.  If not, we can refine the implementation as
    # needed.

    # Some responses are without content by definition
    # FIXME: #12363
    # Make sure we finish even when it isn't one of these, but there
    # is no content.
    if (
      $request->[REQ_HTTP_REQUEST]->method eq 'HEAD'
      or $input->code =~ /^(?:1|[23]04)/
      or (
        defined($input->content_length())
        and $input->content_length() == 0
      )
    ) {
      if (_try_redirect($request_id, $input, $request)) {
        my $old_request = delete $heap->{request}->{$request_id};
        delete $heap->{wheel_to_request}->{$wheel_id};
        if (defined $old_request) {
          DEBUG and warn "I/O: removed request $request_id";
          $old_request->remove_timeout();
          delete $heap->{ext_request_to_int_id}{$old_request->[REQ_HTTP_REQUEST]};
          $old_request->[REQ_CONNECTION] = undef;
        }
        return;
      }
      $request->[REQ_STATE] |= RS_DONE;
      $request->remove_timeout();
      _finish_request($heap, $request);
      return;
    }
    else {
      # If we have content length, and it's more than the maximum we
      # requested, then fail without bothering with the content.
      if (
        defined($heap->{factory}->max_response_size())
        and defined($input->content_length())
        and $input->content_length() > $heap->{factory}->max_response_size()
      ) {
        _internal_cancel(
          $heap, $request_id, 406,
          "Response content length " . $input->content_length() .
          " is greater than specified MaxSize of " .
          $heap->{factory}->max_response_size() .
          ".  Use range requests to retrieve specific amounts of content."
        );
        return;
      }

      $request->[REQ_STATE] |= RS_IN_CONTENT;
      $request->[REQ_STATE] &= ~RS_IN_HEAD;
      #FIXME: probably want to find out when the content from this
      #       request is in, and only then do the new request, so we
      #       can reuse the connection.
      if (_try_redirect($request_id, $input, $request)) {
        my $old_request = delete $heap->{request}->{$request_id};
        delete $heap->{wheel_to_request}->{$wheel_id};
        if (defined $old_request) {
          DEBUG and warn "I/O: removed request $request_id";
          delete $heap->{ext_request_to_int_id}{$old_request->[REQ_HTTP_REQUEST]};
          $old_request->remove_timeout();
          $old_request->[REQ_CONNECTION]->close();
          $old_request->[REQ_CONNECTION] = undef;
        }
        return;
      }

      # RFC 2616 14.41:  If multiple encodings have been applied to an
      # entity, the transfer-codings MUST be listed in the order in
      # which they were applied.

      my ($filter, @filters);

      # Transfer encoding.

      my $te = $input->header('Transfer-Encoding');
      if (defined $te) {
        my @te = split(/\s*,\s*/, lc($te));

        while (@te and exists $te_filters{$te[-1]}) {
          my $encoding = pop @te;
          my $fclass = $te_filters{$encoding};
          push @filters, $fclass->new();
        }

        if (@te) {
          $input->header('Transfer-Encoding', join(', ', @te));
        }
        else {
          $input->header('Transfer-Encoding', undef);
        }
      }

      # Content encoding.

      my $ce = $input->header('Content-Encoding');
      if (defined $ce) {
        my @ce = split(/\s*,\s*/, lc($ce));

        while (@ce and exists $te_filters{$ce[-1]}) {
          my $encoding = pop @ce;
          my $fclass = $te_filters{$encoding};
          push @filters, $fclass->new();
        }

        if (@ce) {
          $input->header('Content-Encoding', join(', ', @ce));
        }
        else {
          $input->header('Content-Encoding', undef);
        }
      }

      if (@filters > 1) {
        $filter = POE::Filter::Stackable->new( Filters => \@filters );
      }
      elsif (@filters) {
        $filter = $filters[0];
      }
      else {
        # Punt if we have no specified filters.
        $filter = POE::Filter::Stream->new;
      }

      # do this last, because it triggers a read
      $request->wheel->set_input_filter($filter);
    }
    return;
  }

# }}} HEAD

# {{{ content

  # We're in a content state.
  if ($request->[REQ_STATE] & RS_IN_CONTENT) {
    if (ref($input) and UNIVERSAL::isa($input, 'HTTP::Response')) {
      # there was a problem in the input filter
      # $request->close_connection;
    }
    else {
      my $is_done = $request->add_content ($input);
    }
  }

# }}} content

# {{{ deliver reponse if complete

# POST response without disconnecting
  if (
    $request->[REQ_STATE] & RS_DONE and
    not $request->[REQ_STATE] & RS_POSTED
  ) {
    $request->remove_timeout;
    _finish_request($heap, $request);
  }

# }}} deliver reponse if complete

}

# }}} _poco_weeble_io_read


#------------------------------------------------------------------------------
# Generate a hex dump of some input. This is not a POE function.
# {{{ _hexdump

sub _hexdump {
  my $data = shift;

  my $dump;
  my $offset = 0;
  while (length $data) {
    my $line = substr($data, 0, 16);
    substr($data, 0, 16) = '';

    my $hexdump  = unpack 'H*', $line;
    $hexdump =~ s/(..)/$1 /g;

    $line =~ tr[ -~][.]c;
    $dump .= sprintf( "%04x %-47.47s - %s\n", $offset, $hexdump, $line );
    $offset += 16;
  }

  return $dump;
}

# }}} _hexdump

# Check for and handle redirect.  Returns true if redirect should
# occur, or false if there's no redirect.

sub _try_redirect {
  my ($request_id, $input, $request) = @_;

  if (my $newrequest = $request->check_redirect) {
    DEBUG and warn(
      "Redirected $request_id ", $input->code, " to <",
      $newrequest->uri, ">"
    );
    my @proxy;
    if ($request->[REQ_USING_PROXY]) {
      push @proxy, (
        'http://' .  $request->host .  ':' .  $request->port .  '/'
      );
    }

    $poe_kernel->yield(
      request =>
      $request,
      $newrequest,
      "_redir_".$request->ID,
      $request->[REQ_PROG_POSTBACK],
      @proxy
    );

    return 1;
  }

  return;
}

# Complete a request. This was moved out of _poco_weeble_io_error(). This is
# not a POE function.
# {{{ _finish_request

sub _finish_request {
  my ($heap, $request) = @_;

  my $request_id = $request->ID;
  if (DEBUG) {
    my ($pkg, $file, $line) = caller();
    warn(
      "XXX: calling _finish_request(request id = $request_id)" .
      "at $file line $line"
    );
  }

  # XXX What does this do?
  $request->add_eof;

  # KeepAlive: added the RS_POSTED flag
  $request->[REQ_STATE] |= RS_POSTED;

  my $wheel_id = defined $request->wheel ? $request->wheel->ID : "(undef)";
  DEBUG and warn "Wheel from request is ", $wheel_id;
  # clean up the request
  my $address = "$request->[REQ_HOST]:$request->[REQ_PORT]";

  DEBUG and warn "address is $address";

  return _clear_req_cache( $heap, $request_id );
}

# }}} _finish_request

#{{{ _remove_request
sub _poco_weeble_remove_request {
  my ($kernel, $heap, $request_id) = @_[KERNEL, HEAP, ARG0];

  return _clear_req_cache( $heap, $request_id );
}
#}}} _remove_request

# helper subroutine to remove a request from our caches
#{{{ _clear_req_cache
sub _clear_req_cache {
  my ($heap, $request_id) = @_;

  my $request = delete $heap->{request}->{$request_id};
  if (defined $request) {
    DEBUG and warn "I/O: removed request $request_id";
    $request->remove_timeout();
    delete $heap->{ext_request_to_int_id}{$request->[REQ_HTTP_REQUEST]};
    if (my $wheel = $request->wheel) {
      delete $heap->{wheel_to_request}->{$wheel->ID};
    }
  }

  return;
}
#}}} _clear_req_cache

# Cancel a single request by HTTP::Request object.

sub _poco_weeble_cancel {
  my ($kernel, $heap, $request) = @_[KERNEL, HEAP, ARG0];
  my $request_id = $heap->{ext_request_to_int_id}{$request};
  return unless defined $request_id;
  _internal_cancel(
    $heap, $request_id, 408, "Request timed out (request canceled)"
  );
}

sub _internal_cancel {
  my ($heap, $request_id, $code, $message) = @_;

  my $request = delete $heap->{request}{$request_id};
  return unless defined $request;

  DEBUG and warn "CXL: canceling request $request_id";
  $request->remove_timeout();
  delete $heap->{ext_request_to_int_id}{$request->[REQ_HTTP_REQUEST]};

  if (my $wheel = $request->wheel) {
    my $wheel_id = $wheel->ID;
    DEBUG and warn "CXL: Request $request_id canceling wheel $wheel_id";
    delete $heap->{wheel_to_request}{$wheel_id};
    $wheel = undef;
  }

  if ($request->[REQ_CONNECTION]) {
    $request->[REQ_CONNECTION]->close();
    $request->[REQ_CONNECTION] = undef;
  }
  else {
    # Didn't connect yet; inform connection manager to cancel
    # connection request.
    $heap->{cm}->deallocate($request_id);
  }

  unless ($request->[REQ_STATE] & RS_POSTED) {
    $request->error($code, $message);
  }
}

# Shut down the entire component.
sub _poco_weeble_shutdown {
  my ($kernel, $heap) = @_[KERNEL, HEAP];

  $heap->{is_shut_down} = 1;

  my @request_ids = keys %{$heap->{request}};
  foreach my $request_id (@request_ids) {
    _internal_cancel(
      $heap, $request_id, 408, "Request timed out (component shut down)"
    );
  }

  # Shut down the connection manager subcomponent.
  if (defined $heap->{cm}) {
    DEBUG and warn "CXL: Client::HTTP shutting down Client::Keepalive";
    $heap->{cm}->shutdown();
    delete $heap->{cm};
  }

  # Final cleanup of this component.
  $kernel->alias_remove($heap->{alias});
}

1;

__END__

# {{{ POD

# }}} POD