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 <[email protected]>
Signed-off-by: Alejandro Colomar <[email protected]>
Signed-off-by: Andrew Clayton <[email protected]>

Author: ac000

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;
 };
@@ -166,6 +167,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,
@@ -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;
     }

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);

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