http: Compress application responses

This adds initial support for compressing application responses.

A couple of things to note

1) Compressed responses are sent 'chunked' as we don't know beforehand
   how large the compressed response will be.

2) We only compress responses where we know the Content-Length as we
   need to check with the 'min_length' config parameter. It's also
   currently how we track when we need to close the compression stream
   off.

Co-authored-by: Alejandro Colomar <alx@kernel.org>
Signed-off-by: Alejandro Colomar <alx@kernel.org>
Signed-off-by: Andrew Clayton <a.clayton@nginx.com>

3 files changed, 102 insertions, 1 deletions

diff --git a/src/nxt_http_compression.c b/src/nxt_http_compression.c
index 96882307..28e53a9d 100644
--- a/src/nxt_http_compression.c
+++ b/src/nxt_http_compression.c
@@ -67,6 +67,7 @@ struct nxt_http_comp_ctx_s {
     nxt_uint_t                      idx;
 
     nxt_off_t                       resp_clen;
+    nxt_off_t                       clen_sent;
 
     nxt_http_comp_compressor_ctx_t  ctx;
 };
@@ -167,6 +168,70 @@ nxt_http_comp_bound(size_t size)
 
 
 nxt_int_t
+nxt_http_comp_compress_app_response(nxt_task_t *task, nxt_http_request_t *r,
+                                    nxt_buf_t **b)
+{
+    bool                 last;
+    size_t               buf_len;
+    ssize_t              cbytes;
+    nxt_buf_t            *buf;
+    nxt_off_t            in_len;
+    nxt_http_comp_ctx_t  *ctx = nxt_http_comp_ctx();
+
+    if (ctx->idx == NXT_HTTP_COMP_SCHEME_IDENTITY) {
+        return NXT_OK;
+    }
+
+    if (!nxt_buf_is_port_mmap(*b)) {
+        return NXT_OK;
+    }
+
+    in_len = (*b)->mem.free - (*b)->mem.pos;
+    buf_len = nxt_http_comp_bound(in_len);
+
+    buf = nxt_buf_mem_ts_alloc(task, (*b)->data, buf_len);
+    if (nxt_slow_path(buf == NULL)) {
+        return NXT_ERROR;
+    }
+
+    buf->data = (*b)->data;
+    buf->parent = (*b)->parent;
+
+    last = ctx->clen_sent + in_len == ctx->resp_clen;
+
+    cbytes = nxt_http_comp_compress(buf->mem.start, buf_len,
+                                    (*b)->mem.pos, in_len, last);
+    if (cbytes == -1) {
+        nxt_buf_free(buf->data, buf);
+        return NXT_ERROR;
+    }
+
+    buf->mem.free += cbytes;
+
+    ctx->clen_sent += in_len;
+
+#define nxt_swap_buf(db, sb)                                                  \
+    do {                                                                      \
+        nxt_buf_t  **db_ = (db);                                              \
+        nxt_buf_t  **sb_ = (sb);                                              \
+        nxt_buf_t  *tmp_;                                                     \
+                                                                              \
+        tmp_ = *db_;                                                          \
+        *db_ = *sb_;                                                          \
+        *sb_ = tmp_;                                                          \
+    } while (0)
+
+    nxt_swap_buf(b, &buf);
+
+#undef nxt_swap_buf
+
+    nxt_buf_free(buf->data, buf);
+
+    return NXT_OK;
+}
+
+
+nxt_int_t
 nxt_http_comp_compress_static_response(nxt_task_t *task, nxt_file_t **f,
                                        nxt_file_info_t *fi,
                                        size_t static_buf_len,
@@ -407,6 +472,26 @@ nxt_http_comp_set_header(nxt_http_request_t *r, nxt_uint_t comp_idx)
     f->value = token->start;
     f->value_length = token->length;
 
+    r->resp.content_length = NULL;
+    r->resp.content_length_n = -1;
+
+    if (r->resp.mime_type == NULL) {
+        nxt_http_field_t *f;
+
+        /*
+         * As per RFC 2616 section 4.4 item 3, you should not send
+         * Content-Length when a Transfer-Encoding header is present.
+         */
+        nxt_list_each(f, r->resp.fields) {
+            if (nxt_strcasecmp(f->name,
+                               (const u_char *)"Content-Length") == 0)
+            {
+                f->skip = true;
+                break;
+            }
+        } nxt_list_loop;
+    }
+
     return NXT_OK;
 }
 
@@ -443,6 +528,10 @@ nxt_http_comp_check_compression(nxt_task_t *task, nxt_http_request_t *r)
         return NXT_OK;
     }
 
+    if (r->resp.content_length == NULL && r->resp.content_length_n == -1) {
+        return NXT_OK;
+    }
+
     if (r->resp.content_length_n == 0) {
         return NXT_OK;
     }
diff --git a/src/nxt_http_compression.h b/src/nxt_http_compression.h
index 9f3567c1..f178e984 100644
--- a/src/nxt_http_compression.h
+++ b/src/nxt_http_compression.h
@@ -90,6 +90,8 @@ extern const nxt_http_comp_operations_t  nxt_http_comp_brotli_ops;
 #endif
 
 
+extern nxt_int_t nxt_http_comp_compress_app_response(nxt_task_t *task,
+    nxt_http_request_t *r, nxt_buf_t **b);
 extern nxt_int_t nxt_http_comp_compress_static_response(nxt_task_t *task,
     nxt_file_t **f, nxt_file_info_t *fi, size_t static_buf_len,
     size_t *out_total);
diff --git a/src/nxt_router.c b/src/nxt_router.c
index d2486c9f..af9dad29 100644
--- a/src/nxt_router.c
+++ b/src/nxt_router.c
@@ -4191,8 +4191,13 @@ nxt_router_response_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
 
     if (r->header_sent) {
         nxt_buf_chain_add(&r->out, b);
-        nxt_http_request_send_body(task, r, NULL);
 
+        ret = nxt_http_comp_compress_app_response(task, r, &r->out);
+        if (ret == NXT_ERROR) {
+            goto fail;
+        }
+
+        nxt_http_request_send_body(task, r, NULL);
     } else {
         b_size = nxt_buf_is_mem(b) ? nxt_buf_mem_used_size(&b->mem) : 0;
 
@@ -4272,6 +4277,11 @@ nxt_router_response_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
             nxt_buf_chain_add(&r->out, b);
         }
 
+        ret = nxt_http_comp_check_compression(task, r);
+        if (ret != NXT_OK) {
+            goto fail;
+        }
+
         nxt_http_request_header_send(task, r, nxt_http_request_send_body, NULL);
 
         if (r->websocket_handshake