Implement periodic flushing

Author: calio

README.md

@@ -127,6 +127,10 @@ Available user configurations are listed as follows:
 
     Keepalive pool size used by sock:keepalive. Default to 10.
 
+* `periodic_flush`
+
+    Periodic flush interval (in seconds). Set to `nil` to turn off this feature.
+
 [Back to TOC](#table-of-contents)
 
 initted

lib/resty/logger/socket.lua

@@ -52,6 +52,8 @@ local port
 local path
 local max_buffer_reuse      = 10000        -- reuse buffer for at most 10000
                                            -- times
+local periodic_flush        = nil
+local need_periodic_flush   = nil
 
 -- internal variables
 local buffer_size           = 0
@@ -77,7 +79,6 @@ local logger_initted
 local counter               = 0
 
 
-
 local function _write_error(msg)
     last_error = msg
 end
@@ -279,6 +280,10 @@ local function _flush()
                         .. err
         _write_error(err_msg)
         return nil, err_msg
+    else
+        if debug then
+            ngx_log(DEBUG, "send " .. bytes .. " bytes")
+        end
     end
 
     buffer_size = buffer_size - #send_buffer
@@ -287,8 +292,29 @@ local function _flush()
     return bytes
 end
 
+local function _periodic_flush()
+    if need_periodic_flush then
+        -- no regular flush happened after periodic flush timer had been set
+        if debug then
+            ngx_log(DEBUG, "performing periodic flush")
+        end
+        _flush()
+    else
+        if debug then
+            ngx_log(DEBUG, "no need to perform periodic flush: regular flush "
+                    .. "happened before")
+        end
+        need_periodic_flush = true
+    end
+
+    timer_at(periodic_flush, _periodic_flush)
+end
+
 local function _flush_buffer()
     local ok, err = timer_at(0, _flush)
+
+    need_periodic_flush = false
+
     if not ok then
         _write_error(err)
         return nil, err
@@ -332,6 +358,8 @@ function _M.init(user_config)
             pool_size = v
         elseif k == "max_buffer_reuse" then
             max_buffer_reuse = v
+        elseif k == "periodic_flush" then
+            periodic_flush = v
         end
     end
 
@@ -355,6 +383,15 @@ function _M.init(user_config)
 
     logger_initted = true
 
+    if periodic_flush then
+        if debug then
+            ngx_log(DEBUG, "periodic flush enabled for every "
+                    .. periodic_flush .. " milliseconds")
+        end
+        need_periodic_flush = true
+        timer_at(periodic_flush, _periodic_flush)
+    end
+
     return logger_initted
 end
 

t/sanity.t

@@ -648,3 +648,54 @@ log buffer reuse limit (1) reached, create a new "log_buffer_data"
 --- response_body
 foo
 wrote bytes: 15
+
+
+
+=== TEST 16: flush periodically
+--- http_config eval: $::HttpConfig
+--- config
+    location /t {
+        content_by_lua '
+            ngx.say("foo")
+            local logger = require "resty.logger.socket"
+            if not logger.initted() then
+                local ok, err = logger.init{
+                    host = "127.0.0.1",
+                    port = 29999,
+                    flush_limit = 1000,
+                    drop_limit = 10000,
+                    retry_interval = 1,
+                    timeout = 50,
+                    max_buffer_reuse = 100,
+                    periodic_flush = 0.03, -- 0.03s
+                }
+            end
+
+            local bytes, err
+            bytes, err = logger.log("foo")
+            if err then
+                ngx.log(ngx.ERR, err)
+            end
+            ngx.say("wrote bytes: ", bytes)
+
+            ngx.sleep(0.05)
+
+            bytes, err = logger.log("bar")
+            if err then
+                ngx.log(ngx.ERR, err)
+            end
+            ngx.say("wrote bytes: ", bytes)
+            ngx.sleep(0.05)
+        ';
+    }
+--- request
+GET /t
+--- wait: 0.1
+--- tcp_listen: 29999
+--- tcp_reply:
+--- tcp_query: foobar
+--- tcp_query_len: 6
+--- response_body
+foo
+wrote bytes: 3
+wrote bytes: 3