balancer_by_lua for streams

.travis.yml

@@ -42,6 +42,7 @@ install:
   - git clone https://github.com/openresty/nginx-devel-utils.git
   - git clone https://github.com/simpl/ngx_devel_kit.git ../ndk-nginx-module
   - git clone https://github.com/openresty/lua-nginx-module.git ../lua-nginx-module
+  - git clone https://github.com/openresty/stream-lua-nginx-module.git ../stream-lua-nginx-module
   - git clone https://github.com/openresty/no-pool-nginx.git ../no-pool-nginx
   - git clone https://github.com/openresty/echo-nginx-module.git ../echo-nginx-module
   - git clone https://github.com/openresty/lua-resty-lrucache.git
@@ -68,7 +69,7 @@ script:
   - export LD_LIBRARY_PATH=$PWD/mockeagain:$LD_LIBRARY_PATH
   - export TEST_NGINX_RESOLVER=8.8.4.4
   - export NGX_BUILD_CC=$CC
-  - ngx-build $NGINX_VERSION --with-ipv6 --with-http_realip_module --with-http_ssl_module --with-cc-opt="-I$OPENSSL_INC" --with-ld-opt="-L$OPENSSL_LIB -Wl,-rpath,$OPENSSL_LIB" --add-module=../ndk-nginx-module --add-module=../echo-nginx-module --add-module=../headers-more-nginx-module --add-module=../lua-nginx-module --with-debug > build.log 2>&1 || (cat build.log && exit 1)
+  - ngx-build $NGINX_VERSION --with-ipv6 --with-http_realip_module  --with-http_ssl_module --with-stream --with-stream_ssl_module --with-cc-opt="-I$OPENSSL_INC" --with-ld-opt="-L$OPENSSL_LIB -Wl,-rpath,$OPENSSL_LIB" --add-module=../ndk-nginx-module --add-module=../echo-nginx-module --add-module=../headers-more-nginx-module --add-module=../lua-nginx-module --add-module=../stream-lua-nginx-module --with-debug > build.log 2>&1 || (cat build.log && exit 1)
   - nginx -V
   - ldd `which nginx`|grep -E 'luajit|ssl|pcre'
   - prove -r t

Makefile

@@ -11,10 +11,11 @@ all: ;
 
 install: all
 	$(INSTALL) -d $(DESTDIR)$(LUA_LIB_DIR)/resty/core/
-	$(INSTALL) -d $(DESTDIR)$(LUA_LIB_DIR)/ngx/
+	$(INSTALL) -d $(DESTDIR)$(LUA_LIB_DIR)/ngx/balancer/
 	$(INSTALL) lib/resty/*.lua $(DESTDIR)$(LUA_LIB_DIR)/resty/
 	$(INSTALL) lib/resty/core/*.lua $(DESTDIR)$(LUA_LIB_DIR)/resty/core/
 	$(INSTALL) lib/ngx/*.lua $(DESTDIR)$(LUA_LIB_DIR)/ngx/
+	$(INSTALL) lib/ngx/balancer/*.lua $(DESTDIR)$(LUA_LIB_DIR)/ngx/balancer/
 
 test: all
 	PATH=$(OPENRESTY_PREFIX)/nginx/sbin:$$PATH prove -I../test-nginx/lib -r t

lib/ngx/balancer.lua

@@ -1,10 +1,12 @@
 -- Copyright (C) Yichun Zhang (agentzh)
 
+if ngx.config.subsystem == "stream" then
+   return require "ngx.balancer.stream"
+end
 
 local ffi = require "ffi"
 local base = require "resty.core.base"
 
-
 local C = ffi.C
 local ffi_str = ffi.string
 local errmsg = base.get_errmsg_ptr()

lib/ngx/balancer/stream.lua

@@ -0,0 +1,109 @@
+-- Copyright (C) Shriram Rajagopalan
+-- I hereby assign copyright in this code to the lua-resty-core project,
+-- to be licensed under the same terms as the rest of the code.
+
+local ffi = require "ffi"
+local base = require "resty.core.base"
+
+
+local C = ffi.C
+local ffi_str = ffi.string
+local errmsg = base.get_errmsg_ptr()
+local FFI_OK = base.FFI_OK
+local FFI_ERROR = base.FFI_ERROR
+local int_out = ffi.new("int[1]")
+local getfenv = getfenv
+local error = error
+local type = type
+local tonumber = tonumber
+
+
+ffi.cdef[[
+int ngx_stream_lua_ffi_balancer_set_current_peer(ngx_stream_session_t *r,
+    const unsigned char *addr, size_t addr_len, int port, char **err);
+
+int ngx_stream_lua_ffi_balancer_set_more_tries(ngx_stream_session_t *r,
+    int count, char **err);
+
+int ngx_stream_lua_ffi_balancer_get_last_failure(ngx_stream_session_t *r,
+    int *status, char **err);
+]]
+
+
+local peer_state_names = {
+    [1] = "keepalive",
+    [2] = "next",
+    [4] = "failed",
+}
+
+
+local _M = { version = base.version }
+
+
+function _M.set_current_peer(addr, port)
+    local r = getfenv(0).__ngx_sess
+    if not r then
+        return error("no session found")
+    end
+
+    if not port then
+        port = 0
+    elseif type(port) ~= "number" then
+        port = tonumber(port)
+    end
+
+    local rc = C.ngx_stream_lua_ffi_balancer_set_current_peer(r, addr, #addr,
+                                                            port, errmsg)
+    if rc == FFI_OK then
+        return true
+    end
+
+    return nil, ffi_str(errmsg[0])
+end
+
+
+function _M.set_more_tries(count)
+    local r = getfenv(0).__ngx_sess
+    if not r then
+        return error("no session found")
+    end
+
+    local rc = C.ngx_stream_lua_ffi_balancer_set_more_tries(r, count, errmsg)
+    if rc == FFI_OK then
+        if errmsg[0] == nil then
+            return true
+        end
+        return true, ffi_str(errmsg[0])  -- return the warning
+    end
+
+    return nil, ffi_str(errmsg[0])
+end
+
+
+function _M.get_last_failure()
+    local r = getfenv(0).__ngx_sess
+    if not r then
+        return error("no session found")
+    end
+
+    local state = C.ngx_stream_lua_ffi_balancer_get_last_failure(r,
+                                                                 int_out,
+                                                                 errmsg)
+
+    if state == 0 then
+        return nil
+    end
+
+    if state == FFI_ERROR then
+        return nil, nil, ffi_str(errmsg[0])
+    end
+
+    return peer_state_names[state] or "unknown", int_out[0]
+end
+
+function _M.set_timeouts(connect_timeout, send_timeout, read_timeout)
+   return error("not implemented")
+end
+
+
+return _M

lib/resty/core/base.lua

@@ -79,6 +79,14 @@ if not pcall(ffi.typeof, "ngx_http_request_t") then
 end
 
 
+if not pcall(ffi.typeof, "ngx_stream_session_t") then
+    ffi.cdef[[
+        struct ngx_stream_session_s;
+        typedef struct ngx_stream_session_s  ngx_stream_session_t;
+    ]]
+end
+
+
 if not pcall(ffi.typeof, "ngx_http_lua_ffi_str_t") then
     ffi.cdef[[
         typedef struct {