add make_ngx_resp_writer

* add make_ngx_resp_writer

* add test of make_ngx_resp_writer

* make_ngx_writer add an optional parameter

Author: siyuan

lib/pipe/writer.lua

@@ -1,6 +1,8 @@
 local httpclient = require("acid.httpclient")
 local tableutil = require("acid.tableutil")
 local strutil = require("acid.strutil")
+local rpc_logging = require("acid.rpc_logging")
+local acid_setutil = require("acid.setutil")
 
 local _M = { _VERSION = '1.0' }
 
@@ -9,6 +11,84 @@ local to_str = strutil.to_str
 local BLOCK_SIZE = 1024 * 1024
 local SOCKET_TIMEOUTS = {5 * 1000, 100 * 1000, 100 * 1000}
 
+local INF = math.huge
+
+
+local function write_data_to_ngx(pobj, ident, opts)
+    -- range = {start, end}
+    -- range is a closed interval.
+    local range = opts.range
+    local pipe_log = opts.pipe_log
+
+    local recv_left, recv_right = 0, 0
+    local from, to
+    if range ~= nil then
+        from = range['start'] + 1
+
+        if range['end'] ~= nil then
+            to = range['end'] + 1
+        else
+            to = INF
+        end
+
+        if from > to then
+            return nil, 'InvalidRange', string.format(
+                'start: %d is greater than end: %d', from, to)
+        end
+    end
+
+    while true do
+        local data, err, err_msg
+        if pipe_log ~= nil then
+            rpc_logging.reset_start(pipe_log)
+
+            data, err, err_msg = pobj:read_pipe(ident)
+
+            rpc_logging.set_err(pipe_log, err)
+            rpc_logging.incr_stat(pipe_log, 'downstream', 'sendbody', #(data or ''))
+        else
+            data, err, err_msg = pobj:read_pipe(ident)
+        end
+
+        if err ~= nil then
+            return nil, err, err_msg
+        end
+
+        if data == '' then
+            break
+        end
+
+        recv_left = recv_right + 1
+        recv_right = recv_right + #data
+
+        if from ~= nil then
+            local intersection, err, err_msg =
+                acid_setutil.intersect(from, to, recv_left, recv_right)
+            if err ~= nil then
+                return nil, err, err_msg
+            end
+
+            local f, t = intersection.from, intersection.to
+            if f == nil then
+                data = ''
+            else
+                if t - f + 1 ~= #data then
+                    data = data:sub(f - recv_left + 1, t - recv_left + 1)
+                end
+            end
+        end
+
+        ngx.print(data)
+        local _, err = ngx.flush(true)
+        if err then
+            return nil, 'ClientAborted', err
+        end
+    end
+
+    return recv_right
+end
+
+
 function _M.connect_http(ips, port, verb, uri, opts)
     opts = opts or {}
 
@@ -172,30 +252,25 @@ function _M.make_http_writer(ips, port, verb, uri, opts)
     end
 end
 
-function _M.make_ngx_writer()
-    return function(pobj, ident)
-        local bytes = 0
+function _M.make_ngx_writer(opts)
+    opts = opts or {}
 
-        while true do
-            local data, err_code, err_msg = pobj:read_pipe(ident)
-            if err_code ~= nil then
-                return nil, err_code, err_msg
-            end
+    return function(pobj, ident)
+        return write_data_to_ngx(pobj, ident, opts)
+    end
+end
 
-            if data == '' then
-                break
-            end
 
-            bytes = bytes + #data
+function _M.make_ngx_resp_writer(status, headers, opts)
+    opts = opts or {}
 
-            ngx.print(data)
-            local _, err = ngx.flush(true)
-            if err then
-                return nil, 'ClientAborted', err
-            end
-        end
+    ngx.status = status
+    for k, v in pairs(headers) do
+        ngx.header[k] = v
+    end
 
-        return bytes
+    return function(pobj, ident)
+        return write_data_to_ngx(pobj, ident, opts)
     end
 end
 

t/test_ngx_resp_writer.t

@@ -0,0 +1,54 @@
+# vim:set ft=lua ts=4 sw=4 et ft=perl:
+
+use Test::Nginx::Socket "no_plan";
+
+no_long_string();
+
+run_tests();
+
+__DATA__
+
+=== TEST 1: test make_ngx_resp_writer
+--- http_config
+    lua_check_client_abort on;
+
+    lua_package_path "./lib/?.lua;;";
+    lua_package_cpath "./lib/?.so;;";
+--- config
+    location /t {
+        content_by_lua '
+            local pipe_pipe = require("pipe.pipe")
+
+            local function is_running()
+                return true
+            end
+
+            local read_datas = {"1", "23", "4", "5", "678", "9", "0"}
+            local readers = {pipe_pipe.reader.make_memery_reader(read_datas)}
+
+            local status = 200
+            local headers = {["Content-Length"]=3}
+            local opts = {range ={}}
+            opts.range["start"] = 2
+            opts.range["end"] = 4
+
+            local writers = {pipe_pipe.writer.make_ngx_resp_writer(status, headers, opts)}
+
+            local cpipe, err, err_msg = pipe_pipe:new(readers, writers)
+            if err ~= nil then
+                ngx.log(ngx.ERR, string.format("err: %s, err_msg: %s", err, err_msg))
+                return
+            end
+
+            local _, err, err_msg = cpipe:pipe(is_running)
+            if err ~= nil then
+                ngx.log(ngx.ERR, string.format("err: %s, err_msg: %s", err, err_msg))
+                return
+            end
+        ';
+    }
+--- request
+GET /t
+--- response_body: 345
+--- no_error_log
+[error]