Add max_requests pool option

Pools accept a max_requests option (default infinity), passed
through to shackle 0.8.0's client option of the same name: the
connection is closed and reopened after that many requests, capping
how long a keepalive connection is reused.

The eunit HTTP fixture counts accepted connections so the test can
assert that a pool_size 1 / max_requests 2 pool recycles instead of
holding one connection.

Closes #4

Author: lpgauth

README.md

@@ -72,6 +72,13 @@ ok
     <td>1024</td>
     <td>maximum number of concurrent requests per connection</td>
   </tr>
+  <tr>
+    <td>max_requests</td>
+    <td>pos_integer() | infinity</td>
+    <td>infinity</td>
+    <td>number of requests after which a connection is closed and
+    reopened</td>
+  </tr>
   <tr>
     <td>pool_size</td>
     <td>pos_integer()</td>

include/buoy.hrl

@@ -45,6 +45,7 @@
 -type hostname()      :: binary().
 -type method()        :: get | head | post | put | {custom, binary()}.
 -type option()        :: {backlog_size, pos_integer()} |
+                         {max_requests, pos_integer() | infinity} |
                          {pool_size, pos_integer()} |
                          {pool_strategy, random | round_robin} |
                          {protocol, shackle_socket | shackle_tcp} |

include/buoy_internal.hrl

@@ -13,6 +13,7 @@
 -define(DEFAULT_BODY, undefined).
 -define(DEFAULT_HEADERS, []).
 -define(DEFAULT_IP, "127.0.0.1").
+-define(DEFAULT_MAX_REQUESTS, infinity).
 -define(DEFAULT_PID, self()).
 -define(DEFAULT_POOL_OPTIONS, []).
 -define(DEFAULT_POOL_SIZE, 16).

src/buoy_pool.erl

@@ -98,6 +98,7 @@ terminate() ->
 
 %% private
 client_options(Protocol, Hostname, Port, Options) ->
+    MaxRequests = ?LOOKUP(max_requests, Options, ?DEFAULT_MAX_REQUESTS),
     Reconnect = ?LOOKUP(reconnect, Options, ?DEFAULT_RECONNECT),
     ReconnectTimeMax = ?LOOKUP(reconnect_time_max, Options,
         ?DEFAULT_RECONNECT_MAX),
@@ -107,6 +108,7 @@ client_options(Protocol, Hostname, Port, Options) ->
         ?DEFAULT_SOCKET_OPTIONS),
 
     [{ip, binary_to_list(Hostname)},
+     {max_requests, MaxRequests},
      {port, Port},
      {protocol, shackle_protocol(Protocol, Options)},
      {reconnect, Reconnect},

test/buoy_http_server.erl

@@ -1,13 +1,17 @@
 -module(buoy_http_server).
 
 -export([
+    connection_count/0,
     start/0,
     stop/0
 ]).
 
 -define(PORT, 8080).
 
 %% public
+connection_count() ->
+    counters:get(persistent_term:get({?MODULE, connections}), 1).
+
 start() ->
     Self = self(),
     Pid = spawn(fun () -> init(Self) end),
@@ -30,6 +34,7 @@ stop() ->
 %% private
 init(Parent) ->
     register(?MODULE, self()),
+    persistent_term:put({?MODULE, connections}, counters:new(1, [])),
     {ok, LSocket} = gen_tcp:listen(?PORT, [
         binary,
         {active, false},
@@ -42,6 +47,7 @@ init(Parent) ->
 
 accept(LSocket) ->
     {ok, Socket} = gen_tcp:accept(LSocket),
+    counters:add(persistent_term:get({?MODULE, connections}), 1, 1),
     Pid = spawn_link(fun () ->
         receive go -> connection(Socket) end
     end),

test/buoy_tests.erl

@@ -36,6 +36,14 @@ buoy_test_() ->
         fun telemetry_disabled_subtest/0
     ]}.
 
+buoy_max_requests_test_() ->
+    {setup,
+        fun () -> setup([{max_requests, 2}, {pool_size, 1}]) end,
+        fun (_) -> cleanup() end,
+    [
+        fun max_requests_subtest/0
+    ]}.
+
 buoy_socket_test_() ->
     case list_to_integer(erlang:system_info(otp_release)) >= 28 of
         true ->
@@ -68,6 +76,17 @@ get_subtest() ->
     {ok, ?RESP_2} = buoy:get(?URL(?URL_2), #{}),
     {ok, ?RESP_4} = buoy:get(?URL(?URL_4), #{}).
 
+max_requests_subtest() ->
+    1 = buoy_http_server:connection_count(),
+    lists:foreach(fun (_) ->
+        {ok, ?RESP_1} = buoy:get(?URL(?URL_1), #{}),
+        {ok, ?RESP_1} = buoy:get(?URL(?URL_1), #{}),
+        %% recycling closes the socket after the 2nd response and
+        %% reconnects asynchronously; wait it out before the next pair
+        timer:sleep(100)
+    end, lists:seq(1, 3)),
+    true = buoy_http_server:connection_count() >= 3.
+
 pool_subtest() ->
     {error, pool_already_started} = buoy_pool:start(?URL(?URL_1)),
     ok = buoy_pool:stop(?URL(?URL_1)),