feature: implemented the receiveany() upstream cosocket method.

Signed-off-by: Yichun Zhang (agentzh) <[email protected]>

Author: spacewander

README.markdown

@@ -3221,6 +3221,7 @@ Nginx API for Lua
 * [tcpsock:sslhandshake](#tcpsocksslhandshake)
 * [tcpsock:send](#tcpsocksend)
 * [tcpsock:receive](#tcpsockreceive)
+* [tcpsock:receiveany](#tcpsockreceiveany)
 * [tcpsock:receiveuntil](#tcpsockreceiveuntil)
 * [tcpsock:close](#tcpsockclose)
 * [tcpsock:settimeout](#tcpsocksettimeout)
@@ -7002,6 +7003,7 @@ Creates and returns a TCP or stream-oriented unix domain socket object (also kno
 * [settimeout](#tcpsocksettimeout)
 * [settimeouts](#tcpsocksettimeouts)
 * [setoption](#tcpsocksetoption)
+* [receiveany](#tcpsockreceiveany)
 * [receiveuntil](#tcpsockreceiveuntil)
 * [setkeepalive](#tcpsocksetkeepalive)
 * [getreusedtimes](#tcpsockgetreusedtimes)
@@ -7231,6 +7233,40 @@ This feature was first introduced in the `v0.5.0rc1` release.
 
 [Back to TOC](#nginx-api-for-lua)
 
+tcpsock:receiveany
+------------------
+**syntax:** *data, err = tcpsock:receiveany(max)*
+
+**context:** *rewrite_by_lua*, access_by_lua*, content_by_lua*, ngx.timer.*, ssl_certificate_by_lua*, ssl_session_fetch_by_lua**
+
+Returns any data received by the connected socket, at most `max` bytes.
+
+This method is a synchronous operation just like the [send](#tcpsocksend) method and is 100% nonblocking.
+
+In case of success, it returns the data received; in case of error, it returns `nil` with a string describing the error.
+
+If the received data is more than this size, this method will return with exactly this size of data.
+The remaining data in the underlying receive buffer could be returned in the next reading operation.
+
+Timeout for the reading operation is controlled by the [lua_socket_read_timeout](#lua_socket_read_timeout) config directive and the [settimeouts](#tcpsocksettimeouts) method. And the latter takes priority. For example:
+
+```lua
+
+ sock:settimeouts(1000, 1000, 1000)  -- one second timeout for connect/read/write
+ local data, err = sock:receiveany(10 * 1024 * 1024) -- read any data, at most 10K
+ if not data then
+     ngx.say("failed to read any data: ", err)
+     return
+ end
+ ngx.say("successfully read: ", data)
+```
+
+This method doesn't automatically close the current connection when the read timeout error occurs. For other connection errors, this method always automatically closes the connection.
+
+This feature was first introduced in the `v0.10.14` release.
+
+[Back to TOC](#nginx-api-for-lua)
+
 tcpsock:receiveuntil
 --------------------
 **syntax:** *iterator = tcpsock:receiveuntil(pattern, options?)*

config

@@ -361,6 +361,7 @@ HTTP_LUA_SRCS=" \
             $ngx_addon_dir/src/ngx_http_lua_ssl_session_fetchby.c \
             $ngx_addon_dir/src/ngx_http_lua_ssl.c \
             $ngx_addon_dir/src/ngx_http_lua_log_ringbuf.c \
+            $ngx_addon_dir/src/ngx_http_lua_input_filters.c \
             "
 
 HTTP_LUA_DEPS=" \
@@ -422,6 +423,7 @@ HTTP_LUA_DEPS=" \
             $ngx_addon_dir/src/ngx_http_lua_ssl_session_fetchby.h \
             $ngx_addon_dir/src/ngx_http_lua_ssl.h \
             $ngx_addon_dir/src/ngx_http_lua_log_ringbuf.h \
+            $ngx_addon_dir/src/ngx_http_lua_input_filters.h \
             "
 
 CFLAGS="$CFLAGS -DNDK_SET_VAR"

doc/HttpLuaModule.wiki

@@ -5894,6 +5894,7 @@ Creates and returns a TCP or stream-oriented unix domain socket object (also kno
 * [[#tcpsock:settimeout|settimeout]]
 * [[#tcpsock:settimeouts|settimeouts]]
 * [[#tcpsock:setoption|setoption]]
+* [[#tcpsock:receiveany|receiveany]]
 * [[#tcpsock:receiveuntil|receiveuntil]]
 * [[#tcpsock:setkeepalive|setkeepalive]]
 * [[#tcpsock:getreusedtimes|getreusedtimes]]
@@ -6103,6 +6104,36 @@ Since the <code>v0.8.8</code> release, this method no longer automatically close
 
 This feature was first introduced in the <code>v0.5.0rc1</code> release.
 
+== tcpsock:receiveany ==
+'''syntax:''' ''data, err = tcpsock:receiveany(max)''
+
+'''context:''' ''rewrite_by_lua*, access_by_lua*, content_by_lua*, ngx.timer.*, ssl_certificate_by_lua*, ssl_session_fetch_by_lua*''
+
+Returns any data received by the connected socket, at most <code>max</code> bytes.
+
+This method is a synchronous operation just like the [[#tcpsock:send|send]] method and is 100% nonblocking.
+
+In case of success, it returns the data received; in case of error, it returns <code>nil</code> with a string describing the error.
+
+If the received data is more than this size, this method will return with exactly this size of data.
+The remaining data in the underlying receive buffer could be returned in the next reading operation.
+
+Timeout for the reading operation is controlled by the [[#lua_socket_read_timeout|lua_socket_read_timeout]] config directive and the [[#tcpsock:settimeouts|settimeouts]] method. And the latter takes priority. For example:
+
+<geshi lang="lua">
+    sock:settimeouts(1000, 1000, 1000)  -- one second timeout for connect/read/write
+    local data, err = sock:receiveany(10 * 1024 * 1024) -- read any data, at most 10K
+    if not data then
+        ngx.say("failed to read any data: ", err)
+        return
+    end
+    ngx.say("successfully read: ", data)
+</geshi>
+
+This method doesn't automatically close the current connection when the read timeout error occurs. For other connection errors, this method always automatically closes the connection.
+
+This feature was first introduced in the <code>v0.10.14</code> release.
+
 == tcpsock:receiveuntil ==
 '''syntax:''' ''iterator = tcpsock:receiveuntil(pattern, options?)''
 

src/ngx_http_lua_input_filters.c

@@ -0,0 +1,137 @@
+
+/*
+ * Copyright (C) by OpenResty Inc.
+ */
+
+
+#ifndef DDEBUG
+#define DDEBUG 0
+#endif
+#include "ddebug.h"
+
+
+#include "ngx_http_lua_common.h"
+
+
+ngx_int_t
+ngx_http_lua_read_bytes(ngx_buf_t *src, ngx_chain_t *buf_in, size_t *rest,
+    ssize_t bytes, ngx_log_t *log)
+{
+    if (bytes == 0) {
+        return NGX_ERROR;
+    }
+
+    if ((size_t) bytes >= *rest) {
+
+        buf_in->buf->last += *rest;
+        src->pos += *rest;
+        *rest = 0;
+
+        return NGX_OK;
+    }
+
+    /* bytes < *rest */
+
+    buf_in->buf->last += bytes;
+    src->pos += bytes;
+    *rest -= bytes;
+
+    return NGX_AGAIN;
+}
+
+
+ngx_int_t
+ngx_http_lua_read_all(ngx_buf_t *src, ngx_chain_t *buf_in, ssize_t bytes,
+    ngx_log_t *log)
+{
+    if (bytes == 0) {
+        return NGX_OK;
+    }
+
+    buf_in->buf->last += bytes;
+    src->pos += bytes;
+
+    return NGX_AGAIN;
+}
+
+
+ngx_int_t
+ngx_http_lua_read_any(ngx_buf_t *src, ngx_chain_t *buf_in, size_t *max,
+    ssize_t bytes, ngx_log_t *log)
+{
+    if (bytes == 0) {
+        return NGX_ERROR;
+    }
+
+    if (bytes >= (ssize_t) *max) {
+        bytes = (ssize_t) *max;
+    }
+
+    buf_in->buf->last += bytes;
+    src->pos += bytes;
+
+    return NGX_OK;
+}
+
+
+ngx_int_t
+ngx_http_lua_read_line(ngx_buf_t *src, ngx_chain_t *buf_in, ssize_t bytes,
+    ngx_log_t *log)
+{
+    u_char                      *dst;
+    u_char                       c;
+#if (NGX_DEBUG)
+    u_char                      *begin;
+#endif
+
+#if (NGX_DEBUG)
+    begin = src->pos;
+#endif
+
+    if (bytes == 0) {
+        return NGX_ERROR;
+    }
+
+    dd("already read: %p: %.*s", buf_in,
+       (int) (buf_in->buf->last - buf_in->buf->pos), buf_in->buf->pos);
+
+    dd("data read: %.*s", (int) bytes, src->pos);
+
+    dst = buf_in->buf->last;
+
+    while (bytes--) {
+
+        c = *src->pos++;
+
+        switch (c) {
+        case '\n':
+            ngx_log_debug2(NGX_LOG_DEBUG_HTTP, log, 0,
+                           "lua read the final line part: \"%*s\"",
+                           src->pos - 1 - begin, begin);
+
+            buf_in->buf->last = dst;
+
+            dd("read a line: %p: %.*s", buf_in,
+               (int) (buf_in->buf->last - buf_in->buf->pos), buf_in->buf->pos);
+
+            return NGX_OK;
+
+        case '\r':
+            /* ignore it */
+            break;
+
+        default:
+            *dst++ = c;
+            break;
+        }
+    }
+
+#if (NGX_DEBUG)
+    ngx_log_debug2(NGX_LOG_DEBUG_HTTP, log, 0,
+                   "lua read partial line data: %*s", dst - begin, begin);
+#endif
+
+    buf_in->buf->last = dst;
+
+    return NGX_AGAIN;
+}

src/ngx_http_lua_input_filters.h

@@ -0,0 +1,29 @@
+
+/*
+ * Copyright (C) by OpenResty Inc.
+ */
+
+
+#ifndef _NGX_HTTP_LUA_INPUT_FILTERS_H_INCLUDED_
+#define _NGX_HTTP_LUA_INPUT_FILTERS_H_INCLUDED_
+
+
+#include "ngx_http_lua_common.h"
+
+
+ngx_int_t ngx_http_lua_read_bytes(ngx_buf_t *src, ngx_chain_t *buf_in,
+    size_t *rest, ssize_t bytes, ngx_log_t *log);
+
+ngx_int_t ngx_http_lua_read_all(ngx_buf_t *src, ngx_chain_t *buf_in,
+    ssize_t bytes, ngx_log_t *log);
+
+ngx_int_t ngx_http_lua_read_any(ngx_buf_t *src, ngx_chain_t *buf_in,
+    size_t *max, ssize_t bytes, ngx_log_t *log);
+
+ngx_int_t ngx_http_lua_read_line(ngx_buf_t *src, ngx_chain_t *buf_in,
+    ssize_t bytes, ngx_log_t *log);
+
+
+#endif /* _NGX_HTTP_LUA_INPUT_FILTERS_H_INCLUDED_ */
+
+/* vi:set ft=c ts=4 sw=4 et fdm=marker: */