Merge pull request #15 Add content handler for .pagespeed resources

Pagespeed generates urls like:
    /script.js.pagespeed.ce.FjFFoSiQs4.js
when it moves a resource. Add a content handler that accepts
these requests and passes them through to pagespeed.

src/ngx_base_fetch.cc

@@ -20,11 +20,13 @@
 #include "ngx_pagespeed.h"
 #include "net/instaweb/util/public/google_message_handler.h"
 #include "net/instaweb/util/public/message_handler.h"
+#include "net/instaweb/http/public/response_headers.h"
 
 namespace net_instaweb {
 
 NgxBaseFetch::NgxBaseFetch(ngx_http_request_t* r, int pipe_fd)
-    : request_(r), done_called_(false), pipe_fd_(pipe_fd) {
+    : request_(r), done_called_(false), last_buf_sent_(false),
+      pipe_fd_(pipe_fd) {
 }
 
 NgxBaseFetch::~NgxBaseFetch() { }
@@ -68,68 +70,146 @@ void NgxBaseFetch::PopulateHeaders() {
 
 bool NgxBaseFetch::HandleWrite(const StringPiece& sp,
                                MessageHandler* handler) {
-  // TODO(jefftk): acquire lock on buffer_ here.
+  Lock();
   buffer_.append(sp.data(), sp.size());
-  // TODO(jefftk): release lock here.
+  Unlock();
   return true;
 }
 
-// TODO(jefftk): this is vulnerable to race conditions.  Memory inconsistencies
-// between threads can mean that chain links get dropped, which is of course
-// very bad.  To fix this we should protect buffer_ with a lock that gets
-// acquired both here and in HandleWrite so that we make sure they both have a
-// consistent view of memory.
-//
-// There may also be a race condition if this is called between the last Write()
-// and Done() such that we're sending an empty buffer with last_buf set, which I
-// think nginx will reject.
-ngx_int_t NgxBaseFetch::CollectAccumulatedWrites(ngx_chain_t** link_ptr) {
-  // TODO(jefftk): acquire lock on buffer_ here.
-
-  if (!done_called_ && buffer_.length() == 0) {
+ngx_int_t NgxBaseFetch::CopyBufferToNginx(ngx_chain_t** link_ptr) {
+  if ((last_buf_sent_ || !done_called_) && buffer_.length() == 0) {
     // Nothing to send.  But if done_called_ then we can't short circuit because
-    // we need to set last_buf.
-    *link_ptr = NULL;
-    return NGX_OK;
+    // we need to set last_buf unless last_buf_sent_.
+    return NGX_DECLINED;
   }
 
+  // Below, *link_ptr will be NULL if we're starting the chain, and the head
+  // chain link.
+  *link_ptr = NULL;
+
+  // If non-null, the current last link in the chain.
+  ngx_chain_t* tail_link = NULL;
+
+  // How far into buffer_ we're currently working on.
+  ngx_uint_t offset;
+
+  // TODO(jefftk): look up the nginx buffer size properly.
+  ngx_uint_t max_buffer_size = 8192;  // 8k
+  for (offset = 0 ;
+       offset < buffer_.length() ||
+           // If the pagespeed buffer is empty but Done() has been called we
+           // need to pass through an empty buffer to nginx to communicate
+           // last_buf.  Otherwise we shouldn't generate empty buffers.
+           (offset == 0 && buffer_.length() == 0);
+       offset += max_buffer_size) {
+
     // Prepare a new nginx buffer to put our buffered writes into.
     ngx_buf_t* b = static_cast<ngx_buf_t*>(ngx_calloc_buf(request_->pool));
     if (b == NULL) {
       return NGX_ERROR;
     }
+
+    if (buffer_.length() == 0) {
+      CHECK(offset == 0);
+      b->pos = b->start = b->end = b->last = NULL;
+      // The purpose of this buffer is just to pass along last_buf.
+      b->sync = 1;
+    } else {
+      CHECK(buffer_.length() > offset);
+      ngx_uint_t b_size = buffer_.length() - offset;
+      if (b_size > max_buffer_size) {
+        b_size = max_buffer_size;
+      }
+
       b->start = b->pos = static_cast<u_char*>(
-      ngx_palloc(request_->pool, buffer_.length()));
+          ngx_palloc(request_->pool, b_size));
       if (b->pos == NULL) {
         return NGX_ERROR;
       }
 
-  // Copy our writes over.
-  buffer_.copy(reinterpret_cast<char*>(b->pos), buffer_.length());
-  b->last = b->end = b->pos + buffer_.length();
+      // Copy our writes over.  We're copying from buffer_[offset] up to
+      // buffer_[offset + b_size] into b which has size b_size.
+      buffer_.copy(reinterpret_cast<char*>(b->pos), b_size, offset);
+      b->last = b->end = b->pos + b_size;
 
-  if (buffer_.length()) {
-    b->temporary = 1;
+      b->temporary = 1;  // Identify this buffer as in-memory and mutable.
+    }
+
+    // Prepare a chain link.
+    ngx_chain_t* cl = static_cast<ngx_chain_t*>(
+        ngx_alloc_chain_link(request_->pool));
+    if (cl == NULL) {
+      return NGX_ERROR;
+    }
+
+    cl->buf = b;
+    cl->next = NULL;
+
+    if (*link_ptr == NULL) {
+      // This is the first link in the returned chain.
+      *link_ptr = cl;
     } else {
-    b->sync = 1;
+      // Link us into the chain.
+      CHECK(tail_link != NULL);
+      tail_link->next = cl;
+    }
+
+    tail_link = cl;
   }
 
   // Done with buffer contents now.
   buffer_.clear();
 
-  // TODO(jefftk): release lock here.
-
-  b->last_buf = b->last_in_chain = done_called_;
-  
-  // Prepare a chain link for our new buffer.
-  *link_ptr = static_cast<ngx_chain_t*>(
-      ngx_alloc_chain_link(request_->pool));
-  if (*link_ptr == NULL) {
-    return NGX_ERROR;
+  CHECK(tail_link != NULL);
+  if (done_called_) {
+    tail_link->buf->last_buf = true;
+    last_buf_sent_ = true;
+  }  
+
+  return NGX_OK;
+}
+
+// There may also be a race condition if this is called between the last Write()
+// and Done() such that we're sending an empty buffer with last_buf set, which I
+// think nginx will reject.
+ngx_int_t NgxBaseFetch::CollectAccumulatedWrites(ngx_chain_t** link_ptr) {
+  Lock();
+  ngx_int_t rc = CopyBufferToNginx(link_ptr);
+  Unlock();
+
+  if (rc == NGX_DECLINED) {
+    *link_ptr = NULL;
+    return NGX_OK;
+  }
+  return rc;
+}
+
+ngx_int_t NgxBaseFetch::CollectHeaders(ngx_http_headers_out_t* headers_out) {
+  // Copy from response_headers() into headers_out.
+
+  Lock();
+  const ResponseHeaders* pagespeed_headers = response_headers();
+  Unlock();
+
+  headers_out->status = NGX_HTTP_OK;
+
+  ngx_int_t i;
+  for (i = 0 ; i < pagespeed_headers->NumAttributes() ; i++) {
+    const GoogleString& name = pagespeed_headers->Name(i);
+    const GoogleString& value = pagespeed_headers->Value(i);
+
+    // TODO(jefftk): If we're setting a cache control header we'd like to
+    // prevent any downstream code from changing it.  Specifically, if we're
+    // serving a cache-extended resource the url will change if the resource
+    // does and so we've given it a long lifetime.  If the site owner has done
+    // something like set all css files to a 10-minute cache lifetime, that
+    // shouldn't apply to our generated resources.  See Apache code in
+    // net/instaweb/apache/header_util:AddResponseHeadersToRequest
+
+    // TODO(jefftk): actually copy headers over
+    fprintf(stderr, "Would set header '%s: %s'\n", name.c_str(), value.c_str());
   }
 
-  (*link_ptr)->buf = b;
-  (*link_ptr)->next = NULL;
   return NGX_OK;
 }
 
@@ -151,8 +231,12 @@ void NgxBaseFetch::RequestCollection() {
   }
 }
 
+void NgxBaseFetch::HandleHeadersComplete() {
+  RequestCollection();  // Headers available.
+}
+
 bool NgxBaseFetch::HandleFlush(MessageHandler* handler) {
-  RequestCollection();  // data available.
+  RequestCollection();  // A new part of the response body is available.
   return true;
 }
 

src/ngx_base_fetch.h

@@ -49,30 +49,54 @@ class NgxBaseFetch : public AsyncFetch {
   // Copies the response headers out of request_->headers_out->headers.
   void PopulateHeaders();
 
-  // Puts a chain link in link_ptr if we have any output data buffered.  Returns
+  // Puts a chain in link_ptr if we have any output data buffered.  Returns
   // NGX_OK on success, NGX_ERROR on errors.  If there's no data to send, sends
   // data only if Done() has been called.  Indicates the end of output by
-  // setting last_buf on the chain link
+  // setting last_buf on the last buffer in the chain.
   //
-  // Always sets link_ptr to a single chain link, never a full chain.
+  // Sets link_ptr to a chain of as many buffers are needed for the output.
   //
-  // Called by nginx.
+  // Called by nginx in response to seeing a byte on the pipe.
   ngx_int_t CollectAccumulatedWrites(ngx_chain_t** link_ptr);
 
+  // Copies response headers into headers_out.
+  //
+  // Called by nginx before calling CollectAccumulatedWrites() for the first
+  // time for resource fetches.  Not called at all for proxy fetches.
+  ngx_int_t CollectHeaders(ngx_http_headers_out_t* headers_out);
+
  private:
-  ResponseHeaders response_headers_;
   virtual bool HandleWrite(const StringPiece& sp, MessageHandler* handler);
   virtual bool HandleFlush(MessageHandler* handler);
-  virtual void HandleHeadersComplete() {}
+  virtual void HandleHeadersComplete();
   virtual void HandleDone(bool success);
 
   // Indicate to nginx that we would like it to call
   // CollectAccumulatedWrites().
   void RequestCollection();
 
+  // Lock must be acquired first.
+  // Returns:
+  //   NGX_DECLINED: nothing to send, short circuit.  Buffer not allocated.
+  //   NGX_OK, NGX_ERROR: success, failure
+  // Allocates an nginx buffer, copies our buffer_ contents into it, clears
+  // buffer_.
+  ngx_int_t CopyBufferToNginx(ngx_chain_t** link_ptr);
+
+  // Acquire this lock before manipulating buffer_.
+  // TODO(jefftk): Actually implement this.  Possibly with ngx_shmtx_lock and
+  // ngx_shmtx_unlock.
+  void Lock() {
+    fprintf(stderr, "Would lock buffer_\n");
+  }
+  void Unlock() {
+    fprintf(stderr, "Would unlock buffer_\n");
+  }
+
   ngx_http_request_t* request_;
   GoogleString buffer_;
   bool done_called_;
+  bool last_buf_sent_;
   int pipe_fd_;
 
   DISALLOW_COPY_AND_ASSIGN(NgxBaseFetch);

src/ngx_pagespeed.cc

@@ -45,6 +45,7 @@ extern "C" {
 #include "net/instaweb/util/public/google_url.h"
 #include "net/instaweb/util/public/string.h"
 #include "net/instaweb/util/public/google_message_handler.h"
+#include "net/instaweb/automatic/public/resource_fetch.h"
 
 extern ngx_module_t ngx_pagespeed;
 
@@ -67,7 +68,6 @@ typedef struct {
 
 typedef struct {
   ngx_http_pagespeed_srv_conf_t* cfg;
-  net_instaweb::RewriteDriver* driver;
   net_instaweb::ProxyFetch* proxy_fetch;
   net_instaweb::NgxBaseFetch* base_fetch;
   bool data_received;
@@ -75,6 +75,9 @@ typedef struct {
   ngx_connection_t* pagespeed_connection;
   ngx_http_request_t* r;
   const net_instaweb::ContentType* content_type;
+  bool is_resource_fetch;
+  bool sent_headers;
+  bool write_pending;
 } ngx_http_pagespeed_request_ctx_t;
 
 static ngx_int_t
@@ -113,7 +116,8 @@ static ngx_int_t
 ngx_http_pagespeed_create_connection(ngx_http_pagespeed_request_ctx_t* ctx);
 
 static ngx_int_t
-ngx_http_pagespeed_create_request_context(ngx_http_request_t* r);
+ngx_http_pagespeed_create_request_context(ngx_http_request_t* r,
+                                          bool is_resource_fetch);
 
 static void
 ngx_http_pagespeed_send_to_pagespeed(
@@ -133,6 +137,7 @@ ngx_http_pagespeed_init(ngx_conf_t* cf);
 static ngx_command_t ngx_http_pagespeed_commands[] = {
   { ngx_string("pagespeed"),
     NGX_HTTP_MAIN_CONF|NGX_HTTP_SRV_CONF|NGX_CONF_TAKE1,
+    // TODO(jefftk): replace this callback with an initialization callback.
     ngx_conf_set_flag_slot,
     NGX_HTTP_SRV_CONF_OFFSET,
     offsetof(ngx_http_pagespeed_srv_conf_t, active),
@@ -192,8 +197,10 @@ ngx_http_pagespeed_release_request_context(void* data) {
   ngx_http_pagespeed_request_ctx_t* ctx =
       static_cast<ngx_http_pagespeed_request_ctx_t*>(data);
 
-  // The proxy fetch deleted itself if we called Done(), but if an error
-  // happened before then we need to tell it to delete itself.
+  // proxy_fetch deleted itself if we called Done(), but if an error happened
+  // before then we need to tell it to delete itself.
+  //
+  // If this is a resource fetch then proxy_fetch was never initialized.
   if (ctx->proxy_fetch != NULL) {    
     ctx->proxy_fetch->Done(false /* failure */);
   }
@@ -312,8 +319,7 @@ ngx_http_pagespeed_initialize_server_context(
       cfg->server_context->global_options();
   global_options->SetRewriteLevel(net_instaweb::RewriteOptions::kPassThrough);
   global_options->EnableFiltersByCommaSeparatedList(
-      "collapse_whitespace,remove_comments,remove_quotes,"
-      "inline_css,inline_javascript,rewrite_css", cfg->handler);
+      "extend_cache", cfg->handler);
 }
 
 // Returns:
@@ -347,6 +353,20 @@ ngx_http_pagespeed_update(ngx_http_pagespeed_request_ctx_t* ctx,
   }
 
   // Get output from pagespeed.
+  if (ctx->is_resource_fetch && !ctx->sent_headers) {
+    // For resource fetches, the first pipe-byte tells us headers are available
+    // for fetching.
+    rc = ctx->base_fetch->CollectHeaders(&ctx->r->headers_out);
+    if (rc != NGX_OK) {
+      PDBG(ctx, "problem with CollectHeaders");
+      return rc;
+    }
+
+    ngx_http_send_header(ctx->r);
+    ctx->sent_headers = true;
+  } else {
+    // For proxy fetches and subsequent resource fetch pipe-bytes, the response
+    // body is available for (partial) fetching.
     ngx_chain_t* cl;
     rc = ctx->base_fetch->CollectAccumulatedWrites(&cl);
     if (rc != NGX_OK) {
@@ -354,14 +374,82 @@ ngx_http_pagespeed_update(ngx_http_pagespeed_request_ctx_t* ctx,
       return rc;
     }
 
+    PDBG(ctx, "pagespeed update: %p, done: %d", cl, done);
+
+    // Pass the optimized content along to later body filters.
+    // From Weibin: This function should be called mutiple times. Store the
+    // whole file in one chain buffers is too aggressive. It could consume
+    // too much memory in busy servers.
     rc = ngx_http_next_body_filter(ctx->r, cl);
+    if (rc == NGX_AGAIN && done) {
+      ctx->write_pending = 1;
+      return NGX_OK;
+    }
+
     if (rc != NGX_OK) {
       return rc;
     }
+  }
   
   return done ? NGX_OK : NGX_AGAIN;
 }
 
+static void
+ngx_http_pagespeed_writer(ngx_http_request_t* r)
+{
+    ngx_connection_t* c = r->connection;
+    ngx_event_t* wev = c->write;
+
+    ngx_log_debug2(NGX_LOG_DEBUG_HTTP, wev->log, 0,
+                   "http pagespeed writer handler: \"%V?%V\"",
+                   &r->uri, &r->args);
+
+    if (wev->timedout) {
+      ngx_log_error(NGX_LOG_INFO, c->log, NGX_ETIMEDOUT,
+                    "client timed out");
+      c->timedout = 1;
+
+      ngx_http_finalize_request(r, NGX_HTTP_REQUEST_TIME_OUT);
+      return;
+    }
+
+    int rc = ngx_http_output_filter(r, NULL);
+
+    ngx_log_debug3(NGX_LOG_DEBUG_HTTP, c->log, 0,
+                   "http pagespeed writer output filter: %d, \"%V?%V\"",
+                   rc, &r->uri, &r->args);
+
+    if (rc == NGX_AGAIN) {
+      return;
+    }
+
+    r->write_event_handler = ngx_http_request_empty_handler;
+
+    ngx_http_finalize_request(r, rc);
+}
+
+static ngx_int_t
+ngx_http_set_pagespeed_write_handler(ngx_http_request_t *r)
+{
+    r->http_state = NGX_HTTP_WRITING_REQUEST_STATE;
+
+    r->read_event_handler = ngx_http_request_empty_handler;
+    r->write_event_handler = ngx_http_pagespeed_writer;
+
+    ngx_event_t* wev = r->connection->write;
+
+    ngx_http_core_loc_conf_t* clcf = static_cast<ngx_http_core_loc_conf_t*>(
+        ngx_http_get_module_loc_conf(r, ngx_http_core_module));
+
+    ngx_add_timer(wev, clcf->send_timeout);
+
+    if (ngx_handle_write_event(wev, clcf->send_lowat) != NGX_OK) {
+      return NGX_ERROR;
+    }
+
+    return NGX_OK;
+}
+
 static void
 ngx_http_pagespeed_connection_read_handler(ngx_event_t* ev) {
   CHECK(ev != NULL);
@@ -374,16 +462,29 @@ ngx_http_pagespeed_connection_read_handler(ngx_event_t* ev) {
   CHECK(ctx != NULL);
 
   int rc = ngx_http_pagespeed_update(ctx, ev);
-  CHECK(rc == NGX_OK || rc == NGX_ERROR || rc == NGX_AGAIN);
+  ngx_log_debug1(NGX_LOG_DEBUG_HTTP, ev->log, 0,
+                 "http pagespeed connection read handler rc: %d", rc);
+
   if (rc == NGX_AGAIN) {
-    // request needs more work by pagespeed
+    // Request needs more work by pagespeed.
     rc = ngx_handle_read_event(ev, 0);    
     CHECK(rc == NGX_OK);
-  } else {
+  } else if (rc == NGX_OK) {
+    // Pagespeed is done.  Stop watching the pipe.  If we still have data to
+    // write, set a write handler so we can get called back to make our write.
     ngx_del_event(ev, NGX_READ_EVENT, 0);
     ngx_http_pagespeed_set_buffered(ctx->r, false);
-    ngx_http_finalize_request(
-        ctx->r, rc == NGX_OK ? NGX_DONE : NGX_HTTP_INTERNAL_SERVER_ERROR);
+    if (ctx->write_pending) {
+      if (ngx_http_set_pagespeed_write_handler(ctx->r) != NGX_OK) {
+        ngx_http_finalize_request(ctx->r, NGX_HTTP_INTERNAL_SERVER_ERROR);
+      }
+    } else {
+      ngx_http_finalize_request(ctx->r, NGX_DONE);
+    }
+  } else if (rc == NGX_ERROR) {
+    ngx_http_finalize_request(ctx->r, NGX_HTTP_INTERNAL_SERVER_ERROR);
+  } else {
+    CHECK(false);
   }
 }
 
@@ -417,37 +518,61 @@ ngx_http_pagespeed_create_connection(ngx_http_pagespeed_request_ctx_t* ctx) {
 
 // Set us up for processing a request.
 static ngx_int_t
-ngx_http_pagespeed_create_request_context(ngx_http_request_t* r) {
-  ngx_http_pagespeed_request_ctx_t* ctx =
-      new ngx_http_pagespeed_request_ctx_t();
-  ctx->pipe_fd = -1;
-  ctx->r = r;
-  ctx->cfg = static_cast<ngx_http_pagespeed_srv_conf_t*>(
+ngx_http_pagespeed_create_request_context(ngx_http_request_t* r,
+                                          bool is_resource_fetch) {
+  ngx_http_pagespeed_srv_conf_t* cfg =
+      static_cast<ngx_http_pagespeed_srv_conf_t*>(
           ngx_http_get_module_srv_conf(r, ngx_pagespeed));
 
+  if (cfg->driver_factory == NULL) {
+    // This is the first request handled by this server block.
+    ngx_http_pagespeed_initialize_server_context(cfg);
+  }
+
+  GoogleString url_string = ngx_http_pagespeed_determine_url(r);
+  net_instaweb::GoogleUrl url(url_string);
+
+  if (!url.is_valid()) {
+    ngx_log_error(NGX_LOG_ERR, r->connection->log, 0, "invalid url");
+
+    // Let nginx deal with the error however it wants; we will see a NULL ctx in
+    // the body filter or content handler and do nothing.    
+    return is_resource_fetch ? NGX_DECLINED : NGX_OK;
+  }
+
+  if (is_resource_fetch && !cfg->server_context->IsPagespeedResource(url)) {
+    DBG(r, "Passing on content handling for non-pagespeed resource '%s'",
+        url_string.c_str());
+    return NGX_DECLINED;
+  }
+
   int file_descriptors[2];
   int rc = pipe(file_descriptors);
   if (rc != 0) {
     ngx_log_error(NGX_LOG_ERR, r->connection->log, 0, "pipe() failed");
-    ngx_http_pagespeed_release_request_context(ctx);
     return NGX_ERROR;
   }
 
   if (ngx_nonblocking(file_descriptors[0]) == -1) {
       ngx_log_error(NGX_LOG_EMERG, r->connection->log, ngx_socket_errno,
                     ngx_nonblocking_n " pipe[0] failed");
-      ngx_http_pagespeed_release_request_context(ctx);
       return NGX_ERROR;
   }
 
   if (ngx_nonblocking(file_descriptors[1]) == -1) {
       ngx_log_error(NGX_LOG_EMERG, r->connection->log, ngx_socket_errno,
                     ngx_nonblocking_n " pipe[1] failed");
-      ngx_http_pagespeed_release_request_context(ctx);
       return NGX_ERROR;
   }
 
+  ngx_http_pagespeed_request_ctx_t* ctx =
+      new ngx_http_pagespeed_request_ctx_t();
+  ctx->cfg = cfg;
+  ctx->r = r;
   ctx->pipe_fd = file_descriptors[0];
+  ctx->is_resource_fetch = is_resource_fetch;
+  ctx->write_pending = false;
+
   rc = ngx_http_pagespeed_create_connection(ctx);
   if (rc != NGX_OK) {
     close(file_descriptors[0]);
@@ -465,14 +590,6 @@ ngx_http_pagespeed_create_request_context(ngx_http_request_t* r) {
   // on the proxy fetch below.
   ctx->base_fetch = new net_instaweb::NgxBaseFetch(r, file_descriptors[1]);
 
-  if (ctx->cfg->driver_factory == NULL) {
-    // This is the first request handled by this server block.
-    ngx_http_pagespeed_initialize_server_context(ctx->cfg);
-  }
-
-  GoogleString url_string = ngx_http_pagespeed_determine_url(r);
-  net_instaweb::GoogleUrl request_url(url_string);
-
   // This code is based off of ProxyInterface::ProxyRequestCallback and
   // ProxyFetchFactory::StartNewProxyFetch
 
@@ -492,21 +609,35 @@ ngx_http_pagespeed_create_request_context(ngx_http_request_t* r) {
   // TODO(jefftk): port ProxyInterface::InitiatePropertyCacheLookup so that we
   // have the propery cache in nginx.
 
+  if (is_resource_fetch) {
+    // TODO(jefftk): Set using_spdy appropriately.  See
+    // ProxyInterface::ProxyRequestCallback
+    net_instaweb::ResourceFetch::Start(
+        url, custom_options /* null if there aren't custom options */,
+        false /* using_spdy */, ctx->cfg->server_context, ctx->base_fetch);
+  } else {
     // If we don't have custom options we can use NewRewriteDriver which reuses
     // rewrite drivers and so is faster because there's no wait to construct
     // them.  Otherwise we have to build a new one every time.
+    net_instaweb::RewriteDriver* driver;
     if (custom_options == NULL) {
-    ctx->driver = ctx->cfg->server_context->NewRewriteDriver();
+      driver = ctx->cfg->server_context->NewRewriteDriver();
     } else {
       // NewCustomRewriteDriver takes ownership of custom_options.
-    ctx->driver = ctx->cfg->server_context->NewCustomRewriteDriver(
+      driver = ctx->cfg->server_context->NewCustomRewriteDriver(
           custom_options);
     }
-  ctx->driver->set_log_record(ctx->base_fetch->log_record());
+    driver->set_log_record(ctx->base_fetch->log_record());
     
     // TODO(jefftk): FlushEarlyFlow would go here.
 
-  ngx_http_set_ctx(r, ctx, ngx_pagespeed);
+    // Will call StartParse etc.  The rewrite driver will take care of deleting
+    // itself if necessary.
+    ctx->proxy_fetch = ctx->cfg->proxy_fetch_factory->CreateNewProxyFetch(
+        url_string, ctx->base_fetch, driver,
+        NULL /* property_callback */,
+        NULL /* original_content_fetch */);
+  }
 
   // Set up a cleanup handler on the request.
   ngx_http_cleanup_t* cleanup = ngx_http_cleanup_add(r, 0);
@@ -516,12 +647,7 @@ ngx_http_pagespeed_create_request_context(ngx_http_request_t* r) {
   }
   cleanup->handler = ngx_http_pagespeed_release_request_context;
   cleanup->data = ctx;
-
-  // Will call StartParse etc.
-  ctx->proxy_fetch = ctx->cfg->proxy_fetch_factory->CreateNewProxyFetch(
-      url_string, ctx->base_fetch, ctx->driver,
-      NULL /* property_callback */,
-      NULL /* original_content_fetch */);
+  ngx_http_set_ctx(r, ctx, ngx_pagespeed);
 
   return NGX_OK;
 }
@@ -543,6 +669,7 @@ ngx_http_pagespeed_send_to_pagespeed(
     // pagespeed.
     cur->buf->last_buf = 0;
 
+    CHECK(ctx->proxy_fetch != NULL);
     ctx->proxy_fetch->Write(StringPiece(reinterpret_cast<char*>(cur->buf->pos),
                                         cur->buf->last - cur->buf->pos),
                             ctx->cfg->handler);
@@ -611,6 +738,14 @@ ngx_http_pagespeed_body_filter(ngx_http_request_t* r, ngx_chain_t* in) {
 
 static ngx_int_t
 ngx_http_pagespeed_header_filter(ngx_http_request_t* r) {
+  ngx_http_pagespeed_request_ctx_t* ctx =
+      ngx_http_pagespeed_get_request_context(r);
+
+  if (ctx != NULL) {
+    // ctx will already exist iff this is a pagespeed resource.  Do nothing.
+    return ngx_http_next_header_filter(r);
+  }
+
   // We're modifying content below, so switch to 'Transfer-Encoding: chunked'
   // and calculate on the fly.
   ngx_http_clear_content_length(r);
@@ -627,7 +762,8 @@ ngx_http_pagespeed_header_filter(ngx_http_request_t* r) {
   r->filter_need_in_memory = 1;
 
   if (r->err_status == 0) {
-    int rc = ngx_http_pagespeed_create_request_context(r);
+    int rc = ngx_http_pagespeed_create_request_context(
+        r, false /* not a resource fetch */);
     if (rc != NGX_OK) {
       ngx_http_finalize_request(r, NGX_HTTP_INTERNAL_SERVER_ERROR);
       return rc;
@@ -649,6 +785,30 @@ ngx_http_pagespeed_header_filter(ngx_http_request_t* r) {
   return ngx_http_next_header_filter(r);
 }
 
+// Handle requests for resources like example.css.pagespeed.ce.LyfcM6Wulf.css
+static ngx_int_t
+ngx_http_pagespeed_content_handler(ngx_http_request_t* r) {
+  // TODO(jefftk): return NGX_DECLINED for non-get non-head requests.
+
+  ngx_log_debug1(NGX_LOG_DEBUG_HTTP, r->connection->log, 0,
+                 "http pagespeed handler \"%V\"", &r->uri);
+
+  int rc = ngx_http_pagespeed_create_request_context(
+      r, true /* is a resource fetch */);
+  if (rc != NGX_OK) {
+    return rc;  // rc will be NGX_DECLINED if it's not a pagespeed resource.
+  }
+
+  ngx_http_pagespeed_request_ctx_t* ctx =
+      ngx_http_pagespeed_get_request_context(r);
+  CHECK(ctx != NULL);
+
+  // Tell nginx we're still working on this one.
+  r->count++;
+
+  return NGX_DONE;
+}
+
 static ngx_int_t
 ngx_http_pagespeed_init(ngx_conf_t* cf) {
   ngx_http_pagespeed_srv_conf_t* pagespeed_config;
@@ -661,6 +821,15 @@ ngx_http_pagespeed_init(ngx_conf_t* cf) {
 
     ngx_http_next_body_filter = ngx_http_top_body_filter;
     ngx_http_top_body_filter = ngx_http_pagespeed_body_filter;
+
+    ngx_http_core_main_conf_t* cmcf = static_cast<ngx_http_core_main_conf_t*>(
+        ngx_http_conf_get_module_main_conf(cf, ngx_http_core_module));
+    ngx_http_handler_pt* h = static_cast<ngx_http_handler_pt*>(
+        ngx_array_push(&cmcf->phases[NGX_HTTP_CONTENT_PHASE].handlers));
+    if (h == NULL) {
+      return NGX_ERROR;
+    }
+    *h = ngx_http_pagespeed_content_handler;
   }
 
   return NGX_OK;