[WIP] Compress application responses

Author: ac000

src/nxt_http_compression.c

@@ -140,6 +140,52 @@ static void print_comp_config(size_t n)
     }
 }
 
+nxt_int_t
+nxt_http_comp_compress_response(nxt_http_request_t *r)
+{
+    nxt_buf_t *b = r->out;
+    size_t in_len;
+    size_t buf_len;
+    ssize_t cbytes;
+    uint8_t *buf;
+    bool last = false;
+    nxt_http_comp_compressor_t  *compressor;
+    const nxt_http_comp_operations_t  *cops;
+    const nxt_http_comp_ctx_t *ctx = &compressor_ctx;
+
+    printf("%s: \n", __func__);
+
+    if (ctx->idx == NXT_HTTP_COMP_SCHEME_IDENTITY) {
+        printf("%s: NXT_HTTP_COMP_SCHEME_IDENTITY [skipping/identity]\n",
+               __func__);
+        return NXT_OK;
+    }
+
+    if (b->mem.pos == NULL) {
+        return NXT_OK;
+    }
+
+    compressor = &enabled_compressors[ctx->idx];
+
+    in_len = b->mem.free - b->mem.pos;
+
+    last = !b->next || (b->next && b->next->is_last == 1);
+
+    cops = compressor->type->cops;
+
+    buf_len = cops->bound(&ctx->ctx, in_len);
+    buf = nxt_malloc(buf_len);
+    cbytes = cops->deflate(&ctx->ctx, b->mem.pos, in_len, buf, buf_len, last);
+    printf("%s: cbytes = %ld\n", __func__, cbytes);
+    /* TODO handle new buffer is larger than original buffer */
+    if (cbytes != -1) {
+        b->mem.free = nxt_cpymem(b->mem.pos, buf, cbytes);
+    }
+    nxt_free(buf);
+
+    return NXT_OK;
+}
+
 bool
 nxt_http_comp_wants_compression(void)
 {
@@ -284,6 +330,14 @@ nxt_http_comp_set_header(nxt_http_request_t *r, nxt_uint_t comp_idx)
     static const nxt_str_t  content_encoding_str =
                                     nxt_string("Content-Encoding");
 
+    printf("%s: \n", __func__);
+
+#if 0
+    if (comp_idx == NXT_HTTP_COMP_SCHEME_IDENTITY) {
+        return NXT_OK;
+    }
+#endif
+
     f = nxt_list_add(r->resp.fields);
     if (nxt_slow_path(f == NULL)) {
         return NXT_ERROR;
@@ -298,6 +352,30 @@ 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)
+            {
+                continue;
+            }
+
+            printf("%s: Found (%s: %s), marking as 'skip'\n", __func__,
+                   f->name, f->value);
+            f->skip = true;
+            break;
+        } nxt_list_loop;
+    }
+
     return NXT_OK;
 }
 

src/nxt_http_compression.h

@@ -81,6 +81,7 @@ extern const nxt_http_comp_operations_t  nxt_comp_zstd_ops;
 extern const nxt_http_comp_operations_t  nxt_comp_brotli_ops;
 #endif
 
+extern nxt_int_t nxt_http_comp_compress_response(nxt_http_request_t *r);
 extern bool nxt_http_comp_wants_compression(void);
 extern size_t nxt_http_comp_bound(size_t size);
 extern ssize_t nxt_http_comp_compress(uint8_t *dst, size_t dst_size,

src/nxt_port_queue.h

@@ -81,6 +81,8 @@ nxt_port_queue_recv(nxt_port_queue_t volatile *q, void *p)
     nxt_nncq_atomic_t      i;
     nxt_port_queue_item_t  *qi;
 
+//    printf("%s \n", __func__);
+
     i = nxt_nncq_dequeue(&q->queue);
     if (i == nxt_nncq_empty(&q->queue)) {
         return -1;

src/nxt_router.c

@@ -4086,6 +4086,8 @@ nxt_router_response_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
     nxt_unit_response_t     *resp;
     nxt_request_rpc_data_t  *req_rpc_data;
 
+    printf("%s: \n", __func__);
+
     req_rpc_data = data;
 
     r = req_rpc_data->request;
@@ -4141,8 +4143,11 @@ 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);
 
+        /* XXX Do compression here */
+        nxt_http_comp_compress_response(r);
+
+        nxt_http_request_send_body(task, r, NULL);
     } else {
         b_size = nxt_buf_is_mem(b) ? nxt_buf_mem_used_size(&b->mem) : 0;
 
@@ -4222,6 +4227,12 @@ nxt_router_response_ready_handler(nxt_task_t *task, nxt_port_recv_msg_t *msg,
             nxt_buf_chain_add(&r->out, b);
         }
 
+        /* XXX Check compression / modify headers here */
+        ret = nxt_http_comp_check_compression(task, r);
+        if (nxt_slow_path(ret != NXT_OK)) {
+            goto fail;
+        }
+
         nxt_http_request_header_send(task, r, nxt_http_request_send_body, NULL);
 
         if (r->websocket_handshake